Fix retryer in activity and workflow worker (#250)

* Do not retry certain errors on respond activity/workflow completion and failure

* Rebase, bump up version and update changelog

Author: meiliang86

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## v2.3.0
+- Added cron schedule support.
+- Fix infinite retryer in activity and workflow worker due to non-retryable error.
+- Fixed hanging on testEnv.close when testEnv was not started.
+- Fix for NPE when method has base type return type like int.
+- Fixed JsonDataConverter to correctly report non serializable exceptions.
+
 ## v2.2.0
 - Added support for workflow and activity server side retries.
 - Clean worker shutdown. Replaced Worker shutdown(Duration) with Worker shutdown, shutdownNow and awaitTermination.

build.gradle

@@ -37,7 +37,7 @@ googleJavaFormat {
 }
 
 group = 'com.uber.cadence'
-version = '2.2.0'
+version = '2.3.0'
 
 description = """Uber Cadence Java Client"""
 

src/main/java/com/uber/cadence/common/RetryOptions.java

@@ -83,6 +83,31 @@ public RetryOptions merge(RetryOptions o) {
         .validateBuildWithDefaults();
   }
 
+  @SafeVarargs
+  public final RetryOptions addDoNotRetry(Class<? extends Throwable>... doNotRetry) {
+    if (doNotRetry == null) {
+      return this;
+    }
+
+    double backoffCoefficient = getBackoffCoefficient();
+    if (backoffCoefficient == 0) {
+      backoffCoefficient = DEFAULT_BACKOFF_COEFFICIENT;
+    }
+
+    RetryOptions.Builder builder =
+        new RetryOptions.Builder()
+            .setInitialInterval(getInitialInterval())
+            .setExpiration(getExpiration())
+            .setMaximumInterval(getMaximumInterval())
+            .setBackoffCoefficient(backoffCoefficient)
+            .setDoNotRetry(merge(getDoNotRetry(), Arrays.asList(doNotRetry)));
+
+    if (getMaximumAttempts() > 0) {
+      builder.setMaximumAttempts(getMaximumAttempts());
+    }
+    return builder.validateBuildWithDefaults();
+  }
+
   public static final class Builder {
 
     private Duration initialInterval;

src/main/java/com/uber/cadence/internal/worker/ActivityWorker.java

@@ -17,11 +17,7 @@
 
 package com.uber.cadence.internal.worker;
 
-import com.uber.cadence.PollForActivityTaskResponse;
-import com.uber.cadence.RespondActivityTaskCanceledRequest;
-import com.uber.cadence.RespondActivityTaskCompletedRequest;
-import com.uber.cadence.RespondActivityTaskFailedRequest;
-import com.uber.cadence.WorkflowExecution;
+import com.uber.cadence.*;
 import com.uber.cadence.common.RetryOptions;
 import com.uber.cadence.internal.common.Retryer;
 import com.uber.cadence.internal.logging.LoggerTag;
@@ -224,15 +220,27 @@ private void sendReply(PollForActivityTaskResponse task, ActivityTaskHandler.Res
       RetryOptions ro = response.getRequestRetryOptions();
       RespondActivityTaskCompletedRequest taskCompleted = response.getTaskCompleted();
       if (taskCompleted != null) {
-        ro = options.getReportCompletionRetryOptions().merge(ro);
+        ro =
+            options
+                .getReportCompletionRetryOptions()
+                .merge(ro)
+                .addDoNotRetry(
+                    BadRequestError.class, EntityNotExistsError.class, DomainNotActiveError.class);
         taskCompleted.setTaskToken(task.getTaskToken());
         taskCompleted.setIdentity(options.getIdentity());
         Retryer.retry(ro, () -> service.RespondActivityTaskCompleted(taskCompleted));
         options.getMetricsScope().counter(MetricsType.ACTIVITY_TASK_COMPLETED_COUNTER).inc(1);
       } else {
         RespondActivityTaskFailedRequest taskFailed = response.getTaskFailed();
         if (taskFailed != null) {
-          ro = options.getReportFailureRetryOptions().merge(ro);
+          ro =
+              options
+                  .getReportFailureRetryOptions()
+                  .merge(ro)
+                  .addDoNotRetry(
+                      BadRequestError.class,
+                      EntityNotExistsError.class,
+                      DomainNotActiveError.class);
           taskFailed.setTaskToken(task.getTaskToken());
           taskFailed.setIdentity(options.getIdentity());
           Retryer.retry(ro, () -> service.RespondActivityTaskFailed(taskFailed));
@@ -242,7 +250,14 @@ private void sendReply(PollForActivityTaskResponse task, ActivityTaskHandler.Res
           if (taskCancelled != null) {
             taskCancelled.setTaskToken(task.getTaskToken());
             taskCancelled.setIdentity(options.getIdentity());
-            ro = options.getReportFailureRetryOptions().merge(ro);
+            ro =
+                options
+                    .getReportFailureRetryOptions()
+                    .merge(ro)
+                    .addDoNotRetry(
+                        BadRequestError.class,
+                        EntityNotExistsError.class,
+                        DomainNotActiveError.class);
             Retryer.retry(ro, () -> service.RespondActivityTaskCanceled(taskCancelled));
             options.getMetricsScope().counter(MetricsType.ACTIVITY_TASK_CANCELED_COUNTER).inc(1);
           }

src/main/java/com/uber/cadence/internal/worker/WorkflowWorker.java

@@ -270,14 +270,26 @@ private void sendReply(
       RetryOptions ro = response.getRequestRetryOptions();
       RespondDecisionTaskCompletedRequest taskCompleted = response.getTaskCompleted();
       if (taskCompleted != null) {
-        ro = options.getReportCompletionRetryOptions().merge(ro);
+        ro =
+            options
+                .getReportCompletionRetryOptions()
+                .merge(ro)
+                .addDoNotRetry(
+                    BadRequestError.class, EntityNotExistsError.class, DomainNotActiveError.class);
         taskCompleted.setIdentity(options.getIdentity());
         taskCompleted.setTaskToken(taskToken);
         Retryer.retry(ro, () -> service.RespondDecisionTaskCompleted(taskCompleted));
       } else {
         RespondDecisionTaskFailedRequest taskFailed = response.getTaskFailed();
         if (taskFailed != null) {
-          ro = options.getReportFailureRetryOptions().merge(ro);
+          ro =
+              options
+                  .getReportFailureRetryOptions()
+                  .merge(ro)
+                  .addDoNotRetry(
+                      BadRequestError.class,
+                      EntityNotExistsError.class,
+                      DomainNotActiveError.class);
           taskFailed.setIdentity(options.getIdentity());
           taskFailed.setTaskToken(taskToken);
           Retryer.retry(ro, () -> service.RespondDecisionTaskFailed(taskFailed));

src/test/java/com/uber/cadence/workflow/WorkflowTest.java

@@ -4075,6 +4075,43 @@ public void testLargeHistory() {
     assertEquals("done", result);
   }
 
+  public interface DecisionTimeoutWorkflow {
+    @WorkflowMethod(executionStartToCloseTimeoutSeconds = 10000)
+    String execute(String testName) throws InterruptedException;
+  }
+
+  public static class DecisionTimeoutWorkflowImpl implements DecisionTimeoutWorkflow {
+
+    @Override
+    public String execute(String testName) throws InterruptedException {
+
+      AtomicInteger count = retryCount.get(testName);
+      if (count == null) {
+        count = new AtomicInteger();
+        retryCount.put(testName, count);
+        Thread.sleep(2000);
+      }
+
+      return "some result";
+    }
+  }
+
+  @Test
+  public void testDecisionTimeoutWorkflow() throws InterruptedException {
+    startWorkerFor(DecisionTimeoutWorkflowImpl.class);
+
+    WorkflowOptions options =
+        new WorkflowOptions.Builder()
+            .setTaskList(taskList)
+            .setTaskStartToCloseTimeout(Duration.ofSeconds(1))
+            .build();
+
+    DecisionTimeoutWorkflow stub =
+        workflowClient.newWorkflowStub(DecisionTimeoutWorkflow.class, options);
+    String result = stub.execute(testName.getMethodName());
+    Assert.assertEquals("some result", result);
+  }
+
   private static class FilteredTrace {
 
     private final List<String> impl = Collections.synchronizedList(new ArrayList<>());