SAMZA-2025: InputOperatorImpl should work with filtering InputTransformer

InputOperatorImpl should handle the case where InputTransformer returns null record. It makes having simple filtering operation as part of the transformer easy.

Author: Deepthi Sridharan <[email protected]>

Reviewers: atoomula, prateekm

Closes apache#841 from DEEPTHIKORAT/tranformer

Author: DEEPTHIKORAT

samza-core/src/main/java/org/apache/samza/operators/impl/InputOperatorImpl.java

@@ -55,7 +55,10 @@ public Collection<Object> handleMessage(IncomingMessageEnvelope ime, MessageColl
     } else {
       message = this.inputOpSpec.isKeyed() ? KV.of(ime.getKey(), ime.getMessage()) : ime.getMessage();
     }
-    return Collections.singletonList(message);
+    if (message != null) {
+      return Collections.singletonList(message);
+    }
+    return Collections.emptyList();
   }
 
   @Override

samza-core/src/test/java/org/apache/samza/operators/impl/TestInputOperatorImpl.java

@@ -28,6 +28,7 @@
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 
 public class TestInputOperatorImpl {
@@ -77,4 +78,18 @@ public void testWithInputTransformer() {
     Object result = results.iterator().next();
     assertEquals("123", result);
   }
+
+  @Test
+  public void testWithFilteringInputTransformer() {
+    InputOperatorSpec inputOpSpec =
+        new InputOperatorSpec("stream-id", null, null, (ime) -> null, true, "input-op-id");
+    InputOperatorImpl inputOperator = new InputOperatorImpl(inputOpSpec);
+
+    IncomingMessageEnvelope ime =
+        new IncomingMessageEnvelope(mock(SystemStreamPartition.class), "123", "key", "msg");
+
+    Collection<Object> results =
+        inputOperator.handleMessage(ime, mock(MessageCollector.class), mock(TaskCoordinator.class));
+    assertTrue("Transformer doesn't return any record. Expected an empty collection", results.isEmpty());
+  }
 }