chore: add checks to avoid panics if the users return null

src/main/java/io/numaproj/numaflow/batchmapper/Service.java

@@ -138,29 +138,33 @@ public void onCompleted() {
     private void buildAndStreamResponse(
             BatchResponses responses,
             StreamObserver<MapOuterClass.MapResponse> responseObserver) {
-        responses.getItems().forEach(message -> {
-            List<MapOuterClass.MapResponse.Result> mapResponseResult = new ArrayList<>();
-            message.getItems().forEach(res -> {
-                mapResponseResult.add(
-                        MapOuterClass.MapResponse.Result
-                                .newBuilder()
-                                .setValue(res.getValue()
-                                        == null ? ByteString.EMPTY : ByteString.copyFrom(
-                                        res.getValue()))
-                                .addAllKeys(res.getKeys()
-                                        == null ? new ArrayList<>() : List.of(res.getKeys()))
-                                .addAllTags(res.getTags()
-                                        == null ? new ArrayList<>() : List.of(res.getTags()))
-                                .build()
-                );
+
+        if (responses != null) {
+            responses.getItems().forEach(message -> {
+                List<MapOuterClass.MapResponse.Result> mapResponseResult = new ArrayList<>();
+                message.getItems().forEach(res -> {
+                    mapResponseResult.add(
+                            MapOuterClass.MapResponse.Result
+                                    .newBuilder()
+                                    .setValue(res.getValue()
+                                            == null ? ByteString.EMPTY : ByteString.copyFrom(
+                                            res.getValue()))
+                                    .addAllKeys(res.getKeys()
+                                            == null ? new ArrayList<>() : List.of(res.getKeys()))
+                                    .addAllTags(res.getTags()
+                                            == null ? new ArrayList<>() : List.of(res.getTags()))
+                                    .build()
+                    );
+                });
+                MapOuterClass.MapResponse singleRequestResponse = MapOuterClass.MapResponse
+                        .newBuilder()
+                        .setId(message.getId())
+                        .addAllResults(mapResponseResult)
+                        .build();
+                responseObserver.onNext(singleRequestResponse);
             });
-            MapOuterClass.MapResponse singleRequestResponse = MapOuterClass.MapResponse
-                    .newBuilder()
-                    .setId(message.getId())
-                    .addAllResults(mapResponseResult)
-                    .build();
-            responseObserver.onNext(singleRequestResponse);
-        });
+        }
+
         // Send an EOT message to indicate the end of the transmission for the batch.
         MapOuterClass.MapResponse eotResponse = MapOuterClass.MapResponse
                 .newBuilder()

src/main/java/io/numaproj/numaflow/mapper/MapperActor.java

@@ -5,6 +5,7 @@
 import akka.japi.pf.ReceiveBuilder;
 import com.google.protobuf.ByteString;
 import io.numaproj.numaflow.map.v1.MapOuterClass;
+import io.numaproj.numaflow.sourcetransformer.v1.Sourcetransformer;
 
 import java.time.Instant;
 import java.util.ArrayList;
@@ -68,6 +69,14 @@
      * @return map response
      */
     private MapOuterClass.MapResponse buildResponse(MessageList messageList, String ID) {
+        // Users should not send null as the response, we will let client handle it.
+        if (messageList == null) {
+            return MapOuterClass.MapResponse
+                    .newBuilder()
+                    .setId(ID)
+                    .build();
+        }
+
         MapOuterClass.MapResponse.Builder responseBuilder = MapOuterClass
                 .MapResponse
                 .newBuilder();

src/main/java/io/numaproj/numaflow/mapstreamer/OutputObserverImpl.java

@@ -26,6 +26,9 @@
 
     @Override
     public void send(Message message) {
+        if (message == null) {
+            return;
+        }
         MapOuterClass.MapResponse response = MapOuterClass.MapResponse.newBuilder()
                 .setId(requestID)
                 .addResults(MapOuterClass.MapResponse.Result.newBuilder()

src/main/java/io/numaproj/numaflow/sinker/Service.java

@@ -79,13 +79,15 @@ public void onNext(SinkOuterClass.SinkRequest request) {
                         // End of transmission, write EOF datum to the stream
                         // Wait for the result and send the response back to the client
                         datumStream.writeMessage(HandlerDatum.EOF_DATUM);
-
                         ResponseList responses = result.join();
-                        SinkOuterClass.SinkResponse.Builder responseBuilder = SinkOuterClass.SinkResponse.newBuilder();
-                        for (Response response : responses.getResponses()) {
-                            responseBuilder.addResults(buildResult(response));
+
+                        if (responses != null) {
+                            SinkOuterClass.SinkResponse.Builder responseBuilder = SinkOuterClass.SinkResponse.newBuilder();
+                            for (Response response : responses.getResponses()) {
+                                responseBuilder.addResults(buildResult(response));
+                            }
+                            responseObserver.onNext(responseBuilder.build());
                         }
-                        responseObserver.onNext(responseBuilder.build());
 
                         // send eot response to indicate end of transmission for the batch
                         SinkOuterClass.SinkResponse eotResponse = SinkOuterClass.SinkResponse

src/main/java/io/numaproj/numaflow/sourcetransformer/Server.java

@@ -93,7 +93,7 @@
         // if there are any exceptions, shutdown the server gracefully.
         this.shutdownSignal.whenCompleteAsync((v, e) -> {
             if (e != null) {
-                System.err.println("*** shutting down sink gRPC server because of an exception - " + e.getMessage());
+                System.err.println("*** shutting down transformer gRPC server because of an exception - " + e.getMessage());
                 try {
                     this.stop();
                     // FIXME - this is a workaround to immediately terminate the JVM process

src/main/java/io/numaproj/numaflow/sourcetransformer/TransformerActor.java

@@ -93,6 +93,13 @@
     private Sourcetransformer.SourceTransformResponse buildResponse(
             MessageList messageList,
             String ID) {
+        // Users should not send null as the response, we will let client handle it.
+        if (messageList == null) {
+            return Sourcetransformer.SourceTransformResponse
+                    .newBuilder()
+                    .setId(ID)
+                    .build();
+        }
         Sourcetransformer.SourceTransformResponse.Builder responseBuilder = Sourcetransformer
                 .SourceTransformResponse
                 .newBuilder();