chore: follow udf error format for reduce (#199)

yhl25 · web-flow · commit 0c39eceda9cd · 2025-08-03T22:55:10.000+05:30
diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/accumulator/sorter/StreamSorterFactory.java b/examples/src/main/java/io/numaproj/numaflow/examples/accumulator/sorter/StreamSorterFactory.java
@@ -39,6 +39,7 @@ public static class StreamSorter extends Accumulator {
         private final TreeSet<Datum> sortedBuffer = new TreeSet<>(Comparator
                 .comparing(Datum::getEventTime)
                 .thenComparing(Datum::getID)); // Assuming Datum has a getUniqueId() method
+
         @Override
         public void processMessage(Datum datum, OutputStreamObserver outputStream) {
             log.info("Received datum with event time: {}", datum.toString());
@@ -57,10 +58,10 @@ public void handleEndOfStream(OutputStreamObserver outputStreamObserver) {
 
         private void flushBuffer(OutputStreamObserver outputStream) {
             log.info("Watermark updated, flushing sortedBuffer: {}", latestWm.toEpochMilli());
-            while (!sortedBuffer.isEmpty() && !sortedBuffer
+            while (!sortedBuffer.isEmpty() && sortedBuffer
                     .first()
                     .getEventTime()
-                    .isAfter(latestWm)) {
+                    .isBefore(latestWm)) {
                 Datum datum = sortedBuffer.pollFirst();
                 assert datum != null;
                 outputStream.send(new Message(datum));
diff --git a/src/main/java/io/numaproj/numaflow/accumulator/Service.java b/src/main/java/io/numaproj/numaflow/accumulator/Service.java
@@ -42,8 +42,8 @@ static void handleFailure(
                 // Build gRPC Status
                 com.google.rpc.Status status = com.google.rpc.Status.newBuilder()
                         .setCode(Code.INTERNAL.getNumber())
-                        .setMessage(ExceptionUtils.getExceptionErrorString() + ": " + (
-                                e.getMessage() != null ? e.getMessage() : ""))
+                        .setMessage(
+                                ExceptionUtils.getExceptionErrorString() + ": " + (e.getMessage() != null ? e.getMessage() : ""))
                         .addDetails(Any.pack(DebugInfo.newBuilder()
                                 .setDetail(ExceptionUtils.getStackTrace(e))
                                 .build()))
diff --git a/src/main/java/io/numaproj/numaflow/reducer/Service.java b/src/main/java/io/numaproj/numaflow/reducer/Service.java
@@ -3,13 +3,18 @@
 import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
 import akka.actor.AllDeadLetters;
+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.reduce.v1.ReduceGrpc;
 import io.numaproj.numaflow.reduce.v1.ReduceOuterClass;
 import io.numaproj.numaflow.reducer.metadata.IntervalWindowImpl;
 import io.numaproj.numaflow.reducer.metadata.MetadataImpl;
+import io.numaproj.numaflow.shared.ExceptionUtils;
 import io.numaproj.numaflow.shared.GrpcServerUtils;
 import lombok.extern.slf4j.Slf4j;
 
@@ -37,8 +42,14 @@ static void handleFailure(
                 failureFuture.get();
             } catch (Exception e) {
                 e.printStackTrace();
-                var status = Status.UNKNOWN.withDescription(e.getMessage()).withCause(e);
-                responseObserver.onError(status.asException());
+                com.google.rpc.Status status = com.google.rpc.Status.newBuilder()
+                        .setCode(Code.INTERNAL.getNumber())
+                        .setMessage(ExceptionUtils.getExceptionErrorString() + ": " + (e.getMessage() != null ? e.getMessage() : ""))
+                        .addDetails(Any.pack(DebugInfo.newBuilder()
+                                .setDetail(ExceptionUtils.getStackTrace(e))
+                                .build()))
+                        .build();
+                responseObserver.onError(StatusProto.toStatusRuntimeException(status));
             }
         }).start();
     }
diff --git a/src/main/java/io/numaproj/numaflow/reducestreamer/Service.java b/src/main/java/io/numaproj/numaflow/reducestreamer/Service.java
@@ -3,15 +3,20 @@
 import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
 import akka.actor.AllDeadLetters;
+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.reduce.v1.ReduceGrpc;
 import io.numaproj.numaflow.reduce.v1.ReduceOuterClass;
 import io.numaproj.numaflow.reducestreamer.model.IntervalWindow;
 import io.numaproj.numaflow.reducestreamer.model.Metadata;
 import io.numaproj.numaflow.reducestreamer.model.ReduceStreamer;
 import io.numaproj.numaflow.reducestreamer.model.ReduceStreamerFactory;
+import io.numaproj.numaflow.shared.ExceptionUtils;
 import io.numaproj.numaflow.shared.GrpcServerUtils;
 import lombok.extern.slf4j.Slf4j;
 
@@ -38,8 +43,15 @@ static void handleFailure(
                 failureFuture.get();
             } catch (Exception e) {
                 e.printStackTrace();
-                var status = Status.UNKNOWN.withDescription(e.getMessage()).withCause(e);
-                responseObserver.onError(status.asException());
+                com.google.rpc.Status status = com.google.rpc.Status.newBuilder()
+                        .setCode(Code.INTERNAL.getNumber())
+                        .setMessage(
+                                ExceptionUtils.getExceptionErrorString() + ": " + (e.getMessage() != null ? e.getMessage() : ""))
+                        .addDetails(Any.pack(DebugInfo.newBuilder()
+                                .setDetail(ExceptionUtils.getStackTrace(e))
+                                .build()))
+                        .build();
+                responseObserver.onError(StatusProto.toStatusRuntimeException(status));
             }
         }).start();
     }
diff --git a/src/main/java/io/numaproj/numaflow/sessionreducer/Service.java b/src/main/java/io/numaproj/numaflow/sessionreducer/Service.java
@@ -5,13 +5,17 @@
 import akka.actor.AllDeadLetters;
 import akka.pattern.Patterns;
 import akka.util.Timeout;
+import com.google.protobuf.Any;
 import com.google.protobuf.Empty;
-import io.grpc.Status;
+import com.google.rpc.Code;
+import com.google.rpc.DebugInfo;
+import io.grpc.protobuf.StatusProto;
 import io.grpc.stub.StreamObserver;
 import io.numaproj.numaflow.sessionreduce.v1.SessionReduceGrpc;
 import io.numaproj.numaflow.sessionreduce.v1.Sessionreduce;
 import io.numaproj.numaflow.sessionreducer.model.SessionReducer;
 import io.numaproj.numaflow.sessionreducer.model.SessionReducerFactory;
+import io.numaproj.numaflow.shared.ExceptionUtils;
 import lombok.extern.slf4j.Slf4j;
 import scala.concurrent.Await;
 import scala.concurrent.Future;
@@ -39,8 +43,15 @@ static void handleFailure(
                 failureFuture.get();
             } catch (Exception e) {
                 e.printStackTrace();
-                var status = Status.UNKNOWN.withDescription(e.getMessage()).withCause(e);
-                responseObserver.onError(status.asException());
+                com.google.rpc.Status status = com.google.rpc.Status.newBuilder()
+                        .setCode(Code.INTERNAL.getNumber())
+                        .setMessage(
+                                ExceptionUtils.getExceptionErrorString() + ": " + (e.getMessage() != null ? e.getMessage() : ""))
+                        .addDetails(Any.pack(DebugInfo.newBuilder()
+                                .setDetail(ExceptionUtils.getStackTrace(e))
+                                .build()))
+                        .build();
+                responseObserver.onError(StatusProto.toStatusRuntimeException(status));
             }
         }).start();
     }
