Fix nexus error translation

Author: Quinn-With-Two-Ns

temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java

@@ -8,6 +8,7 @@
 import io.nexusrpc.OperationException;
 import io.nexusrpc.handler.*;
 import io.temporal.api.common.v1.Payload;
+import io.temporal.api.enums.v1.NexusHandlerErrorRetryBehavior;
 import io.temporal.api.nexus.v1.*;
 import io.temporal.client.WorkflowClient;
 import io.temporal.client.WorkflowException;
@@ -42,7 +43,6 @@ public class NexusTaskHandlerImpl implements NexusTaskHandler {
   private final Map<String, ServiceImplInstance> serviceImplInstances =
       Collections.synchronizedMap(new HashMap<>());
   private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
-  private final WorkerInterceptor[] interceptors;
   private final TemporalInterceptorMiddleware nexusServiceInterceptor;
 
   public NexusTaskHandlerImpl(
@@ -55,7 +55,7 @@ public NexusTaskHandlerImpl(
     this.namespace = Objects.requireNonNull(namespace);
     this.taskQueue = Objects.requireNonNull(taskQueue);
     this.dataConverter = Objects.requireNonNull(dataConverter);
-    this.interceptors = Objects.requireNonNull(interceptors);
+    Objects.requireNonNull(interceptors);
     this.nexusServiceInterceptor = new TemporalInterceptorMiddleware(interceptors);
   }
 
@@ -131,6 +131,7 @@ public Result handle(NexusTask task, Scope metricsScope) throws TimeoutException
           HandlerError.newBuilder()
               .setErrorType(e.getErrorType().toString())
               .setFailure(exceptionToNexusFailure(e.getCause(), dataConverter))
+              .setRetryBehavior(mapRetryBehavior(e.getRetryBehavior()))
               .build());
     } catch (Throwable e) {
       return new Result(
@@ -151,6 +152,18 @@ public Result handle(NexusTask task, Scope metricsScope) throws TimeoutException
     }
   }
 
