Merge branch 'main' into task-execution-id

Signed-off-by: Javier Aliaga <[email protected]>

Author: javier-aliaga

.sdkmanrc

@@ -0,0 +1,3 @@
+# Enable auto-env through the sdkman_auto_env config
+# Add key=value pairs of SDKs to use below
+java=11.0.27-tem

client/build.gradle

@@ -11,7 +11,7 @@ plugins {
 }
 
 group 'io.dapr'
-version = '1.5.6'
+version = '1.5.7'
 archivesBaseName = 'durabletask-client'
 
 def grpcVersion = '1.69.0'

client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java

@@ -20,7 +20,8 @@
 import java.util.logging.Logger;
 
 /**
- * Task hub worker that connects to a sidecar process over gRPC to execute orchestrator and activity events.
+ * Task hub worker that connects to a sidecar process over gRPC to execute
+ * orchestrator and activity events.
  */
 public final class DurableTaskGrpcWorker implements AutoCloseable {
 
@@ -39,8 +40,8 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
     private final TaskHubSidecarServiceBlockingStub sidecarClient;
     private final boolean isExecutorServiceManaged;
     private volatile boolean isNormalShutdown = false;
-    private Thread processorThread;
-    
+    private Thread workerThread;
+
     DurableTaskGrpcWorker(DurableTaskGrpcWorkerBuilder builder) {
         this.orchestrationFactories.putAll(builder.orchestrationFactories);
         this.activityFactories.putAll(builder.activityFactories);
@@ -67,46 +68,60 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
 
         this.sidecarClient = TaskHubSidecarServiceGrpc.newBlockingStub(sidecarGrpcChannel);
         this.dataConverter = builder.dataConverter != null ? builder.dataConverter : new JacksonDataConverter();
-        this.maximumTimerInterval = builder.maximumTimerInterval != null ? builder.maximumTimerInterval : DEFAULT_MAXIMUM_TIMER_INTERVAL;
+        this.maximumTimerInterval = builder.maximumTimerInterval != null ? builder.maximumTimerInterval
+                : DEFAULT_MAXIMUM_TIMER_INTERVAL;
         this.workerPool = builder.executorService != null ? builder.executorService : Executors.newCachedThreadPool();
         this.isExecutorServiceManaged = builder.executorService == null;
     }
 
     /**
-     * Establishes a gRPC connection to the sidecar and starts processing work-items in the background.
+     * Establishes a gRPC connection to the sidecar and starts processing work-items
+     * in the background.
      * <p>
-     * This method retries continuously to establish a connection to the sidecar. If a connection fails,
-     * a warning log message will be written and a new connection attempt will be made. This process
-     * continues until either a connection succeeds or the process receives an interrupt signal.
+     * This method retries continuously to establish a connection to the sidecar. If
+     * a connection fails,
+     * a warning log message will be written and a new connection attempt will be
+     * made. This process
+     * continues until either a connection succeeds or the process receives an
+     * interrupt signal.
      */
     public void start() {
-        this.processorThread = new Thread(this::startAndBlock);
-        this.processorThread.start();
+        this.workerThread = new Thread(this::startAndBlock);
+        this.workerThread.start();
     }
 
 
     /**
-     * Closes the internally managed gRPC channel and executor service, if one exists.
+     * Closes the internally managed gRPC channel and executor service, if one
+     * exists.
      * <p>
-     * Only the internally managed GRPC Channel and Executor services are closed. If any of them are supplied,
+     * Only the internally managed GRPC Channel and Executor services are closed. If
+     * any of them are supplied,
      * it is the responsibility of the supplier to take care of them.
      */
     public void close() {
+        this.workerThread.interrupt();
         this.isNormalShutdown = true;
         this.processorThread.interrupt();
         this.shutDownWorkerPool();
         this.closeSideCarChannel();
     }
 
     /**
-     * Establishes a gRPC connection to the sidecar and starts processing work-items on the current thread.
-     * This method call blocks indefinitely, or until the current thread is interrupted.
+     * Establishes a gRPC connection to the sidecar and starts processing work-items
+     * on the current thread.
+     * This method call blocks indefinitely, or until the current thread is
+     * interrupted.
      * <p>
-     * Use can alternatively use the {@link #start} method to run orchestration processing in a background thread.
+     * Use can alternatively use the {@link #start} method to run orchestration
+     * processing in a background thread.
      * <p>
-     * This method retries continuously to establish a connection to the sidecar. If a connection fails,
-     * a warning log message will be written and a new connection attempt will be made. This process
-     * continues until either a connection succeeds or the process receives an interrupt signal.
+     * This method retries continuously to establish a connection to the sidecar. If
+     * a connection fails,
+     * a warning log message will be written and a new connection attempt will be
+     * made. This process
+     * continues until either a connection succeeds or the process receives an
+     * interrupt signal.
      */
     public void startAndBlock() {
         logger.log(Level.INFO, "Durable Task worker is connecting to sidecar at {0}.", this.getSidecarAddress());
@@ -121,8 +136,7 @@ public void startAndBlock() {
                 this.dataConverter,
                 logger);
 
-        // TODO: How do we interrupt manually?
-        while (true && !this.isNormalShutdown) {
+        while (true) {
             try {
                 GetWorkItemsRequest getWorkItemsRequest = GetWorkItemsRequest.newBuilder().build();
                 Iterator<WorkItem> workItemStream = this.sidecarClient.getWorkItems(getWorkItemsRequest);
@@ -149,11 +163,17 @@ public void startAndBlock() {
                                 this.sidecarClient.completeOrchestratorTask(response);
                             } catch (StatusRuntimeException e) {
                                 if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) {
-                                    logger.log(Level.WARNING, "The sidecar at address {0} is unavailable while completing the orchestrator task.", this.getSidecarAddress());
+                                    logger.log(Level.WARNING,
+                                            "The sidecar at address {0} is unavailable while completing the orchestrator task.",
+                                            this.getSidecarAddress());
                                 } else if (e.getStatus().getCode() == Status.Code.CANCELLED) {
-                                    logger.log(Level.WARNING, "Durable Task worker has disconnected from {0} while completing the orchestrator task.", this.getSidecarAddress());
+                                    logger.log(Level.WARNING,
+                                            "Durable Task worker has disconnected from {0} while completing the orchestrator task.",
+                                            this.getSidecarAddress());
                                 } else {
-                                    logger.log(Level.WARNING, "Unexpected failure completing the orchestrator task at {0}.", this.getSidecarAddress());
+                                    logger.log(Level.WARNING,
+                                            "Unexpected failure completing the orchestrator task at {0}.",
+                                            this.getSidecarAddress());
                                 }
                             }
                         });
@@ -193,29 +213,35 @@ public void startAndBlock() {
                                 this.sidecarClient.completeActivityTask(responseBuilder.build());
                             } catch (StatusRuntimeException e) {
                                 if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) {
-                                    logger.log(Level.WARNING, "The sidecar at address {0} is unavailable while completing the activity task.", this.getSidecarAddress());
+                                    logger.log(Level.WARNING,
+                                            "The sidecar at address {0} is unavailable while completing the activity task.",
+                                            this.getSidecarAddress());
                                 } else if (e.getStatus().getCode() == Status.Code.CANCELLED) {
-                                    logger.log(Level.WARNING, "Durable Task worker has disconnected from {0} while completing the activity task.", this.getSidecarAddress());
+                                    logger.log(Level.WARNING,
+                                            "Durable Task worker has disconnected from {0} while completing the activity task.",
+                                            this.getSidecarAddress());
                                 } else {
-                                    logger.log(Level.WARNING, "Unexpected failure completing the activity task at {0}.", this.getSidecarAddress());
+                                    logger.log(Level.WARNING, "Unexpected failure completing the activity task at {0}.",
+                                            this.getSidecarAddress());
                                 }
                             }
                         });
