KAFKA-3902

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableFilter.java

@@ -77,6 +77,9 @@ public void process(K key, Change<V> change) {
             V newValue = computeValue(key, change.newValue);
             V oldValue = sendOldValues ? computeValue(key, change.oldValue) : null;
 
+            if (sendOldValues && oldValue == null && newValue == null)
+                return; // unnecessary to forward here.
+
             context().forward(key, new Change<>(newValue, oldValue));
         }
 

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java

@@ -24,6 +24,10 @@
 import org.apache.kafka.streams.kstream.Predicate;
 import org.apache.kafka.test.KStreamTestDriver;
 import org.apache.kafka.test.MockProcessorSupplier;
+import org.apache.kafka.test.MockReducer;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.ValueMapper;
+import org.apache.kafka.streams.kstream.KeyValueMapper;
 import org.apache.kafka.test.TestUtils;
 import org.junit.After;
 import org.junit.Before;
@@ -254,24 +258,68 @@ public boolean test(String key, Integer value) {
         driver.process(topic1, "C", 1);
 
         proc1.checkAndClearProcessResult("A:(1<-null)", "B:(1<-null)", "C:(1<-null)");
-        proc2.checkAndClearProcessResult("A:(null<-null)", "B:(null<-null)", "C:(null<-null)");
+        proc2.checkEmptyAndClearProcessResult(); // we got nothing since all inputs are odd or filtered out
 
         driver.process(topic1, "A", 2);
         driver.process(topic1, "B", 2);
 
         proc1.checkAndClearProcessResult("A:(2<-1)", "B:(2<-1)");
-        proc2.checkAndClearProcessResult("A:(2<-null)", "B:(2<-null)");
+        proc2.checkAndClearProcessResult("A:(2<-null)", "B:(2<-null)"); // we are informed of 2 making it in for both A and B
 
         driver.process(topic1, "A", 3);
 
         proc1.checkAndClearProcessResult("A:(3<-2)");
-        proc2.checkAndClearProcessResult("A:(null<-2)");
+        proc2.checkAndClearProcessResult("A:(null<-2)"); // no change for B but A is deleted
 
         driver.process(topic1, "A", null);
         driver.process(topic1, "B", null);
 
         proc1.checkAndClearProcessResult("A:(null<-3)", "B:(null<-2)");
-        proc2.checkAndClearProcessResult("A:(null<-null)", "B:(null<-2)");
+        proc2.checkAndClearProcessResult("B:(null<-2)");  // B is deleted from source Table1
     }
 
+    @Test
+    public void testSkipNullOnMaterialization() throws IOException {
+        // do not explicitly set enableSendingOldValues, which is considered a cheat. Let a further downstream stateful operator trigger it instead.
+        // This test may well need review/deletion with future materialization changes, should materialization occur more frequently for instance.
+        KStreamBuilder builder = new KStreamBuilder();
+
+        String topic1 = "topic1";
+
+        KTableImpl<String, String, String> table1 =
+            (KTableImpl<String, String, String>) builder.table(stringSerde, stringSerde, topic1);
+        KTableImpl<String, String, String> table2 = (KTableImpl<String, String, String>) table1.filter(
+            new Predicate<String, String>() {
+                @Override
+                public boolean test(String key, String value) {
+                    return value.compareToIgnoreCase("accept") == 0;
+                }
+            }).mapValues(
+                new ValueMapper<String, String>() {
+                    @Override
+                    public String apply(String value) {
+                        return value;
+                    }
+                }).groupBy(new KeyValueMapper<String, String, KeyValue<String, String>>() {
+                    @Override
+                    public KeyValue<String, String> apply(String first, String second) {
+                        return (second == null || first.compareTo(second) <= 0) ?  new KeyValue<>(first, first) : new KeyValue(second, second);
+                    }
+                }).reduce(MockReducer.STRING_ADDER, MockReducer.STRING_REMOVER, "mock-result");
+
+        MockProcessorSupplier<String, String> proc1 = new MockProcessorSupplier<>();
+        MockProcessorSupplier<String, String> proc2 = new MockProcessorSupplier<>();
+
+        builder.addProcessor("proc1", proc1, table1.name);
+        builder.addProcessor("proc2", proc2, table2.name);
+
+        driver = new KStreamTestDriver(builder, stateDir, stringSerde, stringSerde);
+
+        driver.process(topic1, "A", "reject");
+        driver.process(topic1, "B", "reject");
+        driver.process(topic1, "C", "reject");
+
+        proc1.checkAndClearProcessResult("A:(reject<-null)", "B:(reject<-null)", "C:(reject<-null)");
+        proc2.checkEmptyAndClearProcessResult(); // we got nothing since no input matches (though enableSendingOldValues is not set by test on table2 explicitly)
+    }
 }

streams/src/test/java/org/apache/kafka/test/MockProcessorSupplier.java

@@ -82,6 +82,12 @@ public void checkAndClearProcessResult(String... expected) {
         processed.clear();
     }
 
+    public void checkEmptyAndClearProcessResult() {
+
+        assertEquals("the number of outputs:", 0, processed.size());
+        processed.clear();
+    }
+
     public void checkAndClearPunctuateResult(long... expected) {
         assertEquals("the number of outputs:", expected.length, punctuated.size());