kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject [3/6] kafka git commit: KAFKA-3776: Unify store and downstream caching in streams
Date Fri, 16 Sep 2016 16:58:45 GMT
http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java
index 1a5de5f..37aac0c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java
@@ -26,9 +26,11 @@ import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.TimeWindows;
 import org.apache.kafka.streams.kstream.ValueJoiner;
 import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
 import org.apache.kafka.test.KStreamTestDriver;
 import org.apache.kafka.test.MockAggregator;
 import org.apache.kafka.test.MockInitializer;
+import org.apache.kafka.test.MockProcessorContext;
 import org.apache.kafka.test.MockProcessorSupplier;
 import org.apache.kafka.test.TestUtils;
 import org.junit.After;
@@ -81,40 +83,56 @@ public class KStreamWindowAggregateTest {
             MockProcessorSupplier<Windowed<String>, String> proc2 = new MockProcessorSupplier<>();
             table2.toStream().process(proc2);
 
-            KStreamTestDriver driver = new KStreamTestDriver(builder, baseDir);
+            driver = new KStreamTestDriver(builder, baseDir);
 
-            driver.setTime(0L);
+            setRecordContext(0, topic1);
             driver.process(topic1, "A", "1");
-            driver.setTime(1L);
+            driver.flushState();
+            setRecordContext(1, topic1);
             driver.process(topic1, "B", "2");
-            driver.setTime(2L);
+            driver.flushState();
+            setRecordContext(2, topic1);
             driver.process(topic1, "C", "3");
-            driver.setTime(3L);
+            driver.flushState();
+            setRecordContext(3, topic1);
             driver.process(topic1, "D", "4");
-            driver.setTime(4L);
+            driver.flushState();
+            setRecordContext(4, topic1);
             driver.process(topic1, "A", "1");
+            driver.flushState();
 
-            driver.setTime(5L);
+            setRecordContext(5, topic1);
             driver.process(topic1, "A", "1");
-            driver.setTime(6L);
+            driver.flushState();
+            setRecordContext(6, topic1);
             driver.process(topic1, "B", "2");
-            driver.setTime(7L);
+            driver.flushState();
+            setRecordContext(7, topic1);
             driver.process(topic1, "D", "4");
-            driver.setTime(8L);
+            driver.flushState();
+            setRecordContext(8, topic1);
             driver.process(topic1, "B", "2");
-            driver.setTime(9L);
+            driver.flushState();
+            setRecordContext(9, topic1);
             driver.process(topic1, "C", "3");
-
-            driver.setTime(10L);
+            driver.flushState();
+            setRecordContext(10, topic1);
             driver.process(topic1, "A", "1");
-            driver.setTime(11L);
+            driver.flushState();
+            setRecordContext(11, topic1);
             driver.process(topic1, "B", "2");
-            driver.setTime(12L);
+            driver.flushState();
+            setRecordContext(12, topic1);
+            driver.flushState();
             driver.process(topic1, "D", "4");
-            driver.setTime(13L);
+            driver.flushState();
+            setRecordContext(13, topic1);
             driver.process(topic1, "B", "2");
-            driver.setTime(14L);
+            driver.flushState();
+            setRecordContext(14, topic1);
             driver.process(topic1, "C", "3");
+            driver.flushState();
+
 
             assertEquals(Utils.mkList(
                     "[A@0]:0+1",
@@ -140,6 +158,10 @@ public class KStreamWindowAggregateTest {
         }
     }
 
+    private void setRecordContext(final long time, final String topic) {
+        ((MockProcessorContext) driver.context()).setRecordContext(new ProcessorRecordContext(time, 0, 0, topic));
+    }
+
     @Test
     public void testJoin() throws Exception {
         final File baseDir = Files.createTempDirectory("test").toFile();
@@ -180,18 +202,23 @@ public class KStreamWindowAggregateTest {
                 }
             }).toStream().process(proc3);
 
-            KStreamTestDriver driver = new KStreamTestDriver(builder, baseDir);
+            driver = new KStreamTestDriver(builder, baseDir);
 
-            driver.setTime(0L);
+            setRecordContext(0, topic1);
             driver.process(topic1, "A", "1");
-            driver.setTime(1L);
+            driver.flushState();
+            setRecordContext(1, topic1);
             driver.process(topic1, "B", "2");
-            driver.setTime(2L);
+            driver.flushState();
+            setRecordContext(2, topic1);
             driver.process(topic1, "C", "3");
-            driver.setTime(3L);
+            driver.flushState();
+            setRecordContext(3, topic1);
             driver.process(topic1, "D", "4");
-            driver.setTime(4L);
+            driver.flushState();
+            setRecordContext(4, topic1);
             driver.process(topic1, "A", "1");
+            driver.flushState();
 
             proc1.checkAndClearProcessResult(
                     "[A@0]:0+1",
@@ -209,16 +236,21 @@ public class KStreamWindowAggregateTest {
                     "[A@0]:null"
             );
 
-            driver.setTime(5L);
+            setRecordContext(5, topic1);
             driver.process(topic1, "A", "1");
-            driver.setTime(6L);
+            driver.flushState();
+            setRecordContext(6, topic1);
             driver.process(topic1, "B", "2");
-            driver.setTime(7L);
+            driver.flushState();
+            setRecordContext(7, topic1);
             driver.process(topic1, "D", "4");
-            driver.setTime(8L);
+            driver.flushState();
+            setRecordContext(8, topic1);
             driver.process(topic1, "B", "2");
-            driver.setTime(9L);
+            driver.flushState();
+            setRecordContext(9, topic1);
             driver.process(topic1, "C", "3");
+            driver.flushState();
 
             proc1.checkAndClearProcessResult(
                     "[A@0]:0+1+1+1", "[A@5]:0+1",
@@ -236,16 +268,21 @@ public class KStreamWindowAggregateTest {
                     "[C@0]:null", "[C@5]:null"
             );
 
-            driver.setTime(0L);
+            setRecordContext(0, topic1);
             driver.process(topic2, "A", "a");
-            driver.setTime(1L);
+            driver.flushState();
+            setRecordContext(1, topic1);
             driver.process(topic2, "B", "b");
-            driver.setTime(2L);
+            driver.flushState();
+            setRecordContext(2, topic1);
             driver.process(topic2, "C", "c");
-            driver.setTime(3L);
+            driver.flushState();
+            setRecordContext(3, topic1);
             driver.process(topic2, "D", "d");
-            driver.setTime(4L);
+            driver.flushState();
+            setRecordContext(4, topic1);
             driver.process(topic2, "A", "a");
+            driver.flushState();
 
             proc1.checkAndClearProcessResult();
             proc2.checkAndClearProcessResult(
@@ -262,17 +299,21 @@ public class KStreamWindowAggregateTest {
                     "[D@0]:0+4+4%0+d",
                     "[A@0]:0+1+1+1%0+a+a");
 
-            driver.setTime(5L);
+            setRecordContext(5, topic1);
             driver.process(topic2, "A", "a");
-            driver.setTime(6L);
+            driver.flushState();
+            setRecordContext(6, topic1);
             driver.process(topic2, "B", "b");
-            driver.setTime(7L);
+            driver.flushState();
+            setRecordContext(7, topic1);
             driver.process(topic2, "D", "d");
-            driver.setTime(8L);
+            driver.flushState();
+            setRecordContext(8, topic1);
             driver.process(topic2, "B", "b");
-            driver.setTime(9L);
+            driver.flushState();
+            setRecordContext(9, topic1);
             driver.process(topic2, "C", "c");
-
+            driver.flushState();
             proc1.checkAndClearProcessResult();
             proc2.checkAndClearProcessResult(
                     "[A@0]:0+a+a+a", "[A@5]:0+a",

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java
index a405da4..ba33d5c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java
@@ -83,25 +83,63 @@ public class KTableAggregateTest {
         driver = new KStreamTestDriver(builder, stateDir);
 
         driver.process(topic1, "A", "1");
+        driver.flushState();
         driver.process(topic1, "B", "2");
+        driver.flushState();
         driver.process(topic1, "A", "3");
+        driver.flushState();
         driver.process(topic1, "B", "4");
+        driver.flushState();
         driver.process(topic1, "C", "5");
+        driver.flushState();
         driver.process(topic1, "D", "6");
+        driver.flushState();
         driver.process(topic1, "B", "7");
+        driver.flushState();
         driver.process(topic1, "C", "8");
+        driver.flushState();
 
         assertEquals(Utils.mkList(
                 "A:0+1",
                 "B:0+2",
-                "A:0+1-1", "A:0+1-1+3",
-                "B:0+2-2", "B:0+2-2+4",
+                "A:0+1-1+3",
+                "B:0+2-2+4",
                 "C:0+5",
                 "D:0+6",
-                "B:0+2-2+4-4", "B:0+2-2+4-4+7",
-                "C:0+5-5", "C:0+5-5+8"), proc.processed);
+                "B:0+2-2+4-4+7",
+                "C:0+5-5+8"), proc.processed);
     }
 
+
+    @Test
+    public void testAggCoalesced() throws Exception {
+        final KStreamBuilder builder = new KStreamBuilder();
+        final String topic1 = "topic1";
+        final MockProcessorSupplier<String, String> proc = new MockProcessorSupplier<>();
+
+        KTable<String, String> table1 = builder.table(stringSerde, stringSerde, topic1, "anyStoreName");
+        KTable<String, String> table2 = table1.groupBy(MockKeyValueMapper.<String, String>NoOpKeyValueMapper(),
+            stringSerde,
+            stringSerde
+        ).aggregate(MockInitializer.STRING_INIT,
+            MockAggregator.STRING_ADDER,
+            MockAggregator.STRING_REMOVER,
+            stringSerde,
+            "topic1-Canonized");
+
+        table2.toStream().process(proc);
+
+        driver = new KStreamTestDriver(builder, stateDir);
+
+        driver.process(topic1, "A", "1");
+        driver.process(topic1, "A", "3");
+        driver.process(topic1, "A", "4");
+        driver.flushState();
+        assertEquals(Utils.mkList(
+            "A:0+4"), proc.processed);
+    }
+
+
     @Test
     public void testAggRepartition() throws Exception {
         final KStreamBuilder builder = new KStreamBuilder();
@@ -135,13 +173,21 @@ public class KTableAggregateTest {
         driver = new KStreamTestDriver(builder, stateDir);
 
         driver.process(topic1, "A", "1");
+        driver.flushState();
         driver.process(topic1, "A", null);
+        driver.flushState();
         driver.process(topic1, "A", "1");
+        driver.flushState();
         driver.process(topic1, "B", "2");
+        driver.flushState();
         driver.process(topic1, "null", "3");
+        driver.flushState();
         driver.process(topic1, "B", "4");
+        driver.flushState();
         driver.process(topic1, "NULL", "5");
+        driver.flushState();
         driver.process(topic1, "B", "7");
+        driver.flushState();
 
         assertEquals(Utils.mkList(
                 "1:0+1",
@@ -170,10 +216,17 @@ public class KTableAggregateTest {
         final KStreamTestDriver driver = new KStreamTestDriver(builder, stateDir);
 
         driver.process(input, "A", "green");
+        driver.flushState();
         driver.process(input, "B", "green");
+        driver.flushState();
         driver.process(input, "A", "blue");
+        driver.flushState();
         driver.process(input, "C", "yellow");
+        driver.flushState();
         driver.process(input, "D", "green");
+        driver.flushState();
+        driver.flushState();
+
 
         assertEquals(Utils.mkList(
                  "green:1",
@@ -183,6 +236,35 @@ public class KTableAggregateTest {
                  "green:2"
                  ), proc.processed);
     }
+
+    @Test
+    public void testCountCoalesced() throws IOException {
+        final KStreamBuilder builder = new KStreamBuilder();
+        final String input = "count-test-input";
+        final MockProcessorSupplier<String, Long> proc = new MockProcessorSupplier<>();
+
+        builder.table(Serdes.String(), Serdes.String(), input, "anyStoreName")
+            .groupBy(MockKeyValueMapper.<String, String>SelectValueKeyValueMapper(), stringSerde, stringSerde)
+            .count("count")
+            .toStream()
+            .process(proc);
+
+        final KStreamTestDriver driver = new KStreamTestDriver(builder, stateDir);
+
+        driver.process(input, "A", "green");
+        driver.process(input, "B", "green");
+        driver.process(input, "A", "blue");
+        driver.process(input, "C", "yellow");
+        driver.process(input, "D", "green");
+        driver.flushState();
+
+
+        assertEquals(Utils.mkList(
+            "blue:1",
+            "yellow:1",
+            "green:2"
+            ), proc.processed);
+    }
     
     @Test
     public void testRemoveOldBeforeAddNew() throws IOException {
@@ -216,22 +298,26 @@ public class KTableAggregateTest {
                     public String apply(String key, String value, String aggregate) {
                         return aggregate.replaceAll(value, "");
                     }
-                }, "someStore")
+                }, Serdes.String(), "someStore")
                 .toStream()
                 .process(proc);
 
         final KStreamTestDriver driver = new KStreamTestDriver(builder, stateDir);
 
         driver.process(input, "11", "A");
+        driver.flushState();
         driver.process(input, "12", "B");
+        driver.flushState();
         driver.process(input, "11", null);
+        driver.flushState();
         driver.process(input, "12", "C");
+        driver.flushState();
 
         assertEquals(Utils.mkList(
                  "1:1",
                  "1:12",
                  "1:2",
-                 "1:", "1:2"
+                 "1:2"
                  ), proc.processed);
     }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java
index 6837c56..277d6d2 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java
@@ -90,8 +90,10 @@ public class KTableFilterTest {
         driver.process(topic1, "B", 2);
         driver.process(topic1, "C", 3);
         driver.process(topic1, "D", 4);
+        driver.flushState();
         driver.process(topic1, "A", null);
         driver.process(topic1, "B", null);
+        driver.flushState();
 
         proc2.checkAndClearProcessResult("A:null", "B:2", "C:null", "D:4", "A:null", "B:null");
         proc3.checkAndClearProcessResult("A:1", "B:null", "C:3", "D:null", "A:null", "B:null");
@@ -203,24 +205,25 @@ public class KTableFilterTest {
         driver.process(topic1, "A", 1);
         driver.process(topic1, "B", 1);
         driver.process(topic1, "C", 1);
+        driver.flushState();
 
         proc1.checkAndClearProcessResult("A:(1<-null)", "B:(1<-null)", "C:(1<-null)");
         proc2.checkAndClearProcessResult("A:(null<-null)", "B:(null<-null)", "C:(null<-null)");
 
         driver.process(topic1, "A", 2);
         driver.process(topic1, "B", 2);
-
+        driver.flushState();
         proc1.checkAndClearProcessResult("A:(2<-null)", "B:(2<-null)");
         proc2.checkAndClearProcessResult("A:(2<-null)", "B:(2<-null)");
 
         driver.process(topic1, "A", 3);
-
+        driver.flushState();
         proc1.checkAndClearProcessResult("A:(3<-null)");
         proc2.checkAndClearProcessResult("A:(null<-null)");
 
         driver.process(topic1, "A", null);
         driver.process(topic1, "B", null);
-
+        driver.flushState();
         proc1.checkAndClearProcessResult("A:(null<-null)", "B:(null<-null)");
         proc2.checkAndClearProcessResult("A:(null<-null)", "B:(null<-null)");
     }
@@ -254,24 +257,25 @@ public class KTableFilterTest {
         driver.process(topic1, "A", 1);
         driver.process(topic1, "B", 1);
         driver.process(topic1, "C", 1);
+        driver.flushState();
 
         proc1.checkAndClearProcessResult("A:(1<-null)", "B:(1<-null)", "C:(1<-null)");
         proc2.checkEmptyAndClearProcessResult();
 
         driver.process(topic1, "A", 2);
         driver.process(topic1, "B", 2);
-
+        driver.flushState();
         proc1.checkAndClearProcessResult("A:(2<-1)", "B:(2<-1)");
         proc2.checkAndClearProcessResult("A:(2<-null)", "B:(2<-null)");
 
         driver.process(topic1, "A", 3);
-
+        driver.flushState();
         proc1.checkAndClearProcessResult("A:(3<-2)");
         proc2.checkAndClearProcessResult("A:(null<-2)");
 
         driver.process(topic1, "A", null);
         driver.process(topic1, "B", null);
-
+        driver.flushState();
         proc1.checkAndClearProcessResult("A:(null<-3)", "B:(null<-2)");
         proc2.checkAndClearProcessResult("B:(null<-2)");
     }
@@ -305,7 +309,7 @@ public class KTableFilterTest {
         driver.process(topic1, "A", "reject");
         driver.process(topic1, "B", "reject");
         driver.process(topic1, "C", "reject");
-
+        driver.flushState();
         proc1.checkAndClearProcessResult("A:(reject<-null)", "B:(reject<-null)", "C:(reject<-null)");
         proc2.checkEmptyAndClearProcessResult();
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableForeachTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableForeachTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableForeachTest.java
index 6fbce82..791fa28 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableForeachTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableForeachTest.java
@@ -96,6 +96,7 @@ public class KTableForeachTest {
         for (KeyValue<Integer, String> record: inputRecords) {
             driver.process(topicName, record.key, record.value);
         }
+        driver.flushState();
 
         assertEquals(expectedRecords.size(), actualRecords.size());
         for (int i = 0; i < expectedRecords.size(); i++) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
index 7edaac9..4b9ea06 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
@@ -113,9 +113,14 @@ public class KTableImplTest {
         driver = new KStreamTestDriver(builder, stateDir);
 
         driver.process(topic1, "A", "01");
+        driver.flushState();
         driver.process(topic1, "B", "02");
+        driver.flushState();
         driver.process(topic1, "C", "03");
+        driver.flushState();
         driver.process(topic1, "D", "04");
+        driver.flushState();
+        driver.flushState();
 
         assertEquals(Utils.mkList("A:01", "B:02", "C:03", "D:04"), proc1.processed);
         assertEquals(Utils.mkList("A:1", "B:2", "C:3", "D:4"), proc2.processed);
@@ -173,6 +178,7 @@ public class KTableImplTest {
         driver.process(topic1, "A", "01");
         driver.process(topic1, "B", "01");
         driver.process(topic1, "C", "01");
+        driver.flushState();
 
         assertEquals("01", getter1.get("A"));
         assertEquals("01", getter1.get("B"));
@@ -192,6 +198,7 @@ public class KTableImplTest {
 
         driver.process(topic1, "A", "02");
         driver.process(topic1, "B", "02");
+        driver.flushState();
 
         assertEquals("02", getter1.get("A"));
         assertEquals("02", getter1.get("B"));
@@ -210,6 +217,7 @@ public class KTableImplTest {
         assertEquals("01", getter4.get("C"));
 
         driver.process(topic1, "A", "03");
+        driver.flushState();
 
         assertEquals("03", getter1.get("A"));
         assertEquals("02", getter1.get("B"));
@@ -228,11 +236,13 @@ public class KTableImplTest {
         assertEquals("01", getter4.get("C"));
 
         driver.process(topic1, "A", null);
+        driver.flushState();
 
         assertNull(getter1.get("A"));
         assertEquals("02", getter1.get("B"));
         assertEquals("01", getter1.get("C"));
 
+
         assertNull(getter2.get("A"));
         assertEquals(new Integer(2), getter2.get("B"));
         assertEquals(new Integer(1), getter2.get("C"));

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java
index 3615b46..ba10668 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java
@@ -103,6 +103,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
+        driver.flushState();
 
         processor.checkAndClearProcessResult("0:null", "1:null");
         checkJoinedValues(getter, kv(0, null), kv(1, null), kv(2, null), kv(3, null));
@@ -112,6 +113,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
+        driver.flushState();
 
         processor.checkAndClearProcessResult("0:X0+Y0", "1:X1+Y1");
         checkJoinedValues(getter, kv(0, "X0+Y0"), kv(1, "X1+Y1"), kv(2, null), kv(3, null));
@@ -121,6 +123,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
+        driver.flushState();
 
         processor.checkAndClearProcessResult("0:X0+Y0", "1:X1+Y1", "2:null", "3:null");
         checkJoinedValues(getter, kv(0, "X0+Y0"), kv(1, "X1+Y1"), kv(2, null), kv(3, null));
@@ -129,6 +132,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic2, expectedKeys[i], "YY" + expectedKeys[i]);
         }
+        driver.flushState();
 
         processor.checkAndClearProcessResult("0:X0+YY0", "1:X1+YY1", "2:X2+YY2", "3:X3+YY3");
         checkJoinedValues(getter, kv(0, "X0+YY0"), kv(1, "X1+YY1"), kv(2, "X2+YY2"), kv(3, "X3+YY3"));
@@ -138,6 +142,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
+        driver.flushState();
 
         processor.checkAndClearProcessResult("0:X0+YY0", "1:X1+YY1", "2:X2+YY2", "3:X3+YY3");
         checkJoinedValues(getter, kv(0, "X0+YY0"), kv(1, "X1+YY1"), kv(2, "X2+YY2"), kv(3, "X3+YY3"));
@@ -147,6 +152,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], null);
         }
+        driver.flushState();
 
         processor.checkAndClearProcessResult("0:null", "1:null");
         checkJoinedValues(getter, kv(0, null), kv(1, null), kv(2, "X2+YY2"), kv(3, "X3+YY3"));
@@ -156,6 +162,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "XX" + expectedKeys[i]);
         }
+        driver.flushState();
 
         processor.checkAndClearProcessResult("0:null", "1:null", "2:XX2+YY2", "3:XX3+YY3");
         checkJoinedValues(getter, kv(0, null), kv(1, null), kv(2, "XX2+YY2"), kv(3, "XX3+YY3"));
@@ -191,6 +198,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
+        driver.flushState();
 
         proc.checkAndClearProcessResult("0:(null<-null)", "1:(null<-null)");
 
@@ -199,6 +207,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
+        driver.flushState();
 
         proc.checkAndClearProcessResult("0:(X0+Y0<-null)", "1:(X1+Y1<-null)");
 
@@ -207,6 +216,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
+        driver.flushState();
 
         proc.checkAndClearProcessResult("0:(X0+Y0<-null)", "1:(X1+Y1<-null)", "2:(null<-null)", "3:(null<-null)");
 
@@ -214,7 +224,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic2, expectedKeys[i], "YY" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+YY0<-null)", "1:(X1+YY1<-null)", "2:(X2+YY2<-null)", "3:(X3+YY3<-null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -222,7 +232,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+YY0<-null)", "1:(X1+YY1<-null)", "2:(X2+YY2<-null)", "3:(X3+YY3<-null)");
 
         // push two items with null to the other stream as deletes. this should produce two item.
@@ -230,7 +240,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], null);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(null<-null)", "1:(null<-null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -238,7 +248,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "XX" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(null<-null)", "1:(null<-null)", "2:(XX2+YY2<-null)", "3:(XX3+YY3<-null)");
     }
 
@@ -274,7 +284,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(null<-null)", "1:(null<-null)");
 
         // push two items to the other stream. this should produce two items.
@@ -282,7 +292,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+Y0<-null)", "1:(X1+Y1<-null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -290,14 +300,14 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+Y0<-X0+Y0)", "1:(X1+Y1<-X1+Y1)", "2:(null<-null)", "3:(null<-null)");
 
         // push all items to the other stream. this should produce four items.
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic2, expectedKeys[i], "YY" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+YY0<-X0+Y0)", "1:(X1+YY1<-X1+Y1)", "2:(X2+YY2<-null)", "3:(X3+YY3<-null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -305,7 +315,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+YY0<-X0+YY0)", "1:(X1+YY1<-X1+YY1)", "2:(X2+YY2<-X2+YY2)", "3:(X3+YY3<-X3+YY3)");
 
         // push two items with null to the other stream as deletes. this should produce two item.
@@ -313,7 +323,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], null);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(null<-X0+YY0)", "1:(null<-X1+YY1)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -321,7 +331,7 @@ public class KTableKTableJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "XX" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(null<-null)", "1:(null<-null)", "2:(XX2+YY2<-X2+YY2)", "3:(XX3+YY3<-X3+YY3)");
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java
index ec07116..5f84678 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java
@@ -99,6 +99,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
+        driver.flushState();
 
         processor.checkAndClearProcessResult("0:X0+null", "1:X1+null");
         checkJoinedValues(getter, kv(0, "X0+null"), kv(1, "X1+null"), kv(2, null), kv(3, null));