-                    } 
-                    else if (requestType == RequestCase.HEALTHPING)
-                    {
+                    } else if (requestType == RequestCase.HEALTHPING) {
                         // No-op
                     } else {
-                        logger.log(Level.WARNING, "Received and dropped an unknown '{0}' work-item from the sidecar.", requestType);
+                        logger.log(Level.WARNING, "Received and dropped an unknown '{0}' work-item from the sidecar.",
+                                requestType);
                     }
                 }
             } catch (StatusRuntimeException e) {
                 if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) {
-                    logger.log(Level.INFO, "The sidecar at address {0} is unavailable. Will continue retrying.", this.getSidecarAddress());
+                    logger.log(Level.INFO, "The sidecar at address {0} is unavailable. Will continue retrying.",
+                            this.getSidecarAddress());
                 } else if (e.getStatus().getCode() == Status.Code.CANCELLED) {
                     logger.log(Level.INFO, "Durable Task worker has disconnected from {0}.", this.getSidecarAddress());
                 } else {
-                    logger.log(Level.WARNING, String.format("Unexpected failure connecting to %s", this.getSidecarAddress()), e);
+                    logger.log(Level.WARNING,
+                            String.format("Unexpected failure connecting to %s", this.getSidecarAddress()), e);
                 }
 
                 // Retry after 5 seconds