diff --git a/src/test/java/io/numaproj/numaflow/reducer/ServerErrTest.java b/src/test/java/io/numaproj/numaflow/reducer/ServerErrTest.java
@@ -27,6 +27,7 @@
 import static io.numaproj.numaflow.shared.GrpcServerUtils.WIN_END_KEY;
 import static io.numaproj.numaflow.shared.GrpcServerUtils.WIN_START_KEY;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 public class ServerErrTest {
@@ -107,9 +108,7 @@ public void given_reducerThrows_when_serverRuns_then_outputStreamContainsThrowab
                 }
             }
             try {
-                assertEquals(
-                        "UNKNOWN: java.lang.RuntimeException: unknown exception",
-                        outputStreamObserver.t.getMessage());
+                assertTrue(outputStreamObserver.t.getMessage().contains("UDF_EXECUTION_ERROR"));
             } catch (Throwable e) {
                 exceptionInThread.set(e);
             }
diff --git a/src/test/java/io/numaproj/numaflow/reducestreamer/ServerErrTest.java b/src/test/java/io/numaproj/numaflow/reducestreamer/ServerErrTest.java
@@ -31,6 +31,7 @@
 import static io.numaproj.numaflow.shared.GrpcServerUtils.WIN_END_KEY;
 import static io.numaproj.numaflow.shared.GrpcServerUtils.WIN_START_KEY;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 public class ServerErrTest {
@@ -107,9 +108,7 @@ public void given_reducerThrows_when_serverRuns_then_outputStreamContainsThrowab
                 }
             }
             try {
-                assertEquals(
-                        "UNKNOWN: java.lang.RuntimeException: unknown exception",
-                        outputStreamObserver.t.getMessage());
+                assertTrue(outputStreamObserver.t.getMessage().contains("UDF_EXECUTION_ERROR"));
             } catch (Throwable e) {
                 exceptionInThread.set(e);
             }
