[kafka] branch 2.5 updated: KAFKA-9533: ValueTransform forwards `null` values (#8108)

[bbejeck]

This is an automated email from the ASF dual-hosted git repository.

bbejeck pushed a commit to branch 2.5
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/2.5 by this push:
     new 7bf808a  KAFKA-9533: ValueTransform forwards `null` values (#8108)
7bf808a is described below

commit 7bf808ae0d578769e8a9a3c8a08a9e9a77d628dd
Author: Michael Viamari <mviam...@users.noreply.github.com>
AuthorDate: Wed Feb 19 13:20:35 2020 -0800

    KAFKA-9533: ValueTransform forwards `null` values (#8108)
    
    Fixes a bug where KStream#transformValues would forward null values from 
the provided ValueTransform#transform operation.
    
    A test was added for verification 
KStreamTransformValuesTest#shouldEmitNoRecordIfTransformReturnsNull. A parallel 
test for non-key ValueTransformer was not added, as they share the same code 
path.
    
    Reviewers: Bruno Cadonna <br...@confluent.io>, Bill Bejeck 
<bbej...@gmail.com>
---
 .../kstream/internals/KStreamTransformValues.java    |  5 ++++-
 .../internals/KStreamTransformValuesTest.java        | 20 ++++++++++++++++++++
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git 
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValues.java
 
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValues.java
index 843606b..06216fc 100644
--- 
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValues.java
+++ 
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValues.java
@@ -53,7 +53,10 @@ public class KStreamTransformValues<K, V, R> implements 
ProcessorSupplier<K, V>
 
         @Override
         public void process(final K key, final V value) {
-            context.forward(key, valueTransformer.transform(key, value));
+            final R transformedValue = valueTransformer.transform(key, value);
+            if (transformedValue != null) {
+                context.forward(key, transformedValue);
+            }
         }
 
         @Override
diff --git 
a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java
 
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java
index 196f71c..4dc68e0 100644
--- 
a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java
+++ 
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java
@@ -34,6 +34,7 @@ import org.apache.kafka.streams.TestInputTopic;
 import org.apache.kafka.test.MockProcessorSupplier;
 import org.apache.kafka.test.SingletonNoOpValueTransformer;
 import org.apache.kafka.test.StreamsTestUtils;
+import org.easymock.EasyMock;
 import org.easymock.EasyMockRunner;
 import org.easymock.Mock;
 import org.easymock.MockType;
@@ -42,6 +43,7 @@ import org.junit.runner.RunWith;
 
 import java.util.Properties;
 
+import static org.easymock.EasyMock.mock;
 import static org.hamcrest.CoreMatchers.isA;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertArrayEquals;
@@ -138,6 +140,24 @@ public class KStreamTransformValuesTest {
         assertArrayEquals(expected, 
supplier.theCapturedProcessor().processed.toArray());
     }
 
+    @Test
+    public void shouldEmitNoRecordIfTransformReturnsNull() {
+        final ProcessorContext context = mock(ProcessorContext.class);
+        final ValueTransformerWithKey<Integer, Integer, Integer> 
valueTransformer = mock(ValueTransformerWithKey.class);
+        final KStreamTransformValues.KStreamTransformValuesProcessor<Integer, 
Integer, Integer> processor =
+            new 
KStreamTransformValues.KStreamTransformValuesProcessor<>(valueTransformer);
+        processor.init(context);
+
+        final Integer inputKey = 1;
+        final Integer inputValue = 10;
+        EasyMock.expect(valueTransformer.transform(inputKey, 
inputValue)).andStubReturn(null);
+        EasyMock.replay(context);
+
+        processor.process(inputKey, inputValue);
+
+        EasyMock.verify(context);
+    }
+
     @SuppressWarnings("unchecked")
     @Test
     public void 
shouldInitializeTransformerWithForwardDisabledProcessorContext() {