@@ -108,7 +109,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:X0+Y0", "1:X1+Y1");
         checkJoinedValues(getter, kv(0, "X0+Y0"), kv(1, "X1+Y1"), kv(2, null), kv(3, null));
 
@@ -117,7 +118,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:X0+Y0", "1:X1+Y1", "2:X2+null", "3:X3+null");
         checkJoinedValues(getter, kv(0, "X0+Y0"), kv(1, "X1+Y1"), kv(2, "X2+null"), kv(3, "X3+null"));
 
@@ -125,7 +126,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic2, expectedKeys[i], "YY" + expectedKeys[i]);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:X0+YY0", "1:X1+YY1", "2:X2+YY2", "3:X3+YY3");
         checkJoinedValues(getter, kv(0, "X0+YY0"), kv(1, "X1+YY1"), kv(2, "X2+YY2"), kv(3, "X3+YY3"));
 
@@ -134,7 +135,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:X0+YY0", "1:X1+YY1", "2:X2+YY2", "3:X3+YY3");
         checkJoinedValues(getter, kv(0, "X0+YY0"), kv(1, "X1+YY1"), kv(2, "X2+YY2"), kv(3, "X3+YY3"));
 
@@ -143,7 +144,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], null);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:X0+null", "1:X1+null");
         checkJoinedValues(getter, kv(0, "X0+null"), kv(1, "X1+null"), kv(2, "X2+YY2"), kv(3, "X3+YY3"));
 
