Make sure workflow_task_execution_failed always has a failure_reason (#2419)

Author: Quinn-With-Two-Ns

temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java

@@ -39,6 +39,7 @@
 import io.temporal.serviceclient.RpcRetryOptions;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import io.temporal.worker.MetricsType;
+import io.temporal.worker.NonDeterministicException;
 import io.temporal.worker.WorkerMetricsTag;
 import io.temporal.worker.WorkflowTaskDispatchHandle;
 import io.temporal.worker.tuning.SlotReleaseReason;
@@ -474,11 +475,21 @@ private WorkflowTaskHandler.Result handleTask(
       try {
         return handler.handleWorkflowTask(task);
       } catch (Throwable e) {
+        workflowTypeMetricsScope.counter(MetricsType.WORKFLOW_TASK_NO_COMPLETION_COUNTER).inc(1);
+        // Make sure that the task failure metric has the correct type
+        Scope workflowTaskFailureScope = workflowTypeMetricsScope;
+        if (e instanceof NonDeterministicException) {
+          workflowTaskFailureScope =
+              workflowTaskFailureScope.tagged(
+                  ImmutableMap.of(TASK_FAILURE_TYPE, "NonDeterminismError"));
+        } else {
+          workflowTaskFailureScope =
+              workflowTaskFailureScope.tagged(ImmutableMap.of(TASK_FAILURE_TYPE, "WorkflowError"));
+        }
         // more detailed logging that we can do here is already done inside `handler`
-        workflowTypeMetricsScope
+        workflowTaskFailureScope
             .counter(MetricsType.WORKFLOW_TASK_EXECUTION_FAILURE_COUNTER)
             .inc(1);
-        workflowTypeMetricsScope.counter(MetricsType.WORKFLOW_TASK_NO_COMPLETION_COUNTER).inc(1);
         throw e;
       } finally {
         sw.stop();

temporal-sdk/src/test/java/io/temporal/workflow/determinism/NonDeterministicWorkflowPolicyBlockWorkflowTest.java

@@ -22,16 +22,21 @@
 
 import static org.junit.Assert.*;
 
+import com.google.common.collect.ImmutableMap;
+import com.uber.m3.tally.RootScopeBuilder;
+import com.uber.m3.tally.Scope;
 import io.temporal.api.enums.v1.EventType;
 import io.temporal.api.enums.v1.WorkflowTaskFailedCause;
 import io.temporal.api.history.v1.HistoryEvent;
 import io.temporal.api.history.v1.WorkflowTaskFailedEventAttributes;
 import io.temporal.client.WorkflowFailedException;
 import io.temporal.client.WorkflowOptions;
 import io.temporal.client.WorkflowStub;
+import io.temporal.common.reporter.TestStatsReporter;
 import io.temporal.failure.TimeoutFailure;
 import io.temporal.internal.sync.WorkflowMethodThreadNameStrategy;
 import io.temporal.testing.internal.SDKTestWorkflowRule;
+import io.temporal.worker.MetricsType;
 import io.temporal.worker.NonDeterministicException;
 import io.temporal.worker.WorkerOptions;
 import io.temporal.workflow.shared.TestActivities.TestActivitiesImpl;
@@ -43,6 +48,9 @@
 import org.junit.Test;
 
 public class NonDeterministicWorkflowPolicyBlockWorkflowTest {
+  private final TestStatsReporter reporter = new TestStatsReporter();
+  Scope metricsScope =
+      new RootScopeBuilder().reporter(reporter).reportEvery(com.uber.m3.util.Duration.ofMillis(1));
 
   @Rule
   public SDKTestWorkflowRule testWorkflowRule =
@@ -54,6 +62,7 @@ public class NonDeterministicWorkflowPolicyBlockWorkflowTest {
               WorkerOptions.newBuilder()
                   .setStickyQueueScheduleToStartTimeout(Duration.ZERO)
                   .build())
+          .setMetricsScope(metricsScope)
           .build();
 
   @Test
@@ -89,6 +98,22 @@ public void testNonDeterministicWorkflowPolicyBlockWorkflow() {
     assertEquals(
         NonDeterministicException.class.getName(),
         failedWFTEventAttributes.getFailure().getApplicationFailureInfo().getType());
+    // Verify that the non-deterministic workflow task failure is reported for all the workflow
+    // tasks
+    reporter.assertCounter(
+        MetricsType.WORKFLOW_TASK_EXECUTION_FAILURE_COUNTER,
+        ImmutableMap.of(
+            "task_queue",
+            testWorkflowRule.getTaskQueue(),
+            "namespace",
+            "UnitTest",
+            "workflow_type",
+            "TestWorkflowStringArg",
+            "worker_type",
+            "WorkflowWorker",
+            "failure_reason",
+            "NonDeterminismError"),
+        (i) -> i >= 2);
   }
 
   @Test