fix akka for transformer

Signed-off-by: Keran Yang <[email protected]>

examples/pom.xml

@@ -55,6 +55,9 @@
                             <goal>dockerBuild</goal>
                         </goals>
                         <configuration>
+                            <from>
+                                <image>amazoncorretto:11</image>
+                            </from>
                             <container>
                                 <mainClass>
                                     io.numaproj.numaflow.examples.batchmap.flatmap.BatchFlatMap
@@ -74,6 +77,9 @@
                             <goal>dockerBuild</goal>
                         </goals>
                         <configuration>
+                            <from>
+                                <image>amazoncorretto:11</image>
+                            </from>
                             <container>
                                 <mainClass>
                                     io.numaproj.numaflow.examples.sourcetransformer.eventtimefilter.EventTimeFilterFunction
@@ -93,6 +99,9 @@
                             <goal>dockerBuild</goal>
                         </goals>
                         <configuration>
+                            <from>
+                                <image>amazoncorretto:11</image>
+                            </from>
                             <container>
                                 <mainClass>
                                     io.numaproj.numaflow.examples.mapstream.flatmapstream.FlatMapStreamFunction
@@ -110,6 +119,9 @@
                             <goal>dockerBuild</goal>
                         </goals>
                         <configuration>
+                            <from>
+                                <image>amazoncorretto:11</image>
+                            </from>
                             <container>
                                 <mainClass>
                                     io.numaproj.numaflow.examples.map.flatmap.FlatMapFunction
@@ -195,6 +207,9 @@
                             <goal>dockerBuild</goal>
                         </goals>
                         <configuration>
+                            <from>
+                                <image>amazoncorretto:11</image>
+                            </from>
                             <container>
                                 <mainClass>
                                     io.numaproj.numaflow.examples.map.forward.ForwardFunction
@@ -266,6 +281,9 @@
                             <goal>dockerBuild</goal>
                         </goals>
                         <configuration>
+                            <from>
+                                <image>amazoncorretto:11</image>
+                            </from>
                             <container>
                                 <mainClass>
                                     io.numaproj.numaflow.examples.source.simple.SimpleSource

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