+  private NexusHandlerErrorRetryBehavior mapRetryBehavior(
+      HandlerException.RetryBehavior retryBehavior) {
+    switch (retryBehavior) {
+      case RETRYABLE:
+        return NexusHandlerErrorRetryBehavior.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_RETRYABLE;
+      case NON_RETRYABLE:
+        return NexusHandlerErrorRetryBehavior.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_NON_RETRYABLE;
+      default:
+        return NexusHandlerErrorRetryBehavior.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_UNSPECIFIED;
+    }
+  }
+
   private void cancelOperation(OperationContext context, OperationCancelDetails details) {
     try {
       serviceHandler.cancelOperation(context, details);
@@ -193,7 +206,10 @@ private void convertKnownFailures(Throwable e) {
     }
     if (failure instanceof ApplicationFailure) {
       if (((ApplicationFailure) failure).isNonRetryable()) {
-        throw new HandlerException(HandlerException.ErrorType.BAD_REQUEST, failure);
+        throw new HandlerException(
+            HandlerException.ErrorType.INTERNAL,
+            failure,
+            HandlerException.RetryBehavior.NON_RETRYABLE);
       }
     }
     if (failure instanceof Error) {

temporal-sdk/src/test/java/io/temporal/workflow/nexus/OperationFailMetricTest.java

@@ -27,11 +27,12 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import org.junit.Assert;
-import org.junit.Assume;
 import org.junit.Rule;
 import org.junit.Test;
 
 public class OperationFailMetricTest {
+  private static Map<String, Integer> invocationCount = new ConcurrentHashMap<>();
+
   private final TestStatsReporter reporter = new TestStatsReporter();
 
   @Rule
@@ -43,7 +44,6 @@ public class OperationFailMetricTest {
               new RootScopeBuilder()
                   .reporter(reporter)
                   .reportEvery(com.uber.m3.util.Duration.ofMillis(10)))
-          .setUseExternalService(false)
           .build();
 
   private ImmutableMap.Builder<String, String> getBaseTags() {
@@ -81,6 +81,7 @@ public void failOperationMetrics() {
 
     WorkflowFailedException workflowException =
         Assert.assertThrows(WorkflowFailedException.class, () -> workflowStub.execute("fail"));
+    assertNoRetries("fail");
     ApplicationFailure applicationFailure =
         assertNexusOperationFailure(ApplicationFailure.class, workflowException);
     Assert.assertEquals("intentional failure", applicationFailure.getOriginalMessage());
@@ -107,6 +108,7 @@ public void failOperationApplicationErrorMetrics() {
 
     WorkflowFailedException workflowException =
         Assert.assertThrows(WorkflowFailedException.class, () -> workflowStub.execute("fail-app"));
+    assertNoRetries("fail-app");
     ApplicationFailure applicationFailure =
         assertNexusOperationFailure(ApplicationFailure.class, workflowException);
     Assert.assertEquals("intentional failure", applicationFailure.getOriginalMessage());
@@ -135,8 +137,10 @@ public void failHandlerBadRequestMetrics() {
     WorkflowFailedException workflowException =
         Assert.assertThrows(
             WorkflowFailedException.class, () -> workflowStub.execute("handlererror"));
+    assertNoRetries("handlererror");
     HandlerException handlerException =
         assertNexusOperationFailure(HandlerException.class, workflowException);
+    Assert.assertEquals(HandlerException.ErrorType.BAD_REQUEST, handlerException.getErrorType());
     Assert.assertTrue(handlerException.getCause() instanceof ApplicationFailure);
     ApplicationFailure applicationFailure = (ApplicationFailure) handlerException.getCause();
     Assert.assertEquals("handlererror", applicationFailure.getOriginalMessage());
@@ -165,8 +169,10 @@ public void failHandlerAppBadRequestMetrics() {
     WorkflowFailedException workflowException =
         Assert.assertThrows(
             WorkflowFailedException.class, () -> workflowStub.execute("handlererror-app"));
+    assertNoRetries("handlererror-app");
     HandlerException handlerException =
         assertNexusOperationFailure(HandlerException.class, workflowException);
+    Assert.assertEquals(HandlerException.ErrorType.BAD_REQUEST, handlerException.getErrorType());
     Assert.assertTrue(handlerException.getCause() instanceof ApplicationFailure);
     ApplicationFailure applicationFailure = (ApplicationFailure) handlerException.getCause();
     Assert.assertEquals("intentional failure", applicationFailure.getOriginalMessage());
@@ -192,19 +198,24 @@ public void failHandlerAppBadRequestMetrics() {
 
   @Test
   public void failHandlerAlreadyStartedMetrics() {
-    Assume.assumeFalse("skipping", true);
     TestWorkflow1 workflowStub =
         testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflow1.class);
     WorkflowFailedException workflowException =
         Assert.assertThrows(
             WorkflowFailedException.class, () -> workflowStub.execute("already-started"));
-    ApplicationFailure applicationFailure =
-        assertNexusOperationFailure(ApplicationFailure.class, workflowException);
+    assertNoRetries("already-started");
+    HandlerException handlerException =
+        assertNexusOperationFailure(HandlerException.class, workflowException);
+    Assert.assertEquals(HandlerException.ErrorType.BAD_REQUEST, handlerException.getErrorType());
+    Assert.assertTrue(handlerException.getCause() instanceof ApplicationFailure);
+    ApplicationFailure applicationFailure = (ApplicationFailure) handlerException.getCause();
     Assert.assertEquals(
         "io.temporal.client.WorkflowExecutionAlreadyStarted", applicationFailure.getType());
 
     Map<String, String> execFailedTags =
-        getOperationTags().put(MetricsTag.TASK_FAILURE_TYPE, "operation_failed").buildKeepingLast();
+        getOperationTags()
+            .put(MetricsTag.TASK_FAILURE_TYPE, "handler_error_BAD_REQUEST")
+            .buildKeepingLast();
     Eventually.assertEventually(
         Duration.ofSeconds(3),
         () -> {
@@ -224,6 +235,7 @@ public void failHandlerRetryableApplicationFailureMetrics() {
         testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflow1.class);
     Assert.assertThrows(
         WorkflowFailedException.class, () -> workflowStub.execute("retryable-application-failure"));
+    Assert.assertTrue(invocationCount.get("retryable-application-failure") > 1);
 
     Map<String, String> execFailedTags =
         getOperationTags()
@@ -253,12 +265,16 @@ public void failHandlerNonRetryableApplicationFailureMetrics() {
             () -> workflowStub.execute("non-retryable-application-failure"));
     HandlerException handlerFailure =
         assertNexusOperationFailure(HandlerException.class, workflowException);
+    assertNoRetries("non-retryable-application-failure");
+
     Assert.assertTrue(handlerFailure.getMessage().contains("intentional failure"));
-    Assert.assertEquals(HandlerException.ErrorType.BAD_REQUEST, handlerFailure.getErrorType());
+    Assert.assertEquals(HandlerException.ErrorType.INTERNAL, handlerFailure.getErrorType());
+    Assert.assertEquals(
+        HandlerException.RetryBehavior.NON_RETRYABLE, handlerFailure.getRetryBehavior());
 
     Map<String, String> execFailedTags =
         getOperationTags()
-            .put(MetricsTag.TASK_FAILURE_TYPE, "handler_error_BAD_REQUEST")
+            .put(MetricsTag.TASK_FAILURE_TYPE, "handler_error_INTERNAL")
             .buildKeepingLast();
     Eventually.assertEventually(
         Duration.ofSeconds(3),
@@ -298,6 +314,7 @@ public void failHandlerErrorMetrics() {
         testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflow1.class);
     WorkflowFailedException workflowException =
         Assert.assertThrows(WorkflowFailedException.class, () -> workflowStub.execute("error"));
+    Assert.assertTrue(invocationCount.get("error") > 1);
 
     Map<String, String> execFailedTags =
         getOperationTags()
@@ -324,6 +341,13 @@ public void handlerErrorNonRetryableMetrics() {
     WorkflowFailedException workflowException =
         Assert.assertThrows(
             WorkflowFailedException.class, () -> workflowStub.execute("handlererror-nonretryable"));
+    assertNoRetries("handlererror-nonretryable");
+    HandlerException handlerFailure =
+        assertNexusOperationFailure(HandlerException.class, workflowException);
+    Assert.assertTrue(handlerFailure.getMessage().contains("intentional failure"));
+    Assert.assertEquals(HandlerException.ErrorType.INTERNAL, handlerFailure.getErrorType());
+    Assert.assertEquals(
+        HandlerException.RetryBehavior.NON_RETRYABLE, handlerFailure.getRetryBehavior());
 
     Map<String, String> execFailedTags =
         getOperationTags()
@@ -342,6 +366,10 @@ public void handlerErrorNonRetryableMetrics() {
         });
   }
 
+  private void assertNoRetries(String testCase) {
+    Assert.assertEquals(new Integer(1), invocationCount.get(testCase));
+  }
+
   public static class TestNexus implements TestWorkflow1 {
     @Override
     public String execute(String operation) {
@@ -360,17 +388,13 @@ public String execute(String operation) {
 
   @ServiceImpl(service = TestNexusServices.TestNexusService1.class)
   public class TestNexusServiceImpl {
-    Map<String, Integer> invocationCount = new ConcurrentHashMap<>();
-
     @OperationImpl
     public OperationHandler<String, String> operation() {
       // Implemented inline
       return OperationHandler.sync(
           (ctx, details, operation) -> {
-            invocationCount.put(
-                details.getRequestId(),
-                invocationCount.getOrDefault(details.getRequestId(), 0) + 1);
-            if (invocationCount.get(details.getRequestId()) > 1) {
+            invocationCount.put(operation, invocationCount.getOrDefault(operation, 0) + 1);
+            if (invocationCount.get(operation) > 1) {
               throw OperationException.failure(new RuntimeException("exceeded invocation count"));
             }
             switch (operation) {

temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowService.java

@@ -1038,7 +1038,10 @@ private static Failure handlerErrorToFailure(HandlerError err) {
     return Failure.newBuilder()
         .setMessage(err.getFailure().getMessage())
         .setNexusHandlerFailureInfo(
-            NexusHandlerFailureInfo.newBuilder().setType(err.getErrorType()).build())
+            NexusHandlerFailureInfo.newBuilder()
+                .setType(err.getErrorType())
+                .setRetryBehavior(err.getRetryBehavior())
+                .build())
         .setCause(nexusFailureToAPIFailure(err.getFailure(), false))
         .build();
   }