@Override
        public void putAll(final List<KeyValue<K, V>> entries) {
            for (final KeyValue<K, V> entry : entries) {
                inner.put(entry.key, ValueAndTimestamp.make(entry.value, ConsumerRecord.NO_TIMESTAMP));
            }
        }
Close the driver, its topology, and all processors.
streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java