@@ -152,7 +153,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "XX" + expectedKeys[i]);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:XX0+null", "1:XX1+null", "2:XX2+YY2", "3:XX3+YY3");
         checkJoinedValues(getter, kv(0, "XX0+null"), kv(1, "XX1+null"), kv(2, "XX2+YY2"), kv(3, "XX3+YY3"));
     }
@@ -187,7 +188,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+null<-null)", "1:(X1+null<-null)");
 
         // push two items to the other stream. this should produce two items.
@@ -195,7 +196,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+Y0<-null)", "1:(X1+Y1<-null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -203,14 +204,14 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+Y0<-null)", "1:(X1+Y1<-null)", "2:(X2+null<-null)", "3:(X3+null<-null)");
 
         // push all items to the other stream. this should produce four items.
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic2, expectedKeys[i], "YY" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+YY0<-null)", "1:(X1+YY1<-null)", "2:(X2+YY2<-null)", "3:(X3+YY3<-null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -218,7 +219,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+YY0<-null)", "1:(X1+YY1<-null)", "2:(X2+YY2<-null)", "3:(X3+YY3<-null)");
 
         // push two items with null to the other stream as deletes. this should produce two item.
@@ -226,7 +227,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], null);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+null<-null)", "1:(X1+null<-null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -234,7 +235,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "XX" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(XX0+null<-null)", "1:(XX1+null<-null)", "2:(XX2+YY2<-null)", "3:(XX3+YY3<-null)");
     }
 
