Only call RespondDecisionTaskFailure on the first attempt to avoid spinning (#115)

Author: mfateev

src/main/java/com/uber/cadence/internal/replay/ReplayDecisionTaskHandler.java

@@ -48,11 +48,19 @@ public ReplayDecisionTaskHandler(String domain, ReplayWorkflowFactory asyncWorkf
     }
 
     @Override
-    public DecisionTaskHandler.Result handleDecisionTask(DecisionTaskWithHistoryIterator decisionTaskIterator) {
+    public DecisionTaskHandler.Result handleDecisionTask(DecisionTaskWithHistoryIterator decisionTaskIterator) throws Exception {
         try {
             return handleDecisionTaskImpl(decisionTaskIterator);
         } catch (Throwable e) {
             PollForDecisionTaskResponse decisionTask = decisionTaskIterator.getDecisionTask();
+            // Only fail decision on first attempt, subsequent failure on the same decision task will timeout.
+            // This is to avoid spin on the failed decision task.
+            if (decisionTask.getAttempt() > 0) {
+                if (e instanceof Error) {
+                    throw (Error) e;
+                }
+                throw (Exception) e;
+            }
             if (log.isErrorEnabled()) {
                 WorkflowExecution execution = decisionTask.getWorkflowExecution();
                 log.error("Workflow task failure. startedEventId=" + decisionTask.getStartedEventId()

src/main/java/com/uber/cadence/internal/sync/SyncWorkflowWorker.java

@@ -84,7 +84,7 @@ public void resumePolling() {
         worker.resumePolling();
     }
 
-    public <R> R queryWorkflowExecution(WorkflowExecution execution, String queryType, Class<R> returnType, Object[] args) {
+    public <R> R queryWorkflowExecution(WorkflowExecution execution, String queryType, Class<R> returnType, Object[] args) throws Exception {
         DataConverter dataConverter = options.getDataConverter();
         byte[] serializedArgs = dataConverter.toData(args);
         byte[] result = worker.queryWorkflowExecution(execution, queryType, serializedArgs);

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

@@ -85,7 +85,7 @@ public String toString() {
      * @return One of the possible decision task replies: RespondDecisionTaskCompletedRequest,
      * RespondQueryTaskCompletedRequest, RespondDecisionTaskFailedRequest
      */
-    Result handleDecisionTask(DecisionTaskWithHistoryIterator decisionTaskIterator);
+    Result handleDecisionTask(DecisionTaskWithHistoryIterator decisionTaskIterator) throws Exception;
 
     /**
      * True if this handler handles at least one workflow type.

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

@@ -17,15 +17,21 @@
 
 package com.uber.cadence.internal.worker;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.time.Duration;
 
 /**
  * Options for component that polls Cadence task lists for tasks.
  */
 public final class PollerOptions {
 
+    private static final Logger log = LoggerFactory.getLogger(PollerOptions.class);
+
     public final static class Builder {
 
+
         private int maximumPollRateIntervalMilliseconds = 1000;
 
         private double maximumPollRatePerSecond;
@@ -126,7 +132,7 @@ public Builder setPollThreadNamePrefix(String pollThreadNamePrefix) {
 
         public PollerOptions build() {
             if (uncaughtExceptionHandler == null) {
-                uncaughtExceptionHandler = (t, e) -> e.printStackTrace();
+                uncaughtExceptionHandler = (t, e) -> log.error("uncaught exception", e);
             }
             return new PollerOptions(maximumPollRateIntervalMilliseconds, maximumPollRatePerSecond,
                     pollBackoffCoefficient, pollBackoffInitialInterval, pollBackoffMaximumInterval,

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

@@ -85,7 +85,7 @@ public void start() {
         }
     }
 
-    public byte[] queryWorkflowExecution(WorkflowExecution execution, String queryType, byte[] args) {
+    public byte[] queryWorkflowExecution(WorkflowExecution execution, String queryType, byte[] args) throws Exception {
         Iterator<HistoryEvent> history = WorkflowExecutionUtils.getHistory(service, domain, execution);
         DecisionTaskWithHistoryIterator historyIterator = new ReplayDecisionTaskWithHistoryIterator(execution, history);
         WorkflowQuery query = new WorkflowQuery();

src/main/java/com/uber/cadence/worker/Worker.java

@@ -231,9 +231,10 @@ public String toString() {
      * @param args       query arguments
      * @param <R>        type of the query result
      * @return query result
+     * @throws Exception if replay failed for any reason
      */
     public <R> R queryWorkflowExecution(WorkflowExecution execution, String queryType, Class<R> returnType,
-                                        Object... args) {
+                                        Object... args) throws Exception {
         if (workflowWorker == null) {
             throw new IllegalStateException("disableWorkflowWorker is set in worker options");
         }

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

@@ -790,10 +790,12 @@ public void testSignalUntyped() {
         WorkflowExecution execution = client.start();
         assertEquals("initial", client.query("QueryableWorkflow::getState", String.class));
         client.signal("testSignal", "Hello ");
-        while(!"Hello ".equals(client.query("QueryableWorkflow::getState", String.class))) {}
+        while (!"Hello ".equals(client.query("QueryableWorkflow::getState", String.class))) {
+        }
         assertEquals("Hello ", client.query("QueryableWorkflow::getState", String.class));
         client.signal("testSignal", "World!");
-        while(!"World!".equals(client.query("QueryableWorkflow::getState", String.class))) {}
+        while (!"World!".equals(client.query("QueryableWorkflow::getState", String.class))) {
+        }
         assertEquals("World!", client.query("QueryableWorkflow::getState", String.class));
         assertEquals("Hello World!", workflowClient.newUntypedWorkflowStub(execution).getResult(String.class));
     }
@@ -1144,6 +1146,35 @@ public void testChildWorkflowAsyncRetry() {
         assertEquals(3, AngryChild.invocationCount);
     }
 
+    private static int testDecisionFailureBackoffReplayCount;
+
+    public static class TestDecisionFailureBackoff implements TestWorkflow1 {
+
+        @Override
+        public String execute() {
+            if (testDecisionFailureBackoffReplayCount++ < 2) {
+                throw new Error("simulated decision failure");
+            }
+            return "result1";
+        }
+    }
+
+    @Test
+    public void testDecisionFailureBackoff() {
+        startWorkerFor(TestDecisionFailureBackoff.class);
+        WorkflowOptions o = new WorkflowOptions.Builder()
+                .setExecutionStartToCloseTimeout(Duration.ofSeconds(10))
+                .setTaskStartToCloseTimeout(Duration.ofSeconds(1))
+                .setTaskList(taskList).build();
+
+        TestWorkflow1 workflowStub = workflowClient.newWorkflowStub(TestWorkflow1.class, o);
+        long start = System.currentTimeMillis();
+        String result = workflowStub.execute();
+        long elapsed = System.currentTimeMillis() - start;
+        assertTrue("spinned on fail decision", elapsed > 1000);
+        assertEquals("result1", result);
+    }
+
     public interface TestActivities {
 
         String activityWithDelay(long milliseconds, boolean heartbeatMoreThanOnce);