diff --git a/src/test/java/io/numaproj/numaflow/sessionreducer/ServerErrTest.java b/src/test/java/io/numaproj/numaflow/sessionreducer/ServerErrTest.java
@@ -22,6 +22,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 public class ServerErrTest {
@@ -78,9 +79,7 @@ public void given_actorThrows_when_serverRuns_then_outputStreamContainsThrowable
             try {
                 // this test triggers a supervisor runtime exception by sending an OPEN request with 2 windows.
                 // we are expecting the error message below.
-                assertEquals(
-                        "UNKNOWN: java.lang.RuntimeException: open operation error: expected exactly one window",
-                        outputStreamObserver.t.getMessage());
+                assertTrue(outputStreamObserver.t.getMessage().contains("expected exactly one window"));
             } catch (Throwable e) {
                 exceptionInThread.set(e);
             }
@@ -148,9 +147,7 @@ public void given_sessionReducerThrows_when_serverRuns_then_outputStreamContains
                 }
             }
             try {
-                assertEquals(
-                        "UNKNOWN: java.lang.RuntimeException: unknown exception",
-                        outputStreamObserver.t.getMessage());
+                assertTrue(outputStreamObserver.t.getMessage().contains("UDF_EXECUTION_ERROR"));
             } catch (Throwable e) {
                 exceptionInThread.set(e);
             }

Original file line number	Diff line number	Diff line change
`@@ -27,6 +27,7 @@`
`27`	`27`	`import static io.numaproj.numaflow.shared.GrpcServerUtils.WIN_END_KEY;`
`28`	`28`	`import static io.numaproj.numaflow.shared.GrpcServerUtils.WIN_START_KEY;`
`29`	`29`	`import static org.junit.Assert.assertEquals;`
	`30`	`+import static org.junit.Assert.assertTrue;`
`30`	`31`	`import static org.junit.Assert.fail;`
`31`	`32`
`32`	`33`	`public class ServerErrTest {`
`@@ -107,9 +108,7 @@ public void given_reducerThrows_when_serverRuns_then_outputStreamContainsThrowab`
`107`	`108`	`}`
`108`	`109`	`}`
`109`	`110`	`try {`
`110`		`- assertEquals(`
`111`		`- "UNKNOWN: java.lang.RuntimeException: unknown exception",`
`112`		`- outputStreamObserver.t.getMessage());`
	`111`	`+ assertTrue(outputStreamObserver.t.getMessage().contains("UDF_EXECUTION_ERROR"));`
`113`	`112`	`} catch (Throwable e) {`
`114`	`113`	`exceptionInThread.set(e);`
`115`	`114`	`}`
Original file line number	Diff line number	Diff line change
`@@ -31,6 +31,7 @@`
`31`	`31`	`import static io.numaproj.numaflow.shared.GrpcServerUtils.WIN_END_KEY;`
`32`	`32`	`import static io.numaproj.numaflow.shared.GrpcServerUtils.WIN_START_KEY;`
`33`	`33`	`import static org.junit.Assert.assertEquals;`
	`34`	`+import static org.junit.Assert.assertTrue;`
`34`	`35`	`import static org.junit.Assert.fail;`
`35`	`36`
`36`	`37`	`public class ServerErrTest {`
`@@ -107,9 +108,7 @@ public void given_reducerThrows_when_serverRuns_then_outputStreamContainsThrowab`
`107`	`108`	`}`
`108`	`109`	`}`
`109`	`110`	`try {`
`110`		`- assertEquals(`
`111`		`- "UNKNOWN: java.lang.RuntimeException: unknown exception",`
`112`		`- outputStreamObserver.t.getMessage());`
	`111`	`+ assertTrue(outputStreamObserver.t.getMessage().contains("UDF_EXECUTION_ERROR"));`
`113`	`112`	`} catch (Throwable e) {`
`114`	`113`	`exceptionInThread.set(e);`
`115`	`114`	`}`