@@ -270,7 +271,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+null<-null)", "1:(X1+null<-null)");
 
         // push two items to the other stream. this should produce two items.
@@ -278,7 +279,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+Y0<-X0+null)", "1:(X1+Y1<-X1+null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -286,14 +287,14 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+Y0<-X0+Y0)", "1:(X1+Y1<-X1+Y1)", "2:(X2+null<-null)", "3:(X3+null<-null)");
 
         // push all items to the other stream. this should produce four items.
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic2, expectedKeys[i], "YY" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+YY0<-X0+Y0)", "1:(X1+YY1<-X1+Y1)", "2:(X2+YY2<-X2+null)", "3:(X3+YY3<-X3+null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -301,7 +302,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+YY0<-X0+YY0)", "1:(X1+YY1<-X1+YY1)", "2:(X2+YY2<-X2+YY2)", "3:(X3+YY3<-X3+YY3)");
 
         // push two items with null to the other stream as deletes. this should produce two item.
@@ -309,7 +310,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], null);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+null<-X0+YY0)", "1:(X1+null<-X1+YY1)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -317,7 +318,7 @@ public class KTableKTableLeftJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "XX" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(XX0+null<-X0+null)", "1:(XX1+null<-X1+null)", "2:(XX2+YY2<-X2+YY2)", "3:(XX3+YY3<-X3+YY3)");
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
index 33dfb04..a6249bc 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
@@ -93,7 +93,6 @@ public class KTableKTableOuterJoinTest {
         KTableValueGetterSupplier<Integer, String> getterSupplier = ((KTableImpl<Integer, String, String>) joined).valueGetterSupplier();
 
         driver = new KStreamTestDriver(builder, stateDir);
-        driver.setTime(0L);
 
         KTableValueGetter<Integer, String> getter = getterSupplier.get();
         getter.init(driver.context());
@@ -103,7 +102,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:X0+null", "1:X1+null");
         checkJoinedValues(getter, kv(0, "X0+null"), kv(1, "X1+null"), kv(2, null), kv(3, null));
 
@@ -112,7 +111,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:X0+Y0", "1:X1+Y1");
         checkJoinedValues(getter, kv(0, "X0+Y0"), kv(1, "X1+Y1"), kv(2, null), kv(3, null));
 
@@ -121,7 +120,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:X0+Y0", "1:X1+Y1", "2:X2+null", "3:X3+null");
         checkJoinedValues(getter, kv(0, "X0+Y0"), kv(1, "X1+Y1"), kv(2, "X2+null"), kv(3, "X3+null"));
 
@@ -129,7 +128,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic2, expectedKeys[i], "YY" + expectedKeys[i]);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:X0+YY0", "1:X1+YY1", "2:X2+YY2", "3:X3+YY3");
         checkJoinedValues(getter, kv(0, "X0+YY0"), kv(1, "X1+YY1"), kv(2, "X2+YY2"), kv(3, "X3+YY3"));
 
@@ -138,7 +137,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:X0+YY0", "1:X1+YY1", "2:X2+YY2", "3:X3+YY3");
         checkJoinedValues(getter, kv(0, "X0+YY0"), kv(1, "X1+YY1"), kv(2, "X2+YY2"), kv(3, "X3+YY3"));
 
