Allow Data Converter code to escape deadlock detection (#1723)

Issue #1301

Author: Spikhalskiy

build.gradle

@@ -9,7 +9,7 @@ plugins {
     id 'net.ltgt.errorprone' version '3.0.1' apply false
     id 'org.cadixdev.licenser' version '0.6.1'
     id 'com.palantir.git-version' version "${palantirGitVersionVersion}" apply false
-    id 'io.github.gradle-nexus.publish-plugin' version '1.2.0'
+    id 'io.github.gradle-nexus.publish-plugin' version '1.3.0'
     id 'com.diffplug.spotless' version '6.17.0' apply false
     id 'com.github.nbaztec.coveralls-jacoco' version "1.2.15" apply false
 
@@ -30,7 +30,7 @@ allprojects {
 
 ext {
     // Platforms
-    grpcVersion = '1.53.0' // [1.38.0,) Needed for io.grpc.protobuf.services.HealthStatusManager
+    grpcVersion = '1.54.0' // [1.38.0,) Needed for io.grpc.protobuf.services.HealthStatusManager
     jacksonVersion = '2.14.2' // [2.9.0,)
     // we don't upgrade to 1.10.x because it requires kotlin 1.6. Users may use 1.10.x in their environments though.
     micrometerVersion = project.hasProperty("edgeDepsTest") ? '1.10.5' : '1.9.9' // [1.0.0,)
@@ -52,7 +52,7 @@ ext {
 
     jsonPathVersion = '2.7.0' // compileOnly
 
-    cronUtilsVersion = '9.2.0' // for test server only
+    cronUtilsVersion = '9.2.1' // for test server only
 
     // Spring Boot 3 requires Java 17, java-sdk builds against 2.x version because we support Java 8.
     // We do test compatibility with Spring Boot 3 in integration tests.

temporal-sdk/src/test/java/io/temporal/client/functional/common/TestWorkflows.java renamed to temporal-sdk/src/main/java/io/temporal/internal/common/NonIdempotentHandle.java

@@ -18,15 +18,15 @@
  * limitations under the License.
  */
 
-package io.temporal.client.functional.common;
+package io.temporal.internal.common;
 
-import io.temporal.workflow.WorkflowInterface;
-import io.temporal.workflow.WorkflowMethod;
-
-public class TestWorkflows {
-  @WorkflowInterface
-  public interface PrimitiveWorkflow {
-    @WorkflowMethod
-    void execute();
-  }
+/**
+ * This interface signifies that the implementation of {@link #close()} method may and likely is not
+ * idempotent, which is in agreement with {@link AutoCloseable#close()} contract. It also narrows
+ * {@link AutoCloseable#close()} contract to not throw any checked exceptions, making it more
+ * convenient to use in try-with-resources blocks.
+ */
+public interface NonIdempotentHandle extends AutoCloseable {
+  @Override
+  void close();
 }

temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java

@@ -38,7 +38,9 @@
 import io.temporal.common.metadata.POJOWorkflowImplMetadata;
 import io.temporal.common.metadata.POJOWorkflowInterfaceMetadata;
 import io.temporal.common.metadata.POJOWorkflowMethodMetadata;
+import io.temporal.internal.WorkflowThreadMarker;
 import io.temporal.internal.common.ActivityOptionUtils;
+import io.temporal.internal.common.NonIdempotentHandle;
 import io.temporal.internal.common.SearchAttributesUtil;
 import io.temporal.internal.logging.ReplayAwareLogger;
 import io.temporal.serviceclient.CheckedExceptionWrapper;
@@ -395,14 +397,6 @@ public static <R> R executeActivity(
     return result.get();
   }
 
-  private static WorkflowOutboundCallsInterceptor getWorkflowOutboundInterceptor() {
-    return getRootWorkflowContext().getWorkflowOutboundInterceptor();
-  }
-
-  static SyncWorkflowContext getRootWorkflowContext() {
-    return DeterministicRunnerImpl.currentThreadInternal().getWorkflowContext();
-  }
-
   public static void await(String reason, Supplier<Boolean> unblockCondition)
       throws DestroyWorkflowThreadError {
     getWorkflowOutboundInterceptor().await(reason, unblockCondition);
@@ -464,19 +458,12 @@ public static Throwable unwrap(Throwable e) {
     return CheckedExceptionWrapper.unwrap(e);
   }
 
-  /** Prohibit instantiation */
-  private WorkflowInternal() {}
-
   /** Returns false if not under workflow code. */
   public static boolean isReplaying() {
     Optional<WorkflowThread> thread = DeterministicRunnerImpl.currentThreadInternalIfPresent();
     return thread.isPresent() && getRootWorkflowContext().isReplaying();
   }
 
-  public static WorkflowInfo getWorkflowInfo() {
-    return new WorkflowInfoImpl(getRootWorkflowContext().getReplayContext());
-  }
-
   public static <T> T getMemo(String key, Class<T> valueClass, Type genericType) {
     Payload memo = getRootWorkflowContext().getReplayContext().getMemo(key);
     if (memo == null) {
@@ -520,6 +507,24 @@ public static void sleep(Duration duration) {
     getWorkflowOutboundInterceptor().sleep(duration);
   }
 
+  public static boolean isWorkflowThread() {
+    return WorkflowThreadMarker.isWorkflowThread();
+  }
+
+  public static <T> T deadlockDetectorOff(Functions.Func<T> func) {
+    if (isWorkflowThread()) {
+      try (NonIdempotentHandle ignored = getWorkflowThread().lockDeadlockDetector()) {
+        return func.apply();
+      }
+    } else {
+      return func.apply();
+    }
+  }
+
+  public static WorkflowInfo getWorkflowInfo() {
+    return new WorkflowInfoImpl(getRootWorkflowContext().getReplayContext());
+  }
+
   public static Scope getMetricsScope() {
     return getRootWorkflowContext().getMetricsScope();
   }
@@ -614,4 +619,19 @@ public static Optional<Exception> getPreviousRunFailure() {
         // converter
         .map(f -> getDataConverter().failureToException(f));
   }
+
+  private static WorkflowOutboundCallsInterceptor getWorkflowOutboundInterceptor() {
+    return getRootWorkflowContext().getWorkflowOutboundInterceptor();
+  }
+
+  static SyncWorkflowContext getRootWorkflowContext() {
+    return DeterministicRunnerImpl.currentThreadInternal().getWorkflowContext();
+  }
+
+  private static WorkflowThread getWorkflowThread() {
+    return DeterministicRunnerImpl.currentThreadInternal();
+  }
+
+  /** Prohibit instantiation */
+  private WorkflowInternal() {}
 }

temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThread.java

@@ -23,6 +23,7 @@
 import static io.temporal.internal.sync.DeterministicRunnerImpl.currentThreadInternal;
 
 import io.temporal.failure.CanceledFailure;
+import io.temporal.internal.common.NonIdempotentHandle;
 import io.temporal.workflow.CancellationScope;
 import java.util.Optional;
 import java.util.concurrent.Future;
@@ -90,6 +91,13 @@ static WorkflowThread newThread(Runnable runnable, boolean detached, String name
    */
   boolean runUntilBlocked(long deadlockDetectionTimeoutMs);
 
+  /**
+   * Disables deadlock detector on this thread
+   *
+   * @return a handle that must be used to unlock the deadlock detector back
+   */
+  NonIdempotentHandle lockDeadlockDetector();
+
   Throwable getUnhandledException();
 
   boolean isDone();