diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamSource.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamSource.java
deleted file mode 100644
index e292ab4ac74a8..0000000000000
--- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamSource.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kafka.streaming.kstream.internals;
-
-import org.apache.kafka.streaming.processor.internals.SourceNode;
-import org.apache.kafka.streaming.processor.TopologyBuilder;
-
-public class KStreamSource<K, V> extends KStreamImpl<K, V> {
-
-    private SourceNode<K, V> source;
-
-    public KStreamSource(TopologyBuilder topology, SourceNode<K, V> source) {
-        super(topology, source.name());
-        this.source = source;
-    }
-
-    public SourceNode<K, V> source() {
-        return source;
-    }
-}
diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorContextImpl.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorContextImpl.java
index 03df3e3cf5a57..a4954b408f806 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorContextImpl.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorContextImpl.java
@@ -217,11 +217,7 @@ public <K, V> void forward(K key, V value) {
     @Override
     @SuppressWarnings("unchecked")
     public <K, V> void forward(K key, V value, int childIndex) {
-<<<<<<< HEAD
         ProcessorNode childNode = (ProcessorNode<K, V>) task.node().children().get(childIndex);
-=======
-        ProcessorNode childNode = (ProcessorNode<K, V, ?, ?>) task.node().children().get(childIndex);
->>>>>>> kstream refactored
         task.node(childNode);
         childNode.process(key, value);
     }
diff --git a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamBranchTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamBranchTest.java
index 217a65f79afde..89c0278f79e1a 100644
--- a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamBranchTest.java
+++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamBranchTest.java
@@ -22,9 +22,8 @@
 import org.apache.kafka.streaming.kstream.KStream;
 import org.apache.kafka.streaming.kstream.KStreamBuilder;
 import org.apache.kafka.streaming.kstream.Predicate;
-import org.apache.kafka.streaming.kstream.internals.KStreamSource;
-import org.apache.kafka.test.MockKStreamBuilder;
-import org.apache.kafka.test.MockProcessor;
+import org.apache.kafka.test.KStreamTestDriver;
+import org.apache.kafka.test.MockProcessorDef;
 import org.junit.Test;
 
 import java.lang.reflect.Array;
@@ -33,15 +32,15 @@
 
 public class KStreamBranchTest {
 
-    private String topic1 = "topic";
+    private String topicName = "topic";
 
-    private KStreamBuilder topology = new MockKStreamBuilder();
     private IntegerDeserializer keyDeserializer = new IntegerDeserializer();
     private StringDeserializer valDeserializer = new StringDeserializer();
 
     @SuppressWarnings("unchecked")
     @Test
     public void testKStreamBranch() {
+        KStreamBuilder builder = new KStreamBuilder();
 
         Predicate<Integer, String> isEven = new Predicate<Integer, String>() {
             @Override
@@ -66,21 +65,22 @@ public boolean apply(Integer key, String value) {
 
         KStream<Integer, String> stream;
         KStream<Integer, String>[] branches;
-        MockProcessor<Integer, String>[] processors;
+        MockProcessorDef<Integer, String>[] processors;
 
-        stream = topology.<Integer, String>from(keyDeserializer, valDeserializer, topic1);
+        stream = builder.from(keyDeserializer, valDeserializer, topicName);
         branches = stream.branch(isEven, isMultipleOfThree, isOdd);
 
         assertEquals(3, branches.length);
 
-        processors = (MockProcessor<Integer, String>[]) Array.newInstance(MockProcessor.class, branches.length);
+        processors = (MockProcessorDef<Integer, String>[]) Array.newInstance(MockProcessorDef.class, branches.length);
         for (int i = 0; i < branches.length; i++) {
-            processors[i] = new MockProcessor<>();
+            processors[i] = new MockProcessorDef<>();
             branches[i].process(processors[i]);
         }
 
+        KStreamTestDriver driver = new KStreamTestDriver(builder);
         for (int i = 0; i < expectedKeys.length; i++) {
-            ((KStreamSource<Integer, String>) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]);
+            driver.process(topicName, expectedKeys[i], "V" + expectedKeys[i]);
         }
 
         assertEquals(3, processors[0].processed.size());
diff --git a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFilterTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFilterTest.java
index 653550e22085d..791005432b926 100644
--- a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFilterTest.java
+++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFilterTest.java
@@ -22,9 +22,8 @@
 import org.apache.kafka.streaming.kstream.KStream;
 import org.apache.kafka.streaming.kstream.KStreamBuilder;
 import org.apache.kafka.streaming.kstream.Predicate;
-import org.apache.kafka.streaming.kstream.internals.KStreamSource;
-import org.apache.kafka.test.MockKStreamBuilder;
-import org.apache.kafka.test.MockProcessor;
+import org.apache.kafka.test.KStreamTestDriver;
+import org.apache.kafka.test.MockProcessorDef;
 
 import org.junit.Test;
 
@@ -34,7 +33,6 @@ public class KStreamFilterTest {
 
     private String topicName = "topic";
 
-    private KStreamBuilder topology = new MockKStreamBuilder();
     private IntegerDeserializer keyDeserializer = new IntegerDeserializer();
     private StringDeserializer valDeserializer = new StringDeserializer();
 
@@ -47,17 +45,19 @@ public boolean apply(Integer key, String value) {
 
     @Test
     public void testFilter() {
+        KStreamBuilder builder = new KStreamBuilder();
         final int[] expectedKeys = new int[]{1, 2, 3, 4, 5, 6, 7};
 
         KStream<Integer, String> stream;
-        MockProcessor<Integer, String> processor;
+        MockProcessorDef<Integer, String> processor;
 
-        processor = new MockProcessor<>();
-        stream = topology.<Integer, String>from(keyDeserializer, valDeserializer, topicName);
+        processor = new MockProcessorDef<>();
+        stream = builder.from(keyDeserializer, valDeserializer, topicName);
         stream.filter(isMultipleOfThree).process(processor);
 
+        KStreamTestDriver driver = new KStreamTestDriver(builder);
         for (int i = 0; i < expectedKeys.length; i++) {
-            ((KStreamSource<Integer, String>) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]);
+            driver.process(topicName, expectedKeys[i], "V" + expectedKeys[i]);
         }
 
         assertEquals(2, processor.processed.size());
@@ -65,17 +65,19 @@ public void testFilter() {
 
     @Test
     public void testFilterOut() {
+        KStreamBuilder builder = new KStreamBuilder();
         final int[] expectedKeys = new int[]{1, 2, 3, 4, 5, 6, 7};
 
         KStream<Integer, String> stream;
-        MockProcessor<Integer, String> processor;
+        MockProcessorDef<Integer, String> processor;
 
-        processor = new MockProcessor<>();
-        stream = topology.<Integer, String>from(keyDeserializer, valDeserializer, topicName);
+        processor = new MockProcessorDef<>();
+        stream = builder.from(keyDeserializer, valDeserializer, topicName);
         stream.filterOut(isMultipleOfThree).process(processor);
 
+        KStreamTestDriver driver = new KStreamTestDriver(builder);
         for (int i = 0; i < expectedKeys.length; i++) {
-            ((KStreamSource<Integer, String>) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]);
+            driver.process(topicName, expectedKeys[i], "V" + expectedKeys[i]);
         }
 
         assertEquals(5, processor.processed.size());
diff --git a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapTest.java
index 25abac1c13f8c..a700d8faa8168 100644
--- a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapTest.java
+++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapTest.java
@@ -22,10 +22,9 @@
 import org.apache.kafka.streaming.kstream.KStream;
 import org.apache.kafka.streaming.kstream.KStreamBuilder;
 import org.apache.kafka.streaming.kstream.KeyValue;
-import org.apache.kafka.streaming.kstream.KeyValueMapper;
-import org.apache.kafka.streaming.kstream.internals.KStreamSource;
-import org.apache.kafka.test.MockKStreamBuilder;
-import org.apache.kafka.test.MockProcessor;
+import org.apache.kafka.streaming.kstream.KeyValueFlatMap;
+import org.apache.kafka.test.KStreamTestDriver;
+import org.apache.kafka.test.MockProcessorDef;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
@@ -36,36 +35,37 @@ public class KStreamFlatMapTest {
 
     private String topicName = "topic";
 
-    private KStreamBuilder topology = new MockKStreamBuilder();
     private IntegerDeserializer keyDeserializer = new IntegerDeserializer();
     private StringDeserializer valDeserializer = new StringDeserializer();
 
     @Test
     public void testFlatMap() {
+        KStreamBuilder builder = new KStreamBuilder();
 
-        KeyValueMapper<Integer, String, String, Iterable<String>> mapper =
-            new KeyValueMapper<Integer, String, String, Iterable<String>>() {
+        KeyValueFlatMap<Integer, String, String, String> mapper =
+            new KeyValueFlatMap<Integer, String, String, String>() {
                 @Override
-                public KeyValue<String, Iterable<String>> apply(Integer key, String value) {
-                    ArrayList<String> result = new ArrayList<String>();
+                public Iterable<KeyValue<String, String>> apply(Integer key, String value) {
+                    ArrayList<KeyValue<String, String>> result = new ArrayList<>();
                     for (int i = 0; i < key; i++) {
-                        result.add(value);
+                        result.add(KeyValue.pair(Integer.toString(key), value));
                     }
-                    return KeyValue.pair(Integer.toString(key * 10), (Iterable<String>) result);
+                    return result;
                 }
             };
 
         final int[] expectedKeys = new int[]{0, 1, 2, 3};
 
         KStream<Integer, String> stream;
-        MockProcessor<String, String> processor;
+        MockProcessorDef<String, String> processor;
 
-        processor = new MockProcessor<>();
-        stream = topology.<Integer, String>from(keyDeserializer, valDeserializer, topicName);
+        processor = new MockProcessorDef<>();
+        stream = builder.from(keyDeserializer, valDeserializer, topicName);
         stream.flatMap(mapper).process(processor);
 
+        KStreamTestDriver driver = new KStreamTestDriver(builder);
         for (int i = 0; i < expectedKeys.length; i++) {
-            ((KStreamSource<Integer, String>) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]);
+            driver.process(topicName, expectedKeys[i], "V" + expectedKeys[i]);
         }
 
         assertEquals(6, processor.processed.size());
diff --git a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValuesTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValuesTest.java
index 8f37c6fc4674f..9678746c7fdf3 100644
--- a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValuesTest.java
+++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValuesTest.java
@@ -22,9 +22,8 @@
 import org.apache.kafka.streaming.kstream.KStream;
 import org.apache.kafka.streaming.kstream.KStreamBuilder;
 import org.apache.kafka.streaming.kstream.ValueMapper;
-import org.apache.kafka.streaming.kstream.internals.KStreamSource;
-import org.apache.kafka.test.MockKStreamBuilder;
-import org.apache.kafka.test.MockProcessor;
+import org.apache.kafka.test.KStreamTestDriver;
+import org.apache.kafka.test.MockProcessorDef;
 import org.junit.Test;
 
 import java.util.ArrayList;
@@ -35,12 +34,12 @@ public class KStreamFlatMapValuesTest {
 
     private String topicName = "topic";
 
-    private KStreamBuilder topology = new MockKStreamBuilder();
     private IntegerDeserializer keyDeserializer = new IntegerDeserializer();
     private StringDeserializer valDeserializer = new StringDeserializer();
 
     @Test
     public void testFlatMapValues() {
+        KStreamBuilder builder = new KStreamBuilder();
 
         ValueMapper<String, Iterable<String>> mapper =
             new ValueMapper<String, Iterable<String>>() {
@@ -56,14 +55,15 @@ public Iterable<String> apply(String value) {
         final int[] expectedKeys = new int[]{0, 1, 2, 3};
 
         KStream<Integer, String> stream;
-        MockProcessor<Integer, String> processor;
+        MockProcessorDef<Integer, String> processor;
 
-        processor = new MockProcessor<>();
-        stream = topology.<Integer, String>from(keyDeserializer, valDeserializer, topicName);
+        processor = new MockProcessorDef<>();
+        stream = builder.from(keyDeserializer, valDeserializer, topicName);
         stream.flatMapValues(mapper).process(processor);
 
+        KStreamTestDriver driver = new KStreamTestDriver(builder);
         for (int i = 0; i < expectedKeys.length; i++) {
-            ((KStreamSource<Integer, String>) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]);
+            driver.process(topicName, expectedKeys[i], "V" + expectedKeys[i]);
         }
 
         assertEquals(8, processor.processed.size());
diff --git a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamJoinTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamJoinTest.java
index 472dfd44eb7f0..5752c353ea505 100644
--- a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamJoinTest.java
+++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamJoinTest.java
@@ -27,11 +27,9 @@
 import org.apache.kafka.streaming.kstream.KeyValueMapper;
 import org.apache.kafka.streaming.kstream.ValueJoiner;
 import org.apache.kafka.streaming.kstream.ValueMapper;
-import org.apache.kafka.streaming.kstream.internals.KStreamSource;
-import org.apache.kafka.test.MockKStreamBuilder;
-import org.apache.kafka.test.MockProcessor;
-import org.apache.kafka.test.MockProcessorContext;
-import org.apache.kafka.test.UnlimitedWindow;
+import org.apache.kafka.test.KStreamTestDriver;
+import org.apache.kafka.test.MockProcessorDef;
+import org.apache.kafka.test.UnlimitedWindowDef;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
@@ -41,7 +39,6 @@ public class KStreamJoinTest {
     private String topic1 = "topic1";
     private String topic2 = "topic2";
 
-    private KStreamBuilder topology = new MockKStreamBuilder();
     private IntegerDeserializer keyDeserializer = new IntegerDeserializer();
     private StringDeserializer valDeserializer = new StringDeserializer();
 
@@ -85,6 +82,7 @@ public KeyValue<Integer, Iterable<String>> apply(Integer key, String value) {
 
     @Test
     public void testJoin() {
+        KStreamBuilder builder = new KStreamBuilder();
 
         final int[] expectedKeys = new int[]{0, 1, 2, 3};
 
@@ -92,25 +90,24 @@ public void testJoin() {
         KStream<Integer, String> stream2;
         KStreamWindowed<Integer, String> windowed1;
         KStreamWindowed<Integer, String> windowed2;
-        MockProcessor<Integer, String> processor;
+        MockProcessorDef<Integer, String> processor;
         String[] expected;
 
-        processor = new MockProcessor<>();
-        stream1 = topology.<Integer, String>from(keyDeserializer, valDeserializer, topic1);
-        stream2 = topology.<Integer, String>from(keyDeserializer, valDeserializer, topic2);
-        windowed1 = stream1.with(new UnlimitedWindow<Integer, String>());
-        windowed2 = stream2.with(new UnlimitedWindow<Integer, String>());
+        processor = new MockProcessorDef<>();
+        stream1 = builder.from(keyDeserializer, valDeserializer, topic1);
+        stream2 = builder.from(keyDeserializer, valDeserializer, topic2);
+        windowed1 = stream1.with(new UnlimitedWindowDef<Integer, String>("window1"));
+        windowed2 = stream2.with(new UnlimitedWindowDef<Integer, String>("window2"));
 
         windowed1.join(windowed2, joiner).process(processor);
 
-        MockProcessorContext context = new MockProcessorContext(null, null);
-        topology.init(context);
-        context.setTime(0L);
+        KStreamTestDriver driver = new KStreamTestDriver(builder);
+        driver.setTime(0L);
 
         // push two items to the main stream. the other stream's window is empty
 
         for (int i = 0; i < 2; i++) {
-            ((KStreamSource<Integer, String>) stream1).source().process(expectedKeys[i], "X" + expectedKeys[i]);
+            driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
 
         assertEquals(0, processor.processed.size());
@@ -118,7 +115,7 @@ public void testJoin() {
         // push two items to the other stream. the main stream's window has two items
 
         for (int i = 0; i < 2; i++) {
-            ((KStreamSource<Integer, String>) stream2).source().process(expectedKeys[i], "Y" + expectedKeys[i]);
+            driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
 
         assertEquals(2, processor.processed.size());
@@ -134,7 +131,7 @@ public void testJoin() {
         // push all items to the main stream. this should produce two items.
 
         for (int i = 0; i < expectedKeys.length; i++) {
-            ((KStreamSource<Integer, String>) stream1).source().process(expectedKeys[i], "X" + expectedKeys[i]);
+            driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
 
         assertEquals(2, processor.processed.size());
@@ -151,7 +148,7 @@ public void testJoin() {
 
         // push all items to the other stream. this should produce 6 items
         for (int i = 0; i < expectedKeys.length; i++) {
-            ((KStreamSource<Integer, String>) stream2).source().process(expectedKeys[i], "Y" + expectedKeys[i]);
+            driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
 
         assertEquals(6, processor.processed.size());
@@ -165,6 +162,7 @@ public void testJoin() {
 
     @Test
     public void testJoinPrior() {
+        KStreamBuilder builder = new KStreamBuilder();
 
         final int[] expectedKeys = new int[]{0, 1, 2, 3};
 
@@ -172,26 +170,24 @@ public void testJoinPrior() {
         KStream<Integer, String> stream2;
         KStreamWindowed<Integer, String> windowed1;
         KStreamWindowed<Integer, String> windowed2;
-        MockProcessor<Integer, String> processor;
+        MockProcessorDef<Integer, String> processor;
         String[] expected;
 
-        processor = new MockProcessor<>();
-        stream1 = topology.<Integer, String>from(keyDeserializer, valDeserializer, topic1);
-        stream2 = topology.<Integer, String>from(keyDeserializer, valDeserializer, topic2);
-        windowed1 = stream1.with(new UnlimitedWindow<Integer, String>());
-        windowed2 = stream2.with(new UnlimitedWindow<Integer, String>());
+        processor = new MockProcessorDef<>();
+        stream1 = builder.from(keyDeserializer, valDeserializer, topic1);
+        stream2 = builder.from(keyDeserializer, valDeserializer, topic2);
+        windowed1 = stream1.with(new UnlimitedWindowDef<Integer, String>("window1"));
+        windowed2 = stream2.with(new UnlimitedWindowDef<Integer, String>("window2"));
 
         windowed1.joinPrior(windowed2, joiner).process(processor);
 
-        MockProcessorContext context = new MockProcessorContext(null, null);
-        topology.init(context);
+        KStreamTestDriver driver = new KStreamTestDriver(builder);
 
         // push two items to the main stream. the other stream's window is empty
 
         for (int i = 0; i < 2; i++) {
-            context.setTime(i);
-
-            ((KStreamSource<Integer, String>) stream1).source().process(expectedKeys[i], "X" + expectedKeys[i]);
+            driver.setTime(i);
+            driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
 
         assertEquals(0, processor.processed.size());
@@ -200,9 +196,8 @@ public void testJoinPrior() {
         // no corresponding item in the main window has a newer timestamp
 
         for (int i = 0; i < 2; i++) {
-            context.setTime(i + 1);
-
-            ((KStreamSource<Integer, String>) stream2).source().process(expectedKeys[i], "Y" + expectedKeys[i]);
+            driver.setTime(i + 1);
+            driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
 
         assertEquals(0, processor.processed.size());
@@ -212,9 +207,8 @@ public void testJoinPrior() {
         // push all items with newer timestamps to the main stream. this should produce two items.
 
         for (int i = 0; i < expectedKeys.length; i++) {
-            context.setTime(i + 2);
-
-            ((KStreamSource<Integer, String>) stream1).source().process(expectedKeys[i], "X" + expectedKeys[i]);
+            driver.setTime(i + 2);
+            driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
 
         assertEquals(2, processor.processed.size());
@@ -231,9 +225,8 @@ public void testJoinPrior() {
 
         // push all items with older timestamps to the other stream. this should produce six items
         for (int i = 0; i < expectedKeys.length; i++) {
-            context.setTime(i);
-
-            ((KStreamSource<Integer, String>) stream2).source().process(expectedKeys[i], "Y" + expectedKeys[i]);
+            driver.setTime(i);
+            driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
 
         assertEquals(6, processor.processed.size());
diff --git a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapTest.java
index 08711cdd8b178..977a4715ac54d 100644
--- a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapTest.java
+++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapTest.java
@@ -23,9 +23,8 @@
 import org.apache.kafka.streaming.kstream.KStreamBuilder;
 import org.apache.kafka.streaming.kstream.KeyValue;
 import org.apache.kafka.streaming.kstream.KeyValueMapper;
-import org.apache.kafka.streaming.kstream.internals.KStreamSource;
-import org.apache.kafka.test.MockKStreamBuilder;
-import org.apache.kafka.test.MockProcessor;
+import org.apache.kafka.test.KStreamTestDriver;
+import org.apache.kafka.test.MockProcessorDef;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
@@ -34,12 +33,12 @@ public class KStreamMapTest {
 
     private String topicName = "topic";
 
-    private KStreamBuilder topology = new MockKStreamBuilder();
     private IntegerDeserializer keyDeserializer = new IntegerDeserializer();
     private StringDeserializer valDeserializer = new StringDeserializer();
 
     @Test
     public void testMap() {
+        KStreamBuilder builder = new KStreamBuilder();
 
         KeyValueMapper<Integer, String, String, Integer> mapper =
             new KeyValueMapper<Integer, String, String, Integer>() {
@@ -52,14 +51,15 @@ public KeyValue<String, Integer> apply(Integer key, String value) {
         final int[] expectedKeys = new int[]{0, 1, 2, 3};
 
         KStream<Integer, String> stream;
-        MockProcessor<String, Integer> processor;
+        MockProcessorDef<String, Integer> processor;
 
-        processor = new MockProcessor<>();
-        stream = topology.<Integer, String>from(keyDeserializer, valDeserializer, topicName);
+        processor = new MockProcessorDef<>();
+        stream = builder.from(keyDeserializer, valDeserializer, topicName);
         stream.map(mapper).process(processor);
 
+        KStreamTestDriver driver = new KStreamTestDriver(builder);
         for (int i = 0; i < expectedKeys.length; i++) {
-            ((KStreamSource<Integer, String>) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]);
+            driver.process(topicName, expectedKeys[i], "V" + expectedKeys[i]);
         }
 
         assertEquals(4, processor.processed.size());
diff --git a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValuesTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValuesTest.java
index 45a793fc7ab9f..f48a03244cf17 100644
--- a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValuesTest.java
+++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValuesTest.java
@@ -22,9 +22,8 @@
 import org.apache.kafka.streaming.kstream.KStream;
 import org.apache.kafka.streaming.kstream.KStreamBuilder;
 import org.apache.kafka.streaming.kstream.ValueMapper;
-import org.apache.kafka.streaming.kstream.internals.KStreamSource;
-import org.apache.kafka.test.MockKStreamBuilder;
-import org.apache.kafka.test.MockProcessor;
+import org.apache.kafka.test.KStreamTestDriver;
+import org.apache.kafka.test.MockProcessorDef;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
@@ -33,12 +32,12 @@ public class KStreamMapValuesTest {
 
     private String topicName = "topic";
 
-    private KStreamBuilder topology = new MockKStreamBuilder();
     private IntegerDeserializer keyDeserializer = new IntegerDeserializer();
     private StringDeserializer valDeserializer = new StringDeserializer();
 
     @Test
     public void testFlatMapValues() {
+        KStreamBuilder builder = new KStreamBuilder();
 
         ValueMapper<String, Integer> mapper =
             new ValueMapper<String, Integer>() {
@@ -51,12 +50,13 @@ public Integer apply(String value) {
         final int[] expectedKeys = new int[]{1, 10, 100, 1000};
 
         KStream<Integer, String> stream;
-        MockProcessor<Integer, Integer> processor = new MockProcessor<>();
-        stream = topology.<Integer, String>from(keyDeserializer, valDeserializer, topicName);
+        MockProcessorDef<Integer, Integer> processor = new MockProcessorDef<>();
+        stream = builder.from(keyDeserializer, valDeserializer, topicName);
         stream.mapValues(mapper).process(processor);
 
+        KStreamTestDriver driver = new KStreamTestDriver(builder);
         for (int i = 0; i < expectedKeys.length; i++) {
-            ((KStreamSource<Integer, String>) stream).source().process(expectedKeys[i], Integer.toString(expectedKeys[i]));
+            driver.process(topicName, expectedKeys[i], Integer.toString(expectedKeys[i]));
         }
 
         assertEquals(4, processor.processed.size());
diff --git a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamSourceTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamSourceTest.java
deleted file mode 100644
index ac295fd7419da..0000000000000
--- a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamSourceTest.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kafka.streaming.kstream.internals;
-
-import org.apache.kafka.common.serialization.StringDeserializer;
-import org.apache.kafka.streaming.kstream.KStream;
-import org.apache.kafka.streaming.kstream.KStreamBuilder;
-import org.apache.kafka.streaming.kstream.internals.KStreamSource;
-import org.apache.kafka.test.MockKStreamBuilder;
-import org.apache.kafka.test.MockProcessor;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
-public class KStreamSourceTest {
-
-    private String topicName = "topic";
-
-    private KStreamBuilder topology = new MockKStreamBuilder();
-    private StringDeserializer keyDeserializer = new StringDeserializer();
-    private StringDeserializer valDeserializer = new StringDeserializer();
-
-    @Test
-    public void testKStreamSource() {
-
-        MockProcessor<String, String> processor = new MockProcessor<>();
-
-        KStream<String, String> stream = topology.<String, String>from(keyDeserializer, valDeserializer, topicName);
-        stream.process(processor);
-
-        final String[] expectedKeys = new String[]{"k1", "k2", "k3"};
-        final String[] expectedValues = new String[]{"v1", "v2", "v3"};
-
-        for (int i = 0; i < expectedKeys.length; i++) {
-            ((KStreamSource<String, String>) stream).source().process(expectedKeys[i], expectedValues[i]);
-        }
-
-        assertEquals(3, processor.processed.size());
-
-        for (int i = 0; i < expectedKeys.length; i++) {
-            assertEquals(expectedKeys[i] + ":" + expectedValues[i], processor.processed.get(i));
-        }
-    }
-}
diff --git a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedTest.java
index 20cc445e814b4..dfa8f3fba3c50 100644
--- a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedTest.java
+++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedTest.java
@@ -21,11 +21,10 @@
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.streaming.kstream.KStream;
 import org.apache.kafka.streaming.kstream.KStreamBuilder;
+import org.apache.kafka.streaming.kstream.Window;
 import org.apache.kafka.streaming.kstream.WindowDef;
-import org.apache.kafka.streaming.kstream.internals.KStreamSource;
-import org.apache.kafka.test.MockKStreamBuilder;
-import org.apache.kafka.test.MockProcessorContext;
-import org.apache.kafka.test.UnlimitedWindow;
+import org.apache.kafka.test.KStreamTestDriver;
+import org.apache.kafka.test.UnlimitedWindowDef;
 import org.junit.Test;
 
 import java.util.Iterator;
@@ -35,31 +34,32 @@
 public class KStreamWindowedTest {
 
     private String topicName = "topic";
+    private String windowName = "MyWindow";
 
-    private KStreamBuilder topology = new MockKStreamBuilder();
     private IntegerDeserializer keyDeserializer = new IntegerDeserializer();
     private StringDeserializer valDeserializer = new StringDeserializer();
 
     @Test
     public void testWindowedStream() {
+        KStreamBuilder builder = new KStreamBuilder();
 
         final int[] expectedKeys = new int[]{0, 1, 2, 3};
 
         KStream<Integer, String> stream;
-        WindowDef<Integer, String> window;
+        WindowDef<Integer, String> windowDef;
 
-        window = new UnlimitedWindow<>();
-        stream = topology.<Integer, String>from(keyDeserializer, valDeserializer, topicName);
-        stream.with(window);
+        windowDef = new UnlimitedWindowDef<>(windowName);
+        stream = builder.from(keyDeserializer, valDeserializer, topicName);
+        stream.with(windowDef);
 
-        MockProcessorContext context = new MockProcessorContext(null, null);
-        topology.init(context);
-        context.setTime(0L);
+        KStreamTestDriver driver = new KStreamTestDriver(builder);
+        Window<Integer, String> window = (Window<Integer, String>) driver.getStateStore(windowName);
+        driver.setTime(0L);
 
         // two items in the window
 
         for (int i = 0; i < 2; i++) {
-            ((KStreamSource<Integer, String>) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]);
+            driver.process(topicName, expectedKeys[i], "V" + expectedKeys[i]);
         }
 
         assertEquals(1, countItem(window.find(0, 0L)));
@@ -70,7 +70,7 @@ public void testWindowedStream() {
         // previous two items + all items, thus two are duplicates, in the window
 
         for (int i = 0; i < expectedKeys.length; i++) {
-            ((KStreamSource<Integer, String>) stream).source().process(expectedKeys[i], "Y" + expectedKeys[i]);
+            driver.process(topicName, expectedKeys[i], "Y" + expectedKeys[i]);
         }
 
         assertEquals(2, countItem(window.find(0, 0L)));
diff --git a/stream/src/test/java/org/apache/kafka/test/KStreamTestDriver.java b/stream/src/test/java/org/apache/kafka/test/KStreamTestDriver.java
new file mode 100644
index 0000000000000..e026fcc67bd50
--- /dev/null
+++ b/stream/src/test/java/org/apache/kafka/test/KStreamTestDriver.java
@@ -0,0 +1,55 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.test;
+
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.streaming.kstream.KStreamBuilder;
+import org.apache.kafka.streaming.processor.StateStore;
+import org.apache.kafka.streaming.processor.internals.ProcessorTopology;
+
+public class KStreamTestDriver {
+
+    private final ProcessorTopology topology;
+    private final MockProcessorContext context;
+
+    public KStreamTestDriver(KStreamBuilder builder) {
+        this(builder, null, null);
+    }
+
+    public KStreamTestDriver(KStreamBuilder builder, Serializer<?> serializer, Deserializer<?> deserializer) {
+        this.topology = builder.build();
+        this.context = new MockProcessorContext(serializer, deserializer);
+
+        this.topology.init(context);
+    }
+
+    public void process(String topicName, Object key, Object value) {
+        context.node(topology.source(topicName));
+        context.forward(key, value);
+    }
+
+    public void setTime(long timestamp) {
+        context.setTime(timestamp);
+    }
+
+    public StateStore getStateStore(String name) {
+        return context.getStateStore(name);
+    }
+
+}
diff --git a/stream/src/test/java/org/apache/kafka/test/MockKStreamBuilder.java b/stream/src/test/java/org/apache/kafka/test/MockKStreamBuilder.java
deleted file mode 100644
index aa3de2b7a050b..0000000000000
--- a/stream/src/test/java/org/apache/kafka/test/MockKStreamBuilder.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kafka.test;
-
-import org.apache.kafka.streaming.kstream.KStreamBuilder;
-
-public class MockKStreamBuilder extends KStreamBuilder {
-    public MockKStreamBuilder() {
-        super();
-    }
-}
diff --git a/stream/src/test/java/org/apache/kafka/test/MockProcessorContext.java b/stream/src/test/java/org/apache/kafka/test/MockProcessorContext.java
index 7d930b10ac07f..b8f65252bea39 100644
--- a/stream/src/test/java/org/apache/kafka/test/MockProcessorContext.java
+++ b/stream/src/test/java/org/apache/kafka/test/MockProcessorContext.java
@@ -24,13 +24,20 @@
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.streaming.processor.internals.ProcessorNode;
 
 import java.io.File;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 public class MockProcessorContext implements ProcessorContext {
 
-    Serializer serializer;
-    Deserializer deserializer;
+    private Serializer serializer;
+    private Deserializer deserializer;
+    private ProcessorNode node;
+
+    private Map<String, StateStore> storeMap = new HashMap<>();
 
     long timestamp = -1L;
 
@@ -44,7 +51,7 @@ public void setTime(long timestamp) {
     }
 
     @Override
-    public boolean joinable(ProcessorContext other) {
+    public boolean joinable() {
         // TODO
         return true;
     }
@@ -86,7 +93,13 @@ public Metrics metrics() {
 
     @Override
     public void register(StateStore store, RestoreFunc func) {
-        throw new UnsupportedOperationException("restore() not supported.");
+        if (func != null) new UnsupportedOperationException("RestoreFunc not supported.");
+        storeMap.put(store.name(), store);
+    }
+
+    @Override
+    public StateStore getStateStore(String name) {
+        return storeMap.get(name);
     }
 
     @Override
@@ -95,8 +108,20 @@ public void schedule(Processor processor, long interval) {
     }
 
     @Override
+    @SuppressWarnings("unchecked")
     public <K, V> void forward(K key, V value) {
-        throw new UnsupportedOperationException("forward() not supported");
+        for (ProcessorNode childNode : (List<ProcessorNode<K, V>>) node().children()) {
+            node(childNode);
+            childNode.process(key, value);
+        }
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public <K, V> void forward(K key, V value, int childIndex) {
+        ProcessorNode childNode = (ProcessorNode<K, V>) node().children().get(childIndex);
+        node(childNode);
+        childNode.process(key, value);
     }
 
     @Override
@@ -124,4 +149,11 @@ public long timestamp() {
         return this.timestamp;
     }
 
+    public void node(ProcessorNode node) {
+        this.node = node;
+    }
+
+    public ProcessorNode node() {
+        return this.node;
+    }
 }
diff --git a/stream/src/test/java/org/apache/kafka/test/MockProcessor.java b/stream/src/test/java/org/apache/kafka/test/MockProcessorDef.java
similarity index 58%
rename from stream/src/test/java/org/apache/kafka/test/MockProcessor.java
rename to stream/src/test/java/org/apache/kafka/test/MockProcessorDef.java
index 81dbf49a5fe87..472f91918a48c 100644
--- a/stream/src/test/java/org/apache/kafka/test/MockProcessor.java
+++ b/stream/src/test/java/org/apache/kafka/test/MockProcessorDef.java
@@ -18,24 +18,41 @@
 package org.apache.kafka.test;
 
 import org.apache.kafka.streaming.processor.Processor;
+import org.apache.kafka.streaming.processor.ProcessorContext;
+import org.apache.kafka.streaming.processor.ProcessorDef;
 
 import java.util.ArrayList;
 
-public class MockProcessor<K1, V1> extends Processor<K1, V1, Object, Object> {
+public class MockProcessorDef<K, V> implements ProcessorDef {
+
     public final ArrayList<String> processed = new ArrayList<>();
     public final ArrayList<Long> punctuated = new ArrayList<>();
 
-    public MockProcessor() {
-        super("MOCK");
+    public Processor instance() {
+        return new MockProcessor();
     }
 
-    @Override
-    public void process(K1 key, V1 value) {
-        processed.add(key + ":" + value);
-    }
+    public class MockProcessor implements Processor<K, V> {
+
+        @Override
+        public void init(ProcessorContext context) {
+            // do nothing
+        }
+
+        @Override
+        public void process(K key, V value) {
+            processed.add(key + ":" + value);
+        }
+
+        @Override
+        public void punctuate(long streamTime) {
+            punctuated.add(streamTime);
+        }
+
+        @Override
+        public void close() {
+            // do nothing
+        }
 
-    @Override
-    public void punctuate(long streamTime) {
-        punctuated.add(streamTime);
     }
 }
diff --git a/stream/src/test/java/org/apache/kafka/test/UnlimitedWindow.java b/stream/src/test/java/org/apache/kafka/test/UnlimitedWindow.java
deleted file mode 100644
index 2dc4c6cbed287..0000000000000
--- a/stream/src/test/java/org/apache/kafka/test/UnlimitedWindow.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kafka.test;
-
-import org.apache.kafka.streaming.processor.ProcessorContext;
-import org.apache.kafka.streaming.kstream.KeyValue;
-import org.apache.kafka.streaming.kstream.WindowDef;
-import org.apache.kafka.streaming.kstream.internals.FilteredIterator;
-import org.apache.kafka.streaming.processor.internals.Stamped;
-
-import java.util.Iterator;
-import java.util.LinkedList;
-
-public class UnlimitedWindow<K, V> implements WindowDef<K, V> {
-
-    private LinkedList<Stamped<KeyValue<K, V>>> list = new LinkedList<>();
-
-    @Override
-    public void init(ProcessorContext context) {
-        // do nothing
-    }
-
-    @Override
-    public Iterator<V> find(final K key, long timestamp) {
-        return find(key, Long.MIN_VALUE, timestamp);
-    }
-
-    @Override
-    public Iterator<V> findAfter(final K key, long timestamp) {
-        return find(key, timestamp, Long.MAX_VALUE);
-    }
-
-    @Override
-    public Iterator<V> findBefore(final K key, long timestamp) {
-        return find(key, Long.MIN_VALUE, Long.MAX_VALUE);
-    }
-
-    private Iterator<V> find(final K key, final long startTime, final long endTime) {
-        return new FilteredIterator<V, Stamped<KeyValue<K, V>>>(list.iterator()) {
-            protected V filter(Stamped<KeyValue<K, V>> item) {
-                if (item.value.key.equals(key) && startTime <= item.timestamp && item.timestamp <= endTime)
-                    return item.value.value;
-                else
-                    return null;
-            }
-        };
-    }
-
-    @Override
-    public void put(K key, V value, long timestamp) {
-        list.add(new Stamped<>(KeyValue.pair(key, value), timestamp));
-    }
-
-    @Override
-    public String name() {
-        return null;
-    }
-
-    @Override
-    public void flush() {
-    }
-
-    @Override
-    public void close() {
-    }
-
-    @Override
-    public boolean persistent() {
-        return false;
-    }
-}
diff --git a/stream/src/test/java/org/apache/kafka/test/UnlimitedWindowDef.java b/stream/src/test/java/org/apache/kafka/test/UnlimitedWindowDef.java
new file mode 100644
index 0000000000000..c6d929725c65f
--- /dev/null
+++ b/stream/src/test/java/org/apache/kafka/test/UnlimitedWindowDef.java
@@ -0,0 +1,104 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.test;
+
+import org.apache.kafka.streaming.kstream.Window;
+import org.apache.kafka.streaming.processor.ProcessorContext;
+import org.apache.kafka.streaming.kstream.KeyValue;
+import org.apache.kafka.streaming.kstream.WindowDef;
+import org.apache.kafka.streaming.kstream.internals.FilteredIterator;
+import org.apache.kafka.streaming.processor.internals.Stamped;
+
+import java.util.Iterator;
+import java.util.LinkedList;
+
+public class UnlimitedWindowDef<K, V> implements WindowDef<K, V> {
+
+    private final String name;
+
+    public UnlimitedWindowDef(String name) {
+        this.name = name;
+    }
+
+    public String name() {
+        return name;
+    }
+
+    public Window<K, V> instance() {
+        return new UnlimitedWindow();
+    }
+
+    public class UnlimitedWindow implements Window<K, V> {
+
+        private LinkedList<Stamped<KeyValue<K, V>>> list = new LinkedList<>();
+
+        @Override
+        public void init (ProcessorContext context){
+            context.register(this, null);
+        }
+
+        @Override
+        public Iterator<V> find(final K key, long timestamp){
+            return find(key, Long.MIN_VALUE, timestamp);
+        }
+
+        @Override
+        public Iterator<V> findAfter(final K key, long timestamp){
+            return find(key, timestamp, Long.MAX_VALUE);
+        }
+
+        @Override
+        public Iterator<V> findBefore(final K key, long timestamp){
+            return find(key, Long.MIN_VALUE, Long.MAX_VALUE);
+        }
+
+        private Iterator<V> find(final K key, final long startTime, final long endTime){
+            return new FilteredIterator<V, Stamped<KeyValue<K, V>>>(list.iterator()) {
+                protected V filter(Stamped<KeyValue<K, V>> item) {
+                    if (item.value.key.equals(key) && startTime <= item.timestamp && item.timestamp <= endTime)
+                        return item.value.value;
+                    else
+                        return null;
+                }
+            };
+        }
+
+        @Override
+        public void put (K key, V value,long timestamp){
+            list.add(new Stamped<>(KeyValue.pair(key, value), timestamp));
+        }
+
+        @Override
+        public String name () {
+            return null;
+        }
+
+        @Override
+        public void flush () {
+        }
+
+        @Override
+        public void close () {
+        }
+
+        @Override
+        public boolean persistent () {
+            return false;
+        }
+    }
+}