@@ -147,7 +146,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], null);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:X0+null", "1:X1+null");
         checkJoinedValues(getter, kv(0, "X0+null"), kv(1, "X1+null"), kv(2, "X2+YY2"), kv(3, "X3+YY3"));
 
@@ -156,7 +155,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "XX" + expectedKeys[i]);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("0:XX0+null", "1:XX1+null", "2:XX2+YY2", "3:XX3+YY3");
         checkJoinedValues(getter, kv(0, "XX0+null"), kv(1, "XX1+null"), kv(2, "XX2+YY2"), kv(3, "XX3+YY3"));
 
@@ -165,7 +164,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 1; i < 3; i++) {
             driver.process(topic1, expectedKeys[i], null);
         }
-
+        driver.flushState();
         processor.checkAndClearProcessResult("1:null", "2:null+YY2");
         checkJoinedValues(getter, kv(0, "XX0+null"), kv(1, null), kv(2, "null+YY2"), kv(3, "XX3+YY3"));
     }
@@ -189,7 +188,6 @@ public class KTableKTableOuterJoinTest {
         builder.addProcessor("proc", proc, ((KTableImpl<?, ?, ?>) joined).name);
 
         driver = new KStreamTestDriver(builder, stateDir);
-        driver.setTime(0L);
 
         assertFalse(((KTableImpl<?, ?, ?>) table1).sendingOldValueEnabled());
         assertFalse(((KTableImpl<?, ?, ?>) table2).sendingOldValueEnabled());
@@ -200,7 +198,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+null<-null)", "1:(X1+null<-null)");
 
         // push two items to the other stream. this should produce two items.
@@ -208,7 +206,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+Y0<-null)", "1:(X1+Y1<-null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -216,14 +214,14 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+Y0<-null)", "1:(X1+Y1<-null)", "2:(X2+null<-null)", "3:(X3+null<-null)");
 
         // push all items to the other stream. this should produce four items.
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic2, expectedKeys[i], "YY" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+YY0<-null)", "1:(X1+YY1<-null)", "2:(X2+YY2<-null)", "3:(X3+YY3<-null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -231,7 +229,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+YY0<-null)", "1:(X1+YY1<-null)", "2:(X2+YY2<-null)", "3:(X3+YY3<-null)");
 
         // push two items with null to the other stream as deletes. this should produce two item.
@@ -239,7 +237,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], null);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+null<-null)", "1:(X1+null<-null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -247,7 +245,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "XX" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(XX0+null<-null)", "1:(XX1+null<-null)", "2:(XX2+YY2<-null)", "3:(XX3+YY3<-null)");
 
         // push middle two items to the primary stream with null. this should produce two items.
@@ -255,7 +253,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 1; i < 3; i++) {
             driver.process(topic1, expectedKeys[i], null);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("1:(null<-null)", "2:(null+YY2<-null)");
     }
 
@@ -280,7 +278,6 @@ public class KTableKTableOuterJoinTest {
         builder.addProcessor("proc", proc, ((KTableImpl<?, ?, ?>) joined).name);
 
         driver = new KStreamTestDriver(builder, stateDir);
-        driver.setTime(0L);
 
         assertTrue(((KTableImpl<?, ?, ?>) table1).sendingOldValueEnabled());
         assertTrue(((KTableImpl<?, ?, ?>) table2).sendingOldValueEnabled());
@@ -291,7 +288,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+null<-null)", "1:(X1+null<-null)");
 
         // push two items to the other stream. this should produce two items.
@@ -299,7 +296,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+Y0<-X0+null)", "1:(X1+Y1<-X1+null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -307,14 +304,14 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+Y0<-X0+Y0)", "1:(X1+Y1<-X1+Y1)", "2:(X2+null<-null)", "3:(X3+null<-null)");
 
         // push all items to the other stream. this should produce four items.
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic2, expectedKeys[i], "YY" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+YY0<-X0+Y0)", "1:(X1+YY1<-X1+Y1)", "2:(X2+YY2<-X2+null)", "3:(X3+YY3<-X3+null)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -322,7 +319,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+YY0<-X0+YY0)", "1:(X1+YY1<-X1+YY1)", "2:(X2+YY2<-X2+YY2)", "3:(X3+YY3<-X3+YY3)");
 
         // push two items with null to the other stream as deletes. this should produce two item.
