Merge branch 'master' into add_application_err_category

THardy98 · web-flow · commit 41824855b661 · 2025-04-25T14:58:03.000-07:00
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
@@ -8,7 +8,7 @@ before we can merge in any of your changes
 
 ## Development Environment
 
-* Java 11+
+* Java 21+
 * Docker to run Temporal Server
 
 ## Build
diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenUtil.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenUtil.java
@@ -37,17 +37,15 @@ public class OperationTokenUtil {
   /**
    * Load a workflow run operation token from an operation token.
    *
-   * @throws FallbackToWorkflowIdException if the operation token is not a workflow run token
    * @throws IllegalArgumentException if the operation token is invalid
    */
-  public static WorkflowRunOperationToken loadWorkflowRunOperationToken(String operationToken)
-      throws FallbackToWorkflowIdException {
+  public static WorkflowRunOperationToken loadWorkflowRunOperationToken(String operationToken) {
     WorkflowRunOperationToken token;
     try {
       JavaType reference = mapper.getTypeFactory().constructType(WorkflowRunOperationToken.class);
       token = mapper.readValue(decoder.decode(operationToken), reference);
     } catch (Exception e) {
-      throw new FallbackToWorkflowIdException("Failed to parse operation token: " + e.getMessage());
+      throw new IllegalArgumentException("Failed to parse operation token: " + e.getMessage());
     }
     if (!token.getType().equals(OperationTokenType.WORKFLOW_RUN)) {
       throw new IllegalArgumentException(
@@ -68,16 +66,7 @@ public static WorkflowRunOperationToken loadWorkflowRunOperationToken(String ope
    * @throws IllegalArgumentException if the operation token is invalid
    */
   public static String loadWorkflowIdFromOperationToken(String operationToken) {
-    try {
-      WorkflowRunOperationToken token = loadWorkflowRunOperationToken(operationToken);
-      return token.getWorkflowId();
-    } catch (OperationTokenUtil.FallbackToWorkflowIdException e) {
-      // Previous versions of the SDK simply used the workflow ID as the operation token
-      // This fallback is provided for backwards compatibility for those cases.
-      // This fallback will be removed in a future release.
-      // See: https://github.com/temporalio/sdk-java/issues/2423
-      return operationToken;
-    }
+    return loadWorkflowRunOperationToken(operationToken).getWorkflowId();
   }
 
   /** Generate a workflow run operation token from a workflow ID and namespace. */
@@ -87,11 +76,5 @@ public static String generateWorkflowRunOperationToken(String workflowId, String
     return encoder.encodeToString(json.getBytes());
   }
 
-  public static class FallbackToWorkflowIdException extends RuntimeException {
-    public FallbackToWorkflowIdException(String message) {
-      super(message);
-    }
-  }
-
   private OperationTokenUtil() {}
 }
diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunnerImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunnerImpl.java
@@ -449,7 +449,7 @@ private WorkflowThread newRootThread(Runnable runnable) {
           "newRootThread can be called only if there is no existing root workflow thread");
     }
     rootWorkflowThread =
-        new WorkflowThreadImpl(
+        new RootWorkflowThreadImpl(
             workflowThreadExecutor,
             workflowContext,
             this,
diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/RootWorkflowThreadImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/RootWorkflowThreadImpl.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.temporal.internal.sync;
+
+import io.temporal.common.context.ContextPropagator;
+import io.temporal.internal.worker.WorkflowExecutorCache;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+import javax.annotation.Nonnull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RootWorkflowThreadImpl extends WorkflowThreadImpl {
+  private static final Logger log = LoggerFactory.getLogger(RootWorkflowThreadImpl.class);
+
+  RootWorkflowThreadImpl(
+      WorkflowThreadExecutor workflowThreadExecutor,
+      SyncWorkflowContext syncWorkflowContext,
+      DeterministicRunnerImpl runner,
+      @Nonnull String name,
+      int priority,
+      boolean detached,
+      CancellationScopeImpl parentCancellationScope,
+      Runnable runnable,
+      WorkflowExecutorCache cache,
+      List<ContextPropagator> contextPropagators,
+      Map<String, Object> propagatedContexts) {
+    super(
+        workflowThreadExecutor,
+        syncWorkflowContext,
+        runner,
+        name,
+        priority,
+        detached,
+        parentCancellationScope,
+        runnable,
+        cache,
+        contextPropagators,
+        propagatedContexts);
+  }
+
+  @Override
+  public void yield(String reason, Supplier<Boolean> unblockCondition)
+      throws DestroyWorkflowThreadError {
+    log.warn(
+        "Detected root workflow thread yielding {}. This can happen by making blocking calls during workflow instance creating, such as executing an activity inside a @WorkflowInit constructor. This can cause issues, like Queries or Updates rejection because the workflow instance creation is delayed",
+        getName());
+    super.yield(reason, unblockCondition);
+  }
+}
diff --git a/temporal-sdk/src/main/java/io/temporal/workflow/WorkflowInit.java b/temporal-sdk/src/main/java/io/temporal/workflow/WorkflowInit.java
@@ -34,7 +34,9 @@
  * thrown by the constructor are treated the same as exceptions thrown by the workflow method.
  *
  * <p>Workflow initialization methods are called before the workflow method, signal handlers, update
- * handlers or query handlers.
+ * handlers or query handlers. Users should be careful not to block in constructors. Blocking in the
+ * constructor will delay handling of the workflow method and signal handlers, and cause rejection
+ * of the update and query handlers.
  *
  * <p>This annotation applies only to workflow implementation constructors.
  */
diff --git a/temporal-sdk/src/test/java/io/temporal/internal/nexus/WorkflowRunTokenTest.java b/temporal-sdk/src/test/java/io/temporal/internal/nexus/WorkflowRunTokenTest.java
@@ -76,10 +76,11 @@ public void deserializeWorkflowRunToken() throws IOException {
   }
 
   @Test
-  public void loadOldWorkflowRunToken() {
+  public void failLoadOldWorkflowRunToken() {
     String operationToken = "AAAAA-BBBBB-CCCCC";
-    Assert.assertEquals(
-        operationToken, OperationTokenUtil.loadWorkflowIdFromOperationToken(operationToken));
+    Assert.assertThrows(
+        IllegalArgumentException.class,
+        () -> OperationTokenUtil.loadWorkflowIdFromOperationToken(operationToken));
   }
 
   @Test
diff --git a/temporal-sdk/src/test/java/io/temporal/internal/testing/ActivityTestingTest.java b/temporal-sdk/src/test/java/io/temporal/internal/testing/ActivityTestingTest.java
@@ -31,6 +31,7 @@
 import io.temporal.client.ActivityCanceledException;
 import io.temporal.failure.ActivityFailure;
 import io.temporal.failure.ApplicationFailure;
+import io.temporal.testing.ActivityRequestedAsyncCompletion;
 import io.temporal.testing.TestActivityEnvironment;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -39,10 +40,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
+import org.junit.*;
 import org.junit.rules.Timeout;
 
 public class ActivityTestingTest {
@@ -111,6 +109,21 @@ public void testFailure() {
     }
   }
 
+  private static class AsyncActivityImpl implements TestActivity {
+    @Override
+    public String activity1(String input) {
+      Activity.getExecutionContext().doNotCompleteOnReturn();
+      return "";
+    }
+  }
+
+  @Test
+  public void testAsyncActivity() {
+    testEnvironment.registerActivitiesImplementations(new AsyncActivityImpl());
+    TestActivity activity = testEnvironment.newActivityStub(TestActivity.class);
+    Assert.assertThrows(ActivityRequestedAsyncCompletion.class, () -> activity.activity1("input1"));
+  }
+
   private static class HeartbeatActivityImpl implements TestActivity {
 
     @Override
diff --git a/temporal-testing/src/main/java/io/temporal/testing/ActivityRequestedAsyncCompletion.java b/temporal-testing/src/main/java/io/temporal/testing/ActivityRequestedAsyncCompletion.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.temporal.testing;
+
+/**
+ * Exception thrown when an activity request to complete asynchronously in the {@link
+ * TestActivityEnvironment}. Intended to be used in unit tests to assert an activity requested async
+ * completion.
+ */
+public final class ActivityRequestedAsyncCompletion extends RuntimeException {
+  private final String activityId;
+  private final boolean manualCompletion;
+
+  public ActivityRequestedAsyncCompletion(String activityId, boolean manualCompletion) {
+    super("activity requested async completion");
+    this.activityId = activityId;
+    this.manualCompletion = manualCompletion;
+  }
+
+  public String getActivityId() {
+    return activityId;
+  }
+
+  public boolean isManualCompletion() {
+    return manualCompletion;
+  }
+}
diff --git a/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironment.java b/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironment.java
@@ -82,6 +82,9 @@ static TestActivityEnvironment newInstance(TestEnvironmentOptions options) {
    * Creates a stub that can be used to invoke activities registered through {@link
    * #registerActivitiesImplementations(Object...)}.
    *
+   * <p>Activity methods may throw {@link ActivityRequestedAsyncCompletion} if the activity
+   * requested async completion.
+   *
    * @param activityInterface activity interface class that the object under test implements.
    * @param <T> Type of the activity interface.
    * @return The stub that implements the activity interface.
@@ -92,6 +95,9 @@ static TestActivityEnvironment newInstance(TestEnvironmentOptions options) {
    * Creates a stub that can be used to invoke activities registered through {@link
    * #registerActivitiesImplementations(Object...)}.
    *
+   * <p>Activity methods may throw {@link ActivityRequestedAsyncCompletion} if the activity
+   * requested async completion.
+   *
    * @param <T> Type of the activity interface.
    * @param activityInterface activity interface class that the object under test implements
    * @param options options that specify the activity invocation parameters
diff --git a/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java b/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java
@@ -20,7 +20,6 @@
 
 package io.temporal.testing;
 
-import com.google.common.base.Defaults;
 import com.google.protobuf.ByteString;
 import com.uber.m3.tally.NoopScope;
 import com.uber.m3.tally.Scope;
@@ -511,40 +510,38 @@ private <T> T getReply(
         Type resultType) {
       DataConverter dataConverter =
           testEnvironmentOptions.getWorkflowClientOptions().getDataConverter();
-      RespondActivityTaskCompletedRequest taskCompleted = response.getTaskCompleted();
-      if (taskCompleted != null) {
+      if (response.getTaskCompleted() != null) {
+        RespondActivityTaskCompletedRequest taskCompleted = response.getTaskCompleted();
         Optional<Payloads> result =
             taskCompleted.hasResult() ? Optional.of(taskCompleted.getResult()) : Optional.empty();
         return dataConverter.fromPayloads(0, result, resultClass, resultType);
-      } else {
+      } else if (response.getTaskFailed() != null) {
         RespondActivityTaskFailedRequest taskFailed =
             response.getTaskFailed().getTaskFailedRequest();
-        if (taskFailed != null) {
-          Exception cause = dataConverter.failureToException(taskFailed.getFailure());
-          throw new ActivityFailure(
-              taskFailed.getFailure().getMessage(),
-              0,
-              0,
-              task.getActivityType().getName(),
-              task.getActivityId(),
-              RetryState.RETRY_STATE_NON_RETRYABLE_FAILURE,
-              "TestActivityEnvironment",
-              cause);
-        } else {
-          RespondActivityTaskCanceledRequest taskCanceled = response.getTaskCanceled();
-          if (taskCanceled != null) {
-            throw new CanceledFailure(
-                "canceled",
-                new EncodedValues(
-                    taskCanceled.hasDetails()
-                        ? Optional.of(taskCanceled.getDetails())
-                        : Optional.empty(),
-                    dataConverter),
-                null);
-          }
-        }
+        Exception cause = dataConverter.failureToException(taskFailed.getFailure());
+        throw new ActivityFailure(
+            taskFailed.getFailure().getMessage(),
+            0,
+            0,
+            task.getActivityType().getName(),
+            task.getActivityId(),
+            RetryState.RETRY_STATE_NON_RETRYABLE_FAILURE,
+            "TestActivityEnvironment",
+            cause);
+      } else if (response.getTaskCanceled() != null) {
+        RespondActivityTaskCanceledRequest taskCanceled = response.getTaskCanceled();
+        throw new CanceledFailure(
+            "canceled",
+            new EncodedValues(
+                taskCanceled.hasDetails()
+                    ? Optional.of(taskCanceled.getDetails())
+                    : Optional.empty(),
+                dataConverter),
+            null);
+      } else {
+        throw new ActivityRequestedAsyncCompletion(
+            task.getActivityId(), response.isManualCompletion());
       }
-      return Defaults.defaultValue(resultClass);
     }
 
     @Override

Original file line number	Diff line number	Diff line change
`@@ -449,7 +449,7 @@ private WorkflowThread newRootThread(Runnable runnable) {`
`449`	`449`	`"newRootThread can be called only if there is no existing root workflow thread");`
`450`	`450`	`}`
`451`	`451`	`rootWorkflowThread =`
`452`		`- new WorkflowThreadImpl(`
	`452`	`+ new RootWorkflowThreadImpl(`
`453`	`453`	`workflowThreadExecutor,`
`454`	`454`	`workflowContext,`
`455`	`455`	`this,`
Original file line number	Diff line number	Diff line change
`@@ -34,7 +34,9 @@`
`34`	`34`	`* thrown by the constructor are treated the same as exceptions thrown by the workflow method.`
`35`	`35`	`*`
`36`	`36`	`* <p>Workflow initialization methods are called before the workflow method, signal handlers, update`
`37`		`- * handlers or query handlers.`
	`37`	`+ * handlers or query handlers. Users should be careful not to block in constructors. Blocking in the`
	`38`	`+ * constructor will delay handling of the workflow method and signal handlers, and cause rejection`
	`39`	`+ * of the update and query handlers.`
`38`	`40`	`*`
`39`	`41`	`* <p>This annotation applies only to workflow implementation constructors.`
`40`	`42`	`*/`
Original file line number	Diff line number	Diff line change
`@@ -76,10 +76,11 @@ public void deserializeWorkflowRunToken() throws IOException {`
`76`	`76`	`}`
`77`	`77`
`78`	`78`	`@Test`
`79`		`- public void loadOldWorkflowRunToken() {`
	`79`	`+ public void failLoadOldWorkflowRunToken() {`
`80`	`80`	`String operationToken = "AAAAA-BBBBB-CCCCC";`
`81`		`- Assert.assertEquals(`
`82`		`- operationToken, OperationTokenUtil.loadWorkflowIdFromOperationToken(operationToken));`
	`81`	`+ Assert.assertThrows(`
	`82`	`+ IllegalArgumentException.class,`
	`83`	`+ () -> OperationTokenUtil.loadWorkflowIdFromOperationToken(operationToken));`
`83`	`84`	`}`
`84`	`85`
`85`	`86`	`@Test`