@@ -229,7 +255,8 @@ else if (requestType == RequestCase.HEALTHPING)
     }
 
     /**
-     * Stops the current worker's listen loop, preventing any new orchestrator or activity events from being processed.
+     * Stops the current worker's listen loop, preventing any new orchestrator or
+     * activity events from being processed.
      */
     public void stop() {
         this.close();
@@ -250,7 +277,8 @@ private void closeSideCarChannel() {
     private void shutDownWorkerPool() {
         if (this.isExecutorServiceManaged) {
             if (!this.isNormalShutdown) {
-                logger.log(Level.WARNING, "ExecutorService shutdown initiated unexpectedly. No new tasks will be accepted");
+                logger.log(Level.WARNING,
+                        "ExecutorService shutdown initiated unexpectedly. No new tasks will be accepted");
             }
 
             this.workerPool.shutdown();

client/src/main/java/io/dapr/durabletask/TaskOrchestrationExecutor.java

@@ -1183,19 +1183,25 @@ private boolean shouldRetry() {
                 boolean shouldRetryBasedOnPolicy = this.policy != null ? this.shouldRetryBasedOnPolicy() : true;
                 boolean shouldRetryBasedOnHandler = this.handler != null ? this.handler.handle(retryContext) : true;
 
-                if (this.policy != null) {
-                    logger.info(() -> String.format("shouldRetryBasedOnPolicy: %s", shouldRetryBasedOnPolicy));
-                }
+                // Only log when not replaying, so only the current attempt is logged and not all previous attempts.
+                if(!this.context.getIsReplaying()) {
+                    if (this.policy != null) {
+                        logger.fine(() -> String.format("shouldRetryBasedOnPolicy: %s", shouldRetryBasedOnPolicy));
+                    }
 
-                if (this.handler != null) {
-                    logger.info(() -> String.format("shouldRetryBasedOnHandler: %s",  shouldRetryBasedOnHandler));
+                    if (this.handler != null) {
+                        logger.fine(() -> String.format("shouldRetryBasedOnHandler: %s",  shouldRetryBasedOnHandler));
+                    }
                 }
 
                 return shouldRetryBasedOnPolicy && shouldRetryBasedOnHandler;
             }
 
             private boolean shouldRetryBasedOnPolicy() {
-                logger.warning(() -> String.format("Retry Policy: %d retries out of total %d performed ", this.attemptNumber, this.policy.getMaxNumberOfAttempts()));
+                // Only log when not replaying, so only the current attempt is logged and not all previous attempts.
+                if(!this.context.getIsReplaying()) {
+                    logger.fine(() -> String.format("Retry Policy: %d retries out of total %d performed ", this.attemptNumber, this.policy.getMaxNumberOfAttempts()));
+                }
 
                 if (this.attemptNumber >= this.policy.getMaxNumberOfAttempts()) {
                     // Max number of attempts exceeded