@@ -122,9 +122,9 @@ public void start() throws Exception {
      * @throws InterruptedException if the current thread is interrupted while waiting
      */
     public void awaitTermination() throws InterruptedException {
-        log.info("sink server is waiting for termination");
+        log.info("map stream server is waiting for termination");
         server.awaitTermination();
-        log.info("sink server has terminated");
+        log.info("map stream server has terminated");
     }
 
     /**

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

@@ -89,6 +89,9 @@ public void start() throws Exception {
                 Server.this.stop();
                 log.info("gracefully shutting down event loop groups");
                 this.grpcServerHelper.gracefullyShutdownEventLoopGroups();
+                // FIXME - this is a workaround to immediately terminate the JVM process
+                // The correct way to do this is to stop all the actors and wait for them to terminate
+                System.exit(0);
             } catch (InterruptedException e) {
                 Thread.interrupted();
                 e.printStackTrace(System.err);
@@ -108,6 +111,9 @@ public void start() throws Exception {
                     Server.this.stop();
                     log.info("gracefully shutting down event loop groups");
                     this.grpcServerHelper.gracefullyShutdownEventLoopGroups();
+                    // FIXME - this is a workaround to immediately terminate the JVM process
+                    // The correct way to do this is to stop all the actors and wait for them to terminate
+                    System.exit(0);
                 } catch (InterruptedException ex) {
                     Thread.interrupted();
                     ex.printStackTrace(System.err);

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

@@ -47,6 +47,8 @@ class TransformSupervisorActor extends AbstractActor {
     private final SourceTransformer transformer;
     private final StreamObserver<Sourcetransformer.SourceTransformResponse> responseObserver;
     private final CompletableFuture<Void> shutdownSignal;
+    private int activeTransformersCount;
+    private Exception userException;
 
     /**
      * Constructor for TransformSupervisorActor.
@@ -62,6 +64,8 @@ public TransformSupervisorActor(
         this.transformer = transformer;
         this.responseObserver = responseObserver;
         this.shutdownSignal = shutdownSignal;
+        this.userException = null;
+        this.activeTransformersCount = 0;
     }
 
     /**
@@ -92,13 +96,13 @@ public static Props props(
      */
     @Override
     public void preRestart(Throwable reason, Optional<Object> message) {
-        log.debug("supervisor pre restart was executed");
-        shutdownSignal.completeExceptionally(reason);
+        getContext().getSystem().log().warning("supervisor pre restart was executed due to: {}", reason.getMessage());
         responseObserver.onError(Status.INTERNAL
                 .withDescription(reason.getMessage())
                 .withCause(reason)
                 .asException());
         Service.transformerActorSystem.stop(getSelf());
+        shutdownSignal.completeExceptionally(reason);
     }
 
     /**
@@ -133,11 +137,16 @@ public Receive createReceive() {
      */
     private void handleFailure(Exception e) {
         log.error("Encountered error in sourceTransformFn - {}", e.getMessage());
-        shutdownSignal.completeExceptionally(e);
-        responseObserver.onError(Status.INTERNAL
-                .withDescription(e.getMessage())
-                .withCause(e)
-                .asException());
+        if (userException == null) {
+            userException = e;
+            // only send the very first exception to the client
+            // one exception should trigger a container restart
+            responseObserver.onError(Status.INTERNAL
+                    .withDescription(e.getMessage())
+                    .withCause(e)
+                    .asException());
+        }
+        activeTransformersCount--;
     }
 
     /**
@@ -147,6 +156,7 @@ private void handleFailure(Exception e) {
      */
     private void sendResponse(Sourcetransformer.SourceTransformResponse transformResponse) {
         responseObserver.onNext(transformResponse);
+        activeTransformersCount--;
     }
 
     /**
@@ -155,13 +165,24 @@ private void sendResponse(Sourcetransformer.SourceTransformResponse transformRes
      * @param transformRequest The SourceTransformRequest to be processed.
      */
     private void processRequest(Sourcetransformer.SourceTransformRequest transformRequest) {
+        if (userException != null) {
+            log.info("a previous transformer actor failed, not processing any more requests");
+            if (activeTransformersCount == 0) {
+                log.info("there is no more active transformer AKKA actors - stopping the system");
+                getContext().getSystem().stop(getSelf());
+                log.info("AKKA system stopped");
+                shutdownSignal.completeExceptionally(userException);
+            }
+            return;
+        }
         // Create a TransformerActor for each incoming request.
         ActorRef transformerActor = getContext()
                 .actorOf(TransformerActor.props(
                         transformer));
 
         // Send the message to the TransformerActor.
         transformerActor.tell(transformRequest, getSelf());
+        activeTransformersCount++;
     }
 
     /**
@@ -170,10 +191,10 @@ private void processRequest(Sourcetransformer.SourceTransformRequest transformRe
      * @param deadLetter The dead letter to be handled.
      */
     private void handleDeadLetters(AllDeadLetters deadLetter) {
-        log.debug("got a dead letter, stopping the execution");
-        shutdownSignal.completeExceptionally(new Throwable("dead letters"));
-        responseObserver.onError(Status.UNKNOWN.withDescription("dead letters").asException());
+        log.info("got a dead letter, stopping the execution");
+        responseObserver.onError(Status.INTERNAL.withDescription("dead letters").asException());
         getContext().getSystem().stop(getSelf());
+        shutdownSignal.completeExceptionally(new Throwable("dead letters"));
     }
 
     /**
@@ -187,8 +208,7 @@ public SupervisorStrategy supervisorStrategy() {
         return new AllForOneStrategy(
                 DeciderBuilder
                         .match(Exception.class, e -> {
-                            shutdownSignal.completeExceptionally(e);
-                            responseObserver.onError(Status.UNKNOWN
+                            responseObserver.onError(Status.INTERNAL
                                     .withDescription(e.getMessage())
                                     .withCause(e)
                                     .asException());