handle error case for source exception

Signed-off-by: adarsh0728 <[email protected]>

src/main/java/io/numaproj/numaflow/shared/ExceptionUtils.java

@@ -0,0 +1,24 @@
+package io.numaproj.numaflow.shared;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+public class ExceptionUtils {
+    /**
+     * Formalized exception error strings
+     */
+    public static final String ERR_SOURCE_EXCEPTION = "UDF_EXECUTION_ERROR(source)";
+    public static final String ERR_TRANSFORMER_EXCEPTION = "UDF_EXECUTION_ERROR(transformer)";
+
+    /**
+     * Converts the stack trace of an exception into a String.
+     *
+     * @param e the exception to extract the stack trace from
+     * @return the stack trace as a String
+     */
+    public static String getStackTrace(Exception e) {
+        StringWriter sw = new StringWriter();
+        e.printStackTrace(new PrintWriter(sw));
+        return sw.toString();
+    }
+}

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

@@ -1,8 +1,13 @@
 package io.numaproj.numaflow.sourcer;
 
+import com.google.protobuf.Any;
 import com.google.protobuf.Empty;
+import com.google.rpc.Code;
+import com.google.rpc.DebugInfo;
 import io.grpc.Status;
+import io.grpc.protobuf.StatusProto;
 import io.grpc.stub.StreamObserver;
+import io.numaproj.numaflow.shared.ExceptionUtils;
 import io.numaproj.numaflow.source.v1.SourceGrpc;
 import io.numaproj.numaflow.source.v1.SourceOuterClass;
 import lombok.AllArgsConstructor;
@@ -15,7 +20,6 @@
 
 import static io.numaproj.numaflow.source.v1.SourceGrpc.getPendingFnMethod;
 
-
 /**
  * Implementation of the gRPC service for the sourcer.
  */
@@ -31,7 +35,8 @@ class Service extends SourceGrpc.SourceImplBase {
      * @param responseObserver the response observer
      */
     @Override
-    public StreamObserver<SourceOuterClass.ReadRequest> readFn(final StreamObserver<SourceOuterClass.ReadResponse> responseObserver) {
+    public StreamObserver<SourceOuterClass.ReadRequest> readFn(
+            final StreamObserver<SourceOuterClass.ReadResponse> responseObserver) {
         return new StreamObserver<>() {
             private boolean handshakeDone = false;
 
@@ -77,12 +82,18 @@ public void onNext(SourceOuterClass.ReadRequest request) {
 
                     responseObserver.onNext(response);
                 } catch (Exception e) {
-                    log.error("Encountered error in readFn onNext - {}", e.getMessage());
+                    String stackTrace = ExceptionUtils.getStackTrace(e);
+                    log.error("Encountered error in readFn onNext - {} {}", e.getMessage(), stackTrace);
                     shutdownSignal.completeExceptionally(e);
-                    responseObserver.onError(Status.INTERNAL
-                            .withDescription(e.getMessage())
-                            .withCause(e)
-                            .asException());
+                    // Build gRPC Status [error]
+                    com.google.rpc.Status status = com.google.rpc.Status.newBuilder()
+                            .setCode(Code.INTERNAL.getNumber())
+                            .setMessage(ExceptionUtils.ERR_SOURCE_EXCEPTION + ": " + (e.getMessage() != null ? e.getMessage() : ""))
+                            .addDetails(Any.pack(DebugInfo.newBuilder()
+                                    .setDetail(stackTrace)
+                                    .build()))
+                            .build();
+                    responseObserver.onError(StatusProto.toStatusRuntimeException(status));
                 }
             }
 
@@ -200,7 +211,8 @@ public void pendingFn(
                 SourceOuterClass.PendingResponse.Result
                         .newBuilder()
                         .setCount(this.sourcer.getPending())
-                        .build()).build());
+                        .build())
+                .build());
         responseObserver.onCompleted();
     }
 
@@ -235,8 +247,8 @@ public void partitionsFn(
         responseObserver.onNext(SourceOuterClass.PartitionsResponse.newBuilder()
                 .setResult(
                         SourceOuterClass.PartitionsResponse.Result.newBuilder()
-                                .addAllPartitions(partitions)).
-                build());
+                                .addAllPartitions(partitions))
+                .build());
         responseObserver.onCompleted();
     }
 }

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

@@ -14,11 +14,10 @@
 import com.google.rpc.Code;
 import com.google.rpc.DebugInfo;
 import io.grpc.protobuf.StatusProto;
+import io.numaproj.numaflow.shared.ExceptionUtils;
 import io.numaproj.numaflow.sourcetransformer.v1.Sourcetransformer;
 import lombok.extern.slf4j.Slf4j;
 
-import java.io.PrintWriter;
-import java.io.StringWriter;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
@@ -55,7 +54,6 @@ class TransformSupervisorActor extends AbstractActor {
     private final CompletableFuture<Void> shutdownSignal;
     private int activeTransformersCount;
     private Exception userException;
-    private static final String ERR_TRANSFORMER_EXCEPTION = "UDF_EXECUTION_ERROR(transformer)";
 
     /**
      * Constructor for TransformSupervisorActor.
@@ -143,7 +141,7 @@ public Receive createReceive() {
      * @param e The exception to be handled.
      */
     private void handleFailure(Exception e) {
-        String stackTrace = getStackTrace(e);
+        String stackTrace = ExceptionUtils.getStackTrace(e);
         log.error("Exception in sourceTransformFn: {} {}", e.getMessage(), stackTrace);
         if (userException == null) {
             userException = e;
@@ -153,7 +151,7 @@ private void handleFailure(Exception e) {
             // Build gRPC Status [error]
             com.google.rpc.Status status = com.google.rpc.Status.newBuilder()
                     .setCode(Code.INTERNAL.getNumber())
-                    .setMessage(ERR_TRANSFORMER_EXCEPTION + ": " + (e.getMessage() != null ? e.getMessage() : ""))
+                    .setMessage(ExceptionUtils.ERR_TRANSFORMER_EXCEPTION + ": " + (e.getMessage() != null ? e.getMessage() : ""))
                     .addDetails(Any.pack(DebugInfo.newBuilder()
                             .setDetail(stackTrace)
                             .build()))
@@ -163,18 +161,6 @@ private void handleFailure(Exception e) {
         activeTransformersCount--;
     }
 
-    /**
-     * Converts the stack trace of an exception into a String.
-     *
-     * @param e the exception to extract the stack trace from
-     * @return the stack trace as a String
-     */
-    private String getStackTrace(Exception e) {
-        StringWriter sw = new StringWriter();
-        e.printStackTrace(new PrintWriter(sw));
-        return sw.toString();
-    }
-
     /**
      * Sends the SourceTransformResponse back to the client.
      *