@@ -330,7 +327,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < 2; i++) {
             driver.process(topic2, expectedKeys[i], null);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(X0+null<-X0+YY0)", "1:(X1+null<-X1+YY1)");
 
         // push all four items to the primary stream. this should produce four items.
@@ -338,7 +335,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 0; i < expectedKeys.length; i++) {
             driver.process(topic1, expectedKeys[i], "XX" + expectedKeys[i]);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("0:(XX0+null<-X0+null)", "1:(XX1+null<-X1+null)", "2:(XX2+YY2<-X2+YY2)", "3:(XX3+YY3<-X3+YY3)");
 
         // push middle two items to the primary stream with null. this should produce two items.
@@ -346,7 +343,7 @@ public class KTableKTableOuterJoinTest {
         for (int i = 1; i < 3; i++) {
             driver.process(topic1, expectedKeys[i], null);
         }
-
+        driver.flushState();
         proc.checkAndClearProcessResult("1:(null<-XX1+null)", "2:(null+YY2<-XX2+YY2)");
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapKeysTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapKeysTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapKeysTest.java
index 78cff18..864b274 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapKeysTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapKeysTest.java
@@ -93,10 +93,10 @@ public class KTableMapKeysTest {
         convertedStream.process(processor);
 
         driver = new KStreamTestDriver(builder, stateDir);
-
         for (int i = 0;  i < originalKeys.length; i++) {
             driver.process(topic1, originalKeys[i], values[i]);
         }
+        driver.flushState();
 
         assertEquals(3, processor.processed.size());
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapValuesTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapValuesTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapValuesTest.java
index 5739397..1d5b8a9 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapValuesTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapValuesTest.java
@@ -82,7 +82,7 @@ public class KTableMapValuesTest {
         driver.process(topic1, "B", "02");
         driver.process(topic1, "C", "03");
         driver.process(topic1, "D", "04");
-
+        driver.flushState();
         assertEquals(Utils.mkList("A:1", "B:2", "C:3", "D:4"), proc2.processed);
     }
 
@@ -120,7 +120,6 @@ public class KTableMapValuesTest {
         KTableValueGetterSupplier<String, String> getterSupplier4 = table4.valueGetterSupplier();
 
         driver = new KStreamTestDriver(builder, stateDir, null, null);
-
         KTableValueGetter<String, String> getter1 = getterSupplier1.get();
         getter1.init(driver.context());
         KTableValueGetter<String, Integer> getter2 = getterSupplier2.get();
@@ -133,6 +132,7 @@ public class KTableMapValuesTest {
         driver.process(topic1, "A", "01");
         driver.process(topic1, "B", "01");
         driver.process(topic1, "C", "01");
+        driver.flushState();
 
         assertEquals("01", getter1.get("A"));
         assertEquals("01", getter1.get("B"));
@@ -152,6 +152,7 @@ public class KTableMapValuesTest {
 
         driver.process(topic1, "A", "02");
         driver.process(topic1, "B", "02");
+        driver.flushState();
 
         assertEquals("02", getter1.get("A"));
         assertEquals("02", getter1.get("B"));
@@ -170,6 +171,7 @@ public class KTableMapValuesTest {
         assertEquals("01", getter4.get("C"));
 
         driver.process(topic1, "A", "03");
+        driver.flushState();
 
         assertEquals("03", getter1.get("A"));
         assertEquals("02", getter1.get("B"));
@@ -188,6 +190,7 @@ public class KTableMapValuesTest {
         assertEquals("01", getter4.get("C"));
 
         driver.process(topic1, "A", null);
+        driver.flushState();
 
         assertNull(getter1.get("A"));
         assertEquals("02", getter1.get("B"));
@@ -227,26 +230,29 @@ public class KTableMapValuesTest {
         builder.addProcessor("proc", proc, table2.name);
 
         driver = new KStreamTestDriver(builder, stateDir, null, null);
-
         assertFalse(table1.sendingOldValueEnabled());
         assertFalse(table2.sendingOldValueEnabled());
 
         driver.process(topic1, "A", "01");
         driver.process(topic1, "B", "01");
         driver.process(topic1, "C", "01");
+        driver.flushState();
 
         proc.checkAndClearProcessResult("A:(1<-null)", "B:(1<-null)", "C:(1<-null)");
 
         driver.process(topic1, "A", "02");
         driver.process(topic1, "B", "02");
+        driver.flushState();
 
         proc.checkAndClearProcessResult("A:(2<-null)", "B:(2<-null)");
 
         driver.process(topic1, "A", "03");
+        driver.flushState();
 
         proc.checkAndClearProcessResult("A:(3<-null)");
 
         driver.process(topic1, "A", null);
+        driver.flushState();
 
         proc.checkAndClearProcessResult("A:(null<-null)");
     }
@@ -274,26 +280,29 @@ public class KTableMapValuesTest {
         builder.addProcessor("proc", proc, table2.name);
 
         driver = new KStreamTestDriver(builder, stateDir, null, null);
-
         assertTrue(table1.sendingOldValueEnabled());
         assertTrue(table2.sendingOldValueEnabled());
 
         driver.process(topic1, "A", "01");
         driver.process(topic1, "B", "01");
         driver.process(topic1, "C", "01");
+        driver.flushState();
 
         proc.checkAndClearProcessResult("A:(1<-null)", "B:(1<-null)", "C:(1<-null)");
 
         driver.process(topic1, "A", "02");
         driver.process(topic1, "B", "02");
+        driver.flushState();
 
         proc.checkAndClearProcessResult("A:(2<-1)", "B:(2<-1)");
 
         driver.process(topic1, "A", "03");
+        driver.flushState();
 
         proc.checkAndClearProcessResult("A:(3<-2)");
 
         driver.process(topic1, "A", null);
+        driver.flushState();
 
         proc.checkAndClearProcessResult("A:(null<-3)");
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableSourceTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableSourceTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableSourceTest.java
index cd1262b..5602555 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableSourceTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableSourceTest.java
@@ -69,13 +69,14 @@ public class KTableSourceTest {
         table1.toStream().process(proc1);
 
         driver = new KStreamTestDriver(builder, stateDir);
-
         driver.process(topic1, "A", 1);
         driver.process(topic1, "B", 2);
         driver.process(topic1, "C", 3);
         driver.process(topic1, "D", 4);
+        driver.flushState();
         driver.process(topic1, "A", null);
         driver.process(topic1, "B", null);
+        driver.flushState();
 
         assertEquals(Utils.mkList("A:1", "B:2", "C:3", "D:4", "A:null", "B:null"), proc1.processed);
     }
@@ -91,7 +92,6 @@ public class KTableSourceTest {
         KTableValueGetterSupplier<String, String> getterSupplier1 = table1.valueGetterSupplier();
 
         driver = new KStreamTestDriver(builder, stateDir, null, null);
-
         KTableValueGetter<String, String> getter1 = getterSupplier1.get();
         getter1.init(driver.context());
 
@@ -138,24 +138,27 @@ public class KTableSourceTest {
         builder.addProcessor("proc1", proc1, table1.name);
 
         driver = new KStreamTestDriver(builder, stateDir, null, null);
-
         driver.process(topic1, "A", "01");
         driver.process(topic1, "B", "01");
         driver.process(topic1, "C", "01");
+        driver.flushState();
 
         proc1.checkAndClearProcessResult("A:(01<-null)", "B:(01<-null)", "C:(01<-null)");
 
         driver.process(topic1, "A", "02");
         driver.process(topic1, "B", "02");
+        driver.flushState();
 
         proc1.checkAndClearProcessResult("A:(02<-null)", "B:(02<-null)");
 
         driver.process(topic1, "A", "03");
+        driver.flushState();
 
         proc1.checkAndClearProcessResult("A:(03<-null)");
 
         driver.process(topic1, "A", null);
         driver.process(topic1, "B", null);
+        driver.flushState();
 
         proc1.checkAndClearProcessResult("A:(null<-null)", "B:(null<-null)");
     }
@@ -181,20 +184,24 @@ public class KTableSourceTest {
         driver.process(topic1, "A", "01");
         driver.process(topic1, "B", "01");
         driver.process(topic1, "C", "01");
+        driver.flushState();
 
         proc1.checkAndClearProcessResult("A:(01<-null)", "B:(01<-null)", "C:(01<-null)");
 
         driver.process(topic1, "A", "02");
         driver.process(topic1, "B", "02");
+        driver.flushState();
 
         proc1.checkAndClearProcessResult("A:(02<-01)", "B:(02<-01)");
 
         driver.process(topic1, "A", "03");
+        driver.flushState();
 
         proc1.checkAndClearProcessResult("A:(03<-02)");
 
         driver.process(topic1, "A", null);
         driver.process(topic1, "B", null);
+        driver.flushState();
 
         proc1.checkAndClearProcessResult("A:(null<-03)", "B:(null<-02)");
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java b/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
index 5495416..630167a 100644
--- a/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
+++ b/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
@@ -38,6 +38,7 @@ import org.apache.kafka.streams.kstream.ForeachAction;
 import org.apache.kafka.streams.kstream.JoinWindows;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KStreamBuilder;
+import org.apache.kafka.streams.processor.AbstractProcessor;
 import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.ValueJoiner;
 import org.apache.kafka.streams.processor.Processor;
@@ -435,7 +436,7 @@ public class SimpleBenchmark {
         source.process(new ProcessorSupplier<Integer, byte[]>() {
             @Override
             public Processor<Integer, byte[]> get() {
-                return new Processor<Integer, byte[]>() {
+                return new AbstractProcessor<Integer, byte[]>() {
 
                     @Override
                     public void init(ProcessorContext context) {
@@ -479,8 +480,7 @@ public class SimpleBenchmark {
         source.process(new ProcessorSupplier<Integer, byte[]>() {
             @Override
             public Processor<Integer, byte[]> get() {
-                return new Processor<Integer, byte[]>() {
-
+                return new AbstractProcessor<Integer, byte[]>() {
                     @Override
                     public void init(ProcessorContext context) {
                     }
@@ -573,8 +573,7 @@ public class SimpleBenchmark {
         source.process(new ProcessorSupplier<Integer, byte[]>() {
             @Override
             public Processor<Integer, byte[]> get() {
-                return new Processor<Integer, byte[]>() {
-
+                return new AbstractProcessor<Integer, byte[]>() {
                     KeyValueStore<Integer, byte[]> store;
 
                     @SuppressWarnings("unchecked")

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java
index a4c008a..7fe5170 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java
@@ -237,13 +237,13 @@ public class TopologyBuilderTest {
         builder.addStateStore(supplier);
         builder.setApplicationId("X");
 
-        assertEquals(0, builder.build(null).stateStoreSuppliers().size());
+        assertEquals(0, builder.build(null).stateStores().size());
 
         builder.addSource("source-1", "topic-1");
         builder.addProcessor("processor-1", new MockProcessorSupplier(), "source-1");
         builder.connectProcessorAndStateStores("processor-1", "store-1");
 
-        List<StateStoreSupplier> suppliers = builder.build(null).stateStoreSuppliers();
+        List<StateStore> suppliers = builder.build(null).stateStores();
         assertEquals(1, suppliers.size());
         assertEquals(supplier.name(), suppliers.get(0).name());
     }
@@ -471,7 +471,7 @@ public class TopologyBuilderTest {
         builder.setApplicationId("appId");
         builder.addSource("source", "topic");
         builder.addProcessor("processor", new MockProcessorSupplier(), "source");
-        builder.addStateStore(new RocksDBWindowStoreSupplier("store", 30000, 3, false, null, null, true, Collections.<String, String>emptyMap()), "processor");
+        builder.addStateStore(new RocksDBWindowStoreSupplier("store", 30000, 3, false, null, null, 10000, true, Collections.<String, String>emptyMap(), false), "processor");
         final Map<Integer, TopicsInfo> topicGroups = builder.topicGroups();
         final TopicsInfo topicsInfo = topicGroups.values().iterator().next();
         final InternalTopicConfig topicConfig = topicsInfo.stateChangelogTopics.get("appId-store-changelog");

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
index 32dce6d..5802b29 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
@@ -29,6 +29,7 @@ import org.apache.kafka.common.serialization.IntegerSerializer;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.processor.TaskId;
 import org.apache.kafka.streams.state.internals.OffsetCheckpoint;
 import org.apache.kafka.test.MockStateStoreSupplier;
@@ -209,7 +210,7 @@ public class ProcessorStateManagerTest {
     public void testNoTopic() throws IOException {
         MockStateStoreSupplier.MockStateStore mockStateStore = new MockStateStoreSupplier.MockStateStore(nonPersistentStoreName, false);
 
-        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, new TaskId(0, 1), noPartitions, new MockRestoreConsumer(), false, stateDirectory, null);
+        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, new TaskId(0, 1), noPartitions, new MockRestoreConsumer(), false, stateDirectory, null, Collections.<StateStore, ProcessorNode>emptyMap());
         try {
             stateMgr.register(mockStateStore, true, mockStateStore.stateRestoreCallback);
         } finally {
@@ -237,7 +238,7 @@ public class ProcessorStateManagerTest {
 
         MockStateStoreSupplier.MockStateStore persistentStore = new MockStateStoreSupplier.MockStateStore("persistentStore", true); // persistent store
 
-        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, taskId, noPartitions, restoreConsumer, false, stateDirectory, null);
+        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, taskId, noPartitions, restoreConsumer, false, stateDirectory, null, Collections.<StateStore, ProcessorNode>emptyMap());
         try {
             restoreConsumer.reset();
 
@@ -286,7 +287,7 @@ public class ProcessorStateManagerTest {
 
         MockStateStoreSupplier.MockStateStore nonPersistentStore = new MockStateStoreSupplier.MockStateStore(nonPersistentStoreName, false); // non persistent store
 
-        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, new TaskId(0, 2), noPartitions, restoreConsumer, false, stateDirectory, null);
+        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, new TaskId(0, 2), noPartitions, restoreConsumer, false, stateDirectory, null, Collections.<StateStore, ProcessorNode>emptyMap());
         try {
             restoreConsumer.reset();
 
@@ -359,7 +360,7 @@ public class ProcessorStateManagerTest {
         // if there is an source partition, inherit the partition id
         Set<TopicPartition> sourcePartitions = Utils.mkSet(new TopicPartition(storeTopicName3, 1));
 
-        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, taskId, sourcePartitions, restoreConsumer, true, stateDirectory, null); // standby
+        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, taskId, sourcePartitions, restoreConsumer, true, stateDirectory, null, Collections.<StateStore, ProcessorNode>emptyMap()); // standby
         try {
             restoreConsumer.reset();
 
@@ -393,7 +394,7 @@ public class ProcessorStateManagerTest {
 
         MockStateStoreSupplier.MockStateStore mockStateStore = new MockStateStoreSupplier.MockStateStore(nonPersistentStoreName, false);
 
-        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, new TaskId(0, 1), noPartitions, restoreConsumer, false, stateDirectory, null);
+        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, new TaskId(0, 1), noPartitions, restoreConsumer, false, stateDirectory, null, Collections.<StateStore, ProcessorNode>emptyMap());
         try {
             stateMgr.register(mockStateStore, true, mockStateStore.stateRestoreCallback);
 
@@ -431,7 +432,7 @@ public class ProcessorStateManagerTest {
         MockStateStoreSupplier.MockStateStore persistentStore = new MockStateStoreSupplier.MockStateStore(persistentStoreName, true);
         MockStateStoreSupplier.MockStateStore nonPersistentStore = new MockStateStoreSupplier.MockStateStore(nonPersistentStoreName, false);
 
-        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, taskId, noPartitions, restoreConsumer, false, stateDirectory, null);
+        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, taskId, noPartitions, restoreConsumer, false, stateDirectory, null, Collections.<StateStore, ProcessorNode>emptyMap());
         try {
             // make sure the checkpoint file is deleted
             assertFalse(checkpointFile.exists());

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
index 09434c3..54ee43c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
@@ -210,6 +210,7 @@ public class ProcessorTopologyTest {
         assertNoOutputRecord(OUTPUT_TOPIC_1);
     }
 
+
     protected void assertNextOutputRecord(String topic, String key, String value) {
         ProducerRecord<String, String> record = driver.readOutput(topic, STRING_DESERIALIZER, STRING_DESERIALIZER);
         assertEquals(topic, record.topic());
@@ -269,6 +270,7 @@ public class ProcessorTopologyTest {
                                     .addSink("counts", OUTPUT_TOPIC_1, "processor");
     }
 
+
     protected TopologyBuilder createSimpleMultiSourceTopology(int partition) {
         return new TopologyBuilder().addSource("source-1", STRING_DESERIALIZER, STRING_DESERIALIZER, INPUT_TOPIC_1)
                 .addProcessor("processor-1", define(new ForwardingProcessor()), "source-1")

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/processor/internals/PunctuationQueueTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/PunctuationQueueTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/PunctuationQueueTest.java
index 2c7aaeb..63b2b75 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/PunctuationQueueTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/PunctuationQueueTest.java
@@ -17,7 +17,7 @@
 
 package org.apache.kafka.streams.processor.internals;
 
-import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.AbstractProcessor;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.junit.Test;
 
@@ -60,7 +60,7 @@ public class PunctuationQueueTest {
         assertEquals(2, processor.punctuatedAt.size());
     }
 
-    private static class TestProcessor implements Processor<String, String> {
+    private static class TestProcessor extends AbstractProcessor<String, String> {
 
         public final ArrayList<Long> punctuatedAt = new ArrayList<>();
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordContextStub.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordContextStub.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordContextStub.java
new file mode 100644
index 0000000..3d5f3e3
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordContextStub.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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.streams.processor.internals;
+
+public class RecordContextStub implements RecordContext {
+
+    private final long offset;
+    private final long timestamp;
+    private final int partition;
+    private final String topic;
+
+    public RecordContextStub() {
+        this(-1, -1, -1, "");
+    }
+    public RecordContextStub(final long offset, final long timestamp, final int partition, final String topic) {
+        this.offset = offset;
+        this.timestamp = timestamp;
+        this.partition = partition;
+        this.topic = topic;
+    }
+
+    @Override
+    public long offset() {
+        return offset;
+    }
+
+    @Override
+    public long timestamp() {
+        return timestamp;
+    }
+
+    @Override
+    public String topic() {
+        return topic;
+    }
+
+    @Override
+    public int partition() {
+        return partition;
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
index 11058c2..268697c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
@@ -29,7 +29,7 @@ import org.apache.kafka.common.serialization.IntegerSerializer;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.processor.StateStoreSupplier;
+import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.processor.TaskId;
 import org.apache.kafka.streams.state.internals.OffsetCheckpoint;
 import org.apache.kafka.test.MockStateStoreSupplier;
@@ -74,12 +74,12 @@ public class StandbyTaskTest {
             Collections.<ProcessorNode>emptyList(),
             Collections.<String, SourceNode>emptyMap(),
             Collections.<String, SinkNode>emptyMap(),
-            Utils.<StateStoreSupplier>mkList(
-                    new MockStateStoreSupplier(storeName1, false),
-                    new MockStateStoreSupplier(storeName2, true)
+            Utils.mkList(
+                    new MockStateStoreSupplier(storeName1, false).get(),
+                    new MockStateStoreSupplier(storeName2, true).get()
             ),
-            Collections.<String, String>emptyMap()
-    );
+            Collections.<String, String>emptyMap(),
+            Collections.<StateStore, ProcessorNode>emptyMap());
 
     private final TopicPartition ktable = new TopicPartition("ktable1", 0);
     private final Set<TopicPartition> ktablePartitions = Utils.mkSet(ktable);
@@ -87,15 +87,15 @@ public class StandbyTaskTest {
             Collections.<ProcessorNode>emptyList(),
             Collections.<String, SourceNode>emptyMap(),
             Collections.<String, SinkNode>emptyMap(),
-            Utils.<StateStoreSupplier>mkList(
-                    new MockStateStoreSupplier(ktable.topic(), true, false)
+            Utils.mkList(
+                    new MockStateStoreSupplier(ktable.topic(), true, false).get()
             ),
             new HashMap<String, String>() {
             {
                 put("ktable1", ktable.topic());
             }
-        }
-    );
+        },
+            Collections.<StateStore, ProcessorNode>emptyMap());
     private File baseDir;
     private StateDirectory stateDirectory;
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
index 32d6aa4..2b05e80 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
@@ -30,7 +30,7 @@ import org.apache.kafka.common.serialization.IntegerSerializer;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.processor.StateStoreSupplier;
+import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.processor.TaskId;
 import org.apache.kafka.test.MockProcessorNode;
 import org.apache.kafka.test.MockSourceNode;
@@ -75,9 +75,9 @@ public class StreamTaskTest {
                 }
             },
             Collections.<String, SinkNode>emptyMap(),
-            Collections.<StateStoreSupplier>emptyList(),
-            Collections.<String, String>emptyMap()
-    );
+            Collections.<StateStore>emptyList(),
+            Collections.<String, String>emptyMap(),
+            Collections.<StateStore, ProcessorNode>emptyMap());
     private File baseDir;
     private StateDirectory stateDirectory;
 
@@ -119,7 +119,7 @@ public class StreamTaskTest {
     @Test
     public void testProcessOrder() throws Exception {
         StreamsConfig config = createConfig(baseDir);
-        StreamTask task = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer, producer, restoreStateConsumer, config, null, stateDirectory);
+        StreamTask task = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer, producer, restoreStateConsumer, config, null, stateDirectory, null);
 
         task.addRecords(partition1, records(
                 new ConsumerRecord<>(partition1.topic(), partition1.partition(), 10, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),
@@ -166,7 +166,7 @@ public class StreamTaskTest {
     @Test
     public void testPauseResume() throws Exception {
         StreamsConfig config = createConfig(baseDir);
-        StreamTask task = new StreamTask(new TaskId(1, 1), "applicationId", partitions, topology, consumer, producer, restoreStateConsumer, config, null, stateDirectory);
+        StreamTask task = new StreamTask(new TaskId(1, 1), "applicationId", partitions, topology, consumer, producer, restoreStateConsumer, config, null, stateDirectory, null);
 
         task.addRecords(partition1, records(
                 new ConsumerRecord<>(partition1.topic(), partition1.partition(), 10, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),
@@ -225,7 +225,7 @@ public class StreamTaskTest {
     @Test
     public void testMaybePunctuate() throws Exception {
         StreamsConfig config = createConfig(baseDir);
-        StreamTask task = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer, producer, restoreStateConsumer, config, null, stateDirectory);
+        StreamTask task = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer, producer, restoreStateConsumer, config, null, stateDirectory, null);
 
         task.addRecords(partition1, records(
                 new ConsumerRecord<>(partition1.topic(), partition1.partition(), 20, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index 1da7592..ccbf8d6 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -133,7 +133,7 @@ public class StreamThreadTest {
                               Consumer<byte[], byte[]> restoreConsumer,
                               StreamsConfig config,
                               StateDirectory stateDirectory) {
-            super(id, applicationId, partitions, topology, consumer, producer, restoreConsumer, config, null, stateDirectory);
+            super(id, applicationId, partitions, topology, consumer, producer, restoreConsumer, config, null, stateDirectory, null);
         }
 
         @Override

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestUtil.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestUtil.java b/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestUtil.java
index f1c237e..7ff738f 100644
--- a/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestUtil.java
+++ b/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestUtil.java
@@ -24,6 +24,7 @@ import org.apache.kafka.streams.kstream.Aggregator;
 import org.apache.kafka.streams.kstream.Initializer;
 import org.apache.kafka.streams.kstream.KeyValueMapper;
 import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.processor.AbstractProcessor;
 import org.apache.kafka.streams.processor.Processor;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.ProcessorSupplier;
@@ -43,7 +44,7 @@ public class SmokeTestUtil {
     public static <T> ProcessorSupplier<String, T> printProcessorSupplier(final String topic, final boolean printOffset) {
         return new ProcessorSupplier<String, T>() {
             public Processor<String, T> get() {
-                return new Processor<String, T>() {
+                return new AbstractProcessor<String, T>() {
                     private int numRecordsProcessed = 0;
                     private ProcessorContext context;
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
index 140ea35..e84e9ba 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
@@ -33,6 +33,7 @@ import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.processor.StreamPartitioner;
 import org.apache.kafka.streams.processor.TaskId;
 import org.apache.kafka.streams.processor.internals.RecordCollector;
+import org.apache.kafka.streams.state.internals.ThreadCache;
 import org.apache.kafka.test.MockProcessorContext;
 import org.apache.kafka.test.MockTimestampExtractor;
 import org.apache.kafka.test.TestUtils;
@@ -178,6 +179,8 @@ public class KeyValueStoreTestDriver<K, V> {
     private final List<KeyValue<K, V>> restorableEntries = new LinkedList<>();
     private final MockProcessorContext context;
     private final Map<String, StateStore> storeMap = new HashMap<>();
+    private static final long DEFAULT_CACHE_SIZE_BYTES = 1 * 1024 * 1024L;
+    private final ThreadCache cache = new ThreadCache(DEFAULT_CACHE_SIZE_BYTES);
     private final StreamsMetrics metrics = new StreamsMetrics() {
         @Override
         public Sensor addLatencySensor(String scopeName, String entityName, String operationName, String... tags) {
@@ -225,7 +228,7 @@ public class KeyValueStoreTestDriver<K, V> {
 
 
 
-        this.context = new MockProcessorContext(null, this.stateDir, serdes.keySerde(), serdes.valueSerde(), recordCollector) {
+        this.context = new MockProcessorContext(null, this.stateDir, serdes.keySerde(), serdes.valueSerde(), recordCollector, null) {
             @Override
             public TaskId taskId() {
                 return new TaskId(0, 1);
@@ -266,6 +269,10 @@ public class KeyValueStoreTestDriver<K, V> {
             public Map<String, Object> appConfigsWithPrefix(String prefix) {
                 return new StreamsConfig(props).originalsWithPrefix(prefix);
             }
+            @Override
+            public ThreadCache getCache() {
+                return cache;
+            }
         };
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/86aa0eb0/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java
index 700655e..13f718c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java
@@ -34,7 +34,6 @@ import org.junit.Test;
 public abstract class AbstractKeyValueStoreTest {
 
 
-
     protected abstract <K, V> KeyValueStore<K, V> createKeyValueStore(ProcessorContext context,
                                                                       Class<K> keyClass, Class<V> valueClass,
                                                                       boolean useContextSerdes);


Mime
View raw message