Update Java SDK for Temporal Sever v1.26.2 (#2357)

Quinn-With-Two-Ns · web-flow · commit 82d3c93d5e5f · 2025-01-06T14:22:42.000-08:00
Lower refreshNexusEndpointsMinWait
diff --git a/docker/github/dynamicconfig/development.yaml b/docker/github/dynamicconfig/development.yaml
@@ -27,4 +27,6 @@ component.nexusoperations.callback.endpoint.template:
 component.callbacks.allowedAddresses:
   - value:
       - Pattern: "localhost:7243"
-        AllowInsecure: true
+        AllowInsecure: true
+system.refreshNexusEndpointsMinWait:
+  - value: 1ms
diff --git a/temporal-sdk/src/main/java/io/temporal/failure/DefaultFailureConverter.java b/temporal-sdk/src/main/java/io/temporal/failure/DefaultFailureConverter.java
@@ -189,7 +189,14 @@ private TemporalFailure failureToExceptionImpl(Failure failure, DataConverter da
         }
       case FAILUREINFO_NOT_SET:
       default:
-        throw new IllegalArgumentException("Failure info not set");
+        // All unknown types are considered to be retryable ApplicationError.
+        return ApplicationFailure.newFromValues(
+            failure.getMessage(),
+            "",
+            false,
+            new EncodedValues(Optional.empty(), dataConverter),
+            cause,
+            null);
     }
   }
 
diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java
@@ -334,6 +334,8 @@ public Throwable wrapFailure(ActivityTask t, Throwable failure) {
           failure);
     }
 
+    // TODO: Suppress warning until the SDK supports deployment
+    @SuppressWarnings("deprecation")
     private void sendReply(
         ByteString taskToken, ActivityTaskHandler.Result response, Scope metricsScope) {
       RespondActivityTaskCompletedRequest taskCompleted = response.getTaskCompleted();
diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java
@@ -486,6 +486,8 @@ private WorkflowTaskHandler.Result handleTask(
       }
     }
 
+    // TODO: Suppress warning until the SDK supports deployment
+    @SuppressWarnings("deprecation")
     private RespondWorkflowTaskCompletedResponse sendTaskCompleted(
         ByteString taskToken,
         RespondWorkflowTaskCompletedRequest.Builder taskCompleted,
@@ -514,6 +516,8 @@ private RespondWorkflowTaskCompletedResponse sendTaskCompleted(
           grpcRetryOptions);
     }
 
+    // TODO: Suppress warning until the SDK supports deployment
+    @SuppressWarnings("deprecation")
     private void sendTaskFailed(
         ByteString taskToken,
         RespondWorkflowTaskFailedRequest.Builder taskFailed,
diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/ActivityTimeoutTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/ActivityTimeoutTest.java
@@ -56,7 +56,6 @@
 import org.hamcrest.CoreMatchers;
 import org.hamcrest.MatcherAssert;
 import org.junit.*;
-import org.junit.rules.Timeout;
 import org.junit.runner.RunWith;
 
 /**
@@ -70,13 +69,11 @@
  */
 @RunWith(JUnitParamsRunner.class)
 public class ActivityTimeoutTest {
-  @Rule
-  public SDKTestWorkflowRule testWorkflowRule =
-      SDKTestWorkflowRule.newBuilder().setDoNotStart(true).build();
-
   // TODO This test takes longer than it should to complete because
   //  of the cached heartbeat that prevents a quick shutdown
-  public @Rule Timeout timeout = Timeout.seconds(15);
+  @Rule
+  public SDKTestWorkflowRule testWorkflowRule =
+      SDKTestWorkflowRule.newBuilder().setTestTimeoutSeconds(15).setDoNotStart(true).build();
 
   /**
    * An activity reaches startToClose timeout once, max retries are set to 1. o
diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/SyncOperationCancelledTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/SyncOperationCancelledTest.java
@@ -58,17 +58,6 @@ public void syncOperationImmediatelyCancelled() {
         "operation canceled before it was started", canceledFailure.getOriginalMessage());
   }
 
-  @Test
-  public void syncOperationCancelled() {
-    TestWorkflows.TestWorkflow1 workflowStub =
-        testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class);
-    WorkflowFailedException exception =
-        Assert.assertThrows(WorkflowFailedException.class, () -> workflowStub.execute(""));
-    Assert.assertTrue(exception.getCause() instanceof NexusOperationFailure);
-    NexusOperationFailure nexusFailure = (NexusOperationFailure) exception.getCause();
-    Assert.assertTrue(nexusFailure.getCause() instanceof CanceledFailure);
-  }
-
   public static class TestNexus implements TestWorkflows.TestWorkflow1 {
     @Override
     public String execute(String input) {
diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/TerminateWorkflowAsyncOperationTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/TerminateWorkflowAsyncOperationTest.java
@@ -29,6 +29,7 @@
 import io.temporal.client.WorkflowOptions;
 import io.temporal.failure.ApplicationFailure;
 import io.temporal.failure.NexusOperationFailure;
+import io.temporal.failure.TerminatedFailure;
 import io.temporal.nexus.WorkflowClientOperationHandlers;
 import io.temporal.testing.internal.SDKTestWorkflowRule;
 import io.temporal.workflow.*;
@@ -54,10 +55,20 @@ public void terminateAsyncOperation() {
         Assert.assertThrows(WorkflowFailedException.class, () -> workflowStub.execute(""));
     Assert.assertTrue(exception.getCause() instanceof NexusOperationFailure);
     NexusOperationFailure nexusFailure = (NexusOperationFailure) exception.getCause();
-    Assert.assertTrue(nexusFailure.getCause() instanceof ApplicationFailure);
-    Assert.assertEquals(
-        "operation terminated",
-        ((ApplicationFailure) nexusFailure.getCause()).getOriginalMessage());
+    // TODO(https://github.com/temporalio/sdk-java/issues/2358): Test server needs to be fixed to
+    // return the correct type
+    Assert.assertTrue(
+        nexusFailure.getCause() instanceof ApplicationFailure
+            || nexusFailure.getCause() instanceof TerminatedFailure);
+    if (nexusFailure.getCause() instanceof ApplicationFailure) {
+      Assert.assertEquals(
+          "operation terminated",
+          ((ApplicationFailure) nexusFailure.getCause()).getOriginalMessage());
+    } else {
+      Assert.assertEquals(
+          "operation terminated",
+          ((TerminatedFailure) nexusFailure.getCause()).getOriginalMessage());
+    }
   }
 
   @Service
diff --git a/temporal-serviceclient/src/main/proto b/temporal-serviceclient/src/main/proto
@@ -1 +1 @@
-Subproject commit f6a14e5ad41fcbc0ea1117348daea90ba2c8c3c3
+Subproject commit bae790a6cfe59936cb26e613af107780deea500e
diff --git a/temporal-test-server/src/main/java/io/temporal/internal/testservice/StateMachines.java b/temporal-test-server/src/main/java/io/temporal/internal/testservice/StateMachines.java
@@ -300,6 +300,7 @@ static final class ActivityTaskData {
     TestServiceRetryState retryState;
     Duration nextBackoffInterval;
     String identity;
+    Timestamp lastAttemptCompleteTime;
 
     ActivityTaskData(
         TestWorkflowStore store, StartWorkflowExecutionRequest startWorkflowExecutionRequest) {
@@ -2112,6 +2113,7 @@ private static RetryState attemptActivityRetry(
       ctx.onCommit(
           (historySize) -> {
             data.retryState = nextAttempt;
+            data.lastAttemptCompleteTime = ctx.currentTime();
             task.setAttempt(nextAttempt.getAttempt());
             task.setCurrentAttemptScheduledTime(ctx.currentTime());
           });
diff --git a/temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowMutableStateImpl.java b/temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowMutableStateImpl.java
@@ -28,6 +28,7 @@
 
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
+import com.google.protobuf.Any;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Timestamp;
@@ -86,6 +87,17 @@
 import org.slf4j.LoggerFactory;
 
 class TestWorkflowMutableStateImpl implements TestWorkflowMutableState {
+  static final Failure FAILED_UPDATE_ON_WF_COMPLETION =
+      Failure.newBuilder()
+          .setMessage(
+              "Workflow Update failed because the Workflow completed before the Update completed.")
+          .setSource("Server")
+          .setApplicationFailureInfo(
+              ApplicationFailureInfo.newBuilder()
+                  .setType("AcceptedUpdateCompletedWorkflow")
+                  .setNonRetryable(true)
+                  .build())
+          .build();
 
   /**
    * If the implementation throws an exception, changes accumulated in the RequestContext will not
@@ -541,6 +553,7 @@ public void completeWorkflowTask(
                     || request.getForceCreateNewWorkflowTask())) {
               scheduleWorkflowTask(ctx);
             }
+
             workflowTaskStateMachine.getData().bufferedEvents.clear();
             Map<String, ConsistentQuery> queries = data.consistentQueryRequests;
             Map<String, WorkflowQueryResult> queryResultsMap = request.getQueryResultsMap();
@@ -1671,6 +1684,27 @@ private void processWorkflowCompletionCallbacks(RequestContext ctx) {
       return;
     }
 
+    updates.forEach(
+        (k, updateStateMachine) -> {
+          if (!(updateStateMachine.getState() == StateMachines.State.COMPLETED
+              || updateStateMachine.getState() == StateMachines.State.FAILED)) {
+            updateStateMachine.action(
+                Action.COMPLETE,
+                ctx,
+                Message.newBuilder()
+                    .setBody(
+                        Any.pack(
+                            Response.newBuilder()
+                                .setOutcome(
+                                    Outcome.newBuilder()
+                                        .setFailure(FAILED_UPDATE_ON_WF_COMPLETION)
+                                        .build())
+                                .build()))
+                    .build(),
+                completionEvent.get().getEventId());
+          }
+        });
+
     for (Callback cb : startRequest.getCompletionCallbacksList()) {
       if (!cb.hasNexus()) {
         // test server only supports nexus callbacks currently
@@ -3101,6 +3135,10 @@ private static PendingActivityInfo constructPendingActivityInfo(
       builder.setLastWorkerIdentity(activityTaskData.identity);
     }
 
+    if (activityTaskData.lastAttemptCompleteTime != null) {
+      builder.setLastAttemptCompleteTime(activityTaskData.lastAttemptCompleteTime);
+    }
+
     // Some ids are only present in the schedule event...
     if (activityTaskData.scheduledEvent != null) {
       populatePendingActivityInfoFromScheduledEvent(builder, activityTaskData.scheduledEvent);
@@ -3145,12 +3183,8 @@ private static void populatePendingActivityInfoFromScheduledEvent(
 
   private static void populatePendingActivityInfoFromPollResponse(
       PendingActivityInfo.Builder builder, PollActivityTaskQueueResponseOrBuilder task) {
-    // In golang, we set one but never both of these fields, depending on the activity state
-    if (builder.getState() == PendingActivityState.PENDING_ACTIVITY_STATE_SCHEDULED) {
-      builder.setScheduledTime(task.getScheduledTime());
-    } else {
-      builder.setLastStartedTime(task.getStartedTime());
-    }
+    builder.setScheduledTime(task.getScheduledTime());
+    builder.setLastStartedTime(task.getStartedTime());
   }
 
   private static void populatePendingActivityInfoFromHeartbeatDetails(
diff --git a/temporal-test-server/src/test/java/io/temporal/testserver/functional/DescribeWorkflowExecutionTest.java b/temporal-test-server/src/test/java/io/temporal/testserver/functional/DescribeWorkflowExecutionTest.java
@@ -175,6 +175,7 @@ public void testSuccessfulActivity() throws InterruptedException {
             .setMaximumAttempts(2)
             // times should be present, but we can't know what the expected value is if this test is
             // going to run against the real server.
+            .setScheduledTime(actual.getScheduledTime())
             .setLastStartedTime(actual.getLastStartedTime())
             .setExpirationTime(actual.getExpirationTime())
             .build();
@@ -266,8 +267,10 @@ public void testFailedActivity() throws InterruptedException {
             .setMaximumAttempts(2)
             // times should be present, but we can't know what the expected value is if this test is
             // going to run against the real server.
+            .setScheduledTime(actual.getScheduledTime())
             .setLastStartedTime(actual.getLastStartedTime())
             .setExpirationTime(actual.getExpirationTime())
+            .setLastAttemptCompleteTime(actual.getLastAttemptCompleteTime())
             // this ends up being a dummy value, but if it weren't, we still wouldn't expect to know
             // it.
             .setLastWorkerIdentity(actual.getLastWorkerIdentity())
@@ -333,6 +336,7 @@ private void testKilledWorkflow(
             .setMaximumAttempts(2)
             // times should be present, but we can't know what the expected value is if this test is
             // going to run against the real server.
+            .setScheduledTime(actual.getScheduledTime())
             .setLastStartedTime(actual.getLastStartedTime())
             .setExpirationTime(actual.getExpirationTime())
             // this ends up being a dummy value, but if it weren't, we still wouldn't expect to know
diff --git a/temporal-test-server/src/test/java/io/temporal/testserver/functional/WorkflowUpdateTest.java b/temporal-test-server/src/test/java/io/temporal/testserver/functional/WorkflowUpdateTest.java
@@ -20,8 +20,6 @@
 
 package io.temporal.testserver.functional;
 
-import static org.junit.Assume.assumeFalse;
-
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.temporal.api.common.v1.Payloads;
@@ -548,7 +546,6 @@ public void updateAndPollByWorkflowId() {
   @Test
   public void getCompletedUpdateOfCompletedWorkflow() {
     // Assert that we can get and poll a completed update from a completed workflow.
-    assumeFalse("Skipping as real server has a bug", SDKTestWorkflowRule.useExternalService);
 
     WorkflowOptions options =
         WorkflowOptions.newBuilder().setTaskQueue(testWorkflowRule.getTaskQueue()).build();
@@ -593,7 +590,7 @@ public void getCompletedUpdateOfCompletedWorkflow() {
 
   @Test
   public void getIncompleteUpdateOfCompletedWorkflow() {
-    // Assert that we can't get an incomplete update of a completed workflow. Expect a NOT_FOUND
+    // Assert that the server fails an incomplete update if the workflow is completed.
     WorkflowOptions options =
         WorkflowOptions.newBuilder().setTaskQueue(testWorkflowRule.getTaskQueue()).build();
 
@@ -617,48 +614,60 @@ public void getIncompleteUpdateOfCompletedWorkflow() {
     workflowStub.signal();
     workflowStub.execute();
 
-    StatusRuntimeException exception =
-        Assert.assertThrows(
-            StatusRuntimeException.class,
-            () ->
-                updateWorkflow(
-                    exec,
-                    "updateId",
-                    UpdateWorkflowExecutionLifecycleStage
-                        .UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED,
-                    TestWorkflows.UpdateType.BLOCK));
-    Assert.assertEquals(Status.NOT_FOUND.getCode(), exception.getStatus().getCode());
-    exception =
-        Assert.assertThrows(
-            StatusRuntimeException.class,
-            () ->
-                updateWorkflow(
-                    exec,
-                    "updateId",
-                    UpdateWorkflowExecutionLifecycleStage
-                        .UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED,
-                    TestWorkflows.UpdateType.BLOCK));
-    Assert.assertEquals(Status.NOT_FOUND.getCode(), exception.getStatus().getCode());
-    exception =
-        Assert.assertThrows(
-            StatusRuntimeException.class,
-            () ->
-                pollWorkflowUpdate(
-                    exec,
-                    "updateId",
-                    UpdateWorkflowExecutionLifecycleStage
-                        .UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED));
-    Assert.assertEquals(Status.NOT_FOUND.getCode(), exception.getStatus().getCode());
-    exception =
-        Assert.assertThrows(
-            StatusRuntimeException.class,
-            () ->
-                pollWorkflowUpdate(
-                    exec,
-                    "updateId",
-                    UpdateWorkflowExecutionLifecycleStage
-                        .UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED));
-    Assert.assertEquals(Status.NOT_FOUND.getCode(), exception.getStatus().getCode());
+    response =
+        updateWorkflow(
+            exec,
+            "updateId",
+            UpdateWorkflowExecutionLifecycleStage
+                .UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED,
+            TestWorkflows.UpdateType.BLOCK);
+    Assert.assertEquals(
+        UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED,
+        response.getStage());
+    assertUpdateOutcomeIsAcceptedUpdateCompletedWorkflow(response.getOutcome());
+
+    response =
+        updateWorkflow(
+            exec,
+            "updateId",
+            UpdateWorkflowExecutionLifecycleStage
+                .UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED,
+            TestWorkflows.UpdateType.BLOCK);
+    Assert.assertEquals(
+        UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED,
+        response.getStage());
+    assertUpdateOutcomeIsAcceptedUpdateCompletedWorkflow(response.getOutcome());
+
+    PollWorkflowExecutionUpdateResponse pollResponse =
+        pollWorkflowUpdate(
+            exec,
+            "updateId",
+            UpdateWorkflowExecutionLifecycleStage
+                .UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED);
+    Assert.assertEquals(
+        UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED,
+        pollResponse.getStage());
+    assertUpdateOutcomeIsAcceptedUpdateCompletedWorkflow(pollResponse.getOutcome());
+
+    pollResponse =
+        pollWorkflowUpdate(
+            exec,
+            "updateId",
+            UpdateWorkflowExecutionLifecycleStage
+                .UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED);
+    Assert.assertEquals(
+        UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED,
+        pollResponse.getStage());
+    assertUpdateOutcomeIsAcceptedUpdateCompletedWorkflow(pollResponse.getOutcome());
+  }
+
+  private void assertUpdateOutcomeIsAcceptedUpdateCompletedWorkflow(Outcome outcome) {
+    Assert.assertEquals(
+        "Workflow Update failed because the Workflow completed before the Update completed.",
+        outcome.getFailure().getMessage());
+    Assert.assertEquals(
+        "AcceptedUpdateCompletedWorkflow",
+        outcome.getFailure().getApplicationFailureInfo().getType());
   }
 
   private UpdateWorkflowExecutionResponse updateWorkflow(

Original file line number	Diff line number	Diff line change
`@@ -189,7 +189,14 @@ private TemporalFailure failureToExceptionImpl(Failure failure, DataConverter da`
`189`	`189`	`}`
`190`	`190`	`case FAILUREINFO_NOT_SET:`
`191`	`191`	`default:`
`192`		`- throw new IllegalArgumentException("Failure info not set");`
	`192`	`+ // All unknown types are considered to be retryable ApplicationError.`
	`193`	`+ return ApplicationFailure.newFromValues(`
	`194`	`+ failure.getMessage(),`
	`195`	`+ "",`
	`196`	`+ false,`
	`197`	`+ new EncodedValues(Optional.empty(), dataConverter),`
	`198`	`+ cause,`
	`199`	`+ null);`
`193`	`200`	`}`
`194`	`201`	`}`
`195`	`202`
Original file line number	Diff line number	Diff line change
`@@ -334,6 +334,8 @@ public Throwable wrapFailure(ActivityTask t, Throwable failure) {`
`334`	`334`	`failure);`
`335`	`335`	`}`
`336`	`336`
	`337`	`+ // TODO: Suppress warning until the SDK supports deployment`
	`338`	`+ @SuppressWarnings("deprecation")`
`337`	`339`	`private void sendReply(`
`338`	`340`	`ByteString taskToken, ActivityTaskHandler.Result response, Scope metricsScope) {`
`339`	`341`	`RespondActivityTaskCompletedRequest taskCompleted = response.getTaskCompleted();`
Original file line number	Diff line number	Diff line change
`@@ -486,6 +486,8 @@ private WorkflowTaskHandler.Result handleTask(`
`486`	`486`	`}`
`487`	`487`	`}`
`488`	`488`
	`489`	`+ // TODO: Suppress warning until the SDK supports deployment`
	`490`	`+ @SuppressWarnings("deprecation")`
`489`	`491`	`private RespondWorkflowTaskCompletedResponse sendTaskCompleted(`
`490`	`492`	`ByteString taskToken,`
`491`	`493`	`RespondWorkflowTaskCompletedRequest.Builder taskCompleted,`
`@@ -514,6 +516,8 @@ private RespondWorkflowTaskCompletedResponse sendTaskCompleted(`
`514`	`516`	`grpcRetryOptions);`
`515`	`517`	`}`
`516`	`518`
	`519`	`+ // TODO: Suppress warning until the SDK supports deployment`
	`520`	`+ @SuppressWarnings("deprecation")`
`517`	`521`	`private void sendTaskFailed(`
`518`	`522`	`ByteString taskToken,`
`519`	`523`	`RespondWorkflowTaskFailedRequest.Builder taskFailed,`