💥 Support Kotlin 2.4 async method references

CHANGELOG.md

@@ -24,6 +24,11 @@ to docs, or any other relevant information.
   notes is kept for historical reference.
 
 ### Fixed
+- Kotlin `Async` method references compiled with Kotlin 2.4 now retain their inline behavior,
+  preventing `NonDeterministicException` when replaying workflows started with Kotlin 2.2. For
+  workflows already started with Kotlin 2.4 and SDK 1.40 or earlier, set the
+  `temporal.kotlin.disableStaticAdapterUnwrapping` JVM system property to preserve the previous
+  behavior during replay.
 - Test server now honors retry expiration deadlines that fall exactly on a whole second. Previously such deadlines were
   ignored and retries were scheduled past them instead of failing with `RETRY_STATE_TIMEOUT`.
 

settings.gradle

@@ -1,3 +1,15 @@
+pluginManagement {
+    resolutionStrategy {
+        eachPlugin {
+            // Exercise Kotlin 2.4 static adapters in the edge test job.
+            if (requested.id.id == 'org.jetbrains.kotlin.jvm' &&
+                    providers.gradleProperty('edgeDepsTest').isPresent()) {
+                useVersion('2.4.20')
+            }
+        }
+    }
+}
+
 rootProject.name='temporal-java-sdk'
 include 'temporal-bom'
 include 'temporal-serviceclient'

temporal-kotlin/README.md

@@ -21,6 +21,20 @@ or to build.gradle:
 compile group: 'io.temporal', name: 'temporal-kotlin', version: 'N.N.N'
 ```
 
+Kotlin 2.4 compiles method references passed to `Async.function` and `Async.procedure` differently
+from Kotlin 2.2. SDK versions through 1.40 do not recognize the Kotlin 2.4 form as a stub method
+reference, so they run it on a new workflow thread. This can change command ordering and cause a
+`NonDeterministicException` when replaying a workflow started with Kotlin 2.2 after upgrading to
+Kotlin 2.4. This module now recognizes the Kotlin 2.4 form and runs it inline, as it did with
+Kotlin 2.2.
+
+If a workflow was already started with Kotlin 2.4 and SDK 1.40 or earlier, the new behavior can
+also change its command ordering during replay. To preserve the old behavior while those workflows
+are still running, set the `temporal.kotlin.disableStaticAdapterUnwrapping` JVM system property,
+for example, `-Dtemporal.kotlin.disableStaticAdapterUnwrapping`. The property's presence disables
+the new unwrapping. It applies to all workflows in the worker, so it can reintroduce the replay
+failure for histories started with Kotlin 2.2.
+
 ## Kotlin extensions
 
 This module adds several Kotlin extensions to make Kotlin code that uses Temporal Java SDK a bit

temporal-kotlin/build.gradle

@@ -12,7 +12,7 @@ tasks.withType(KotlinCompile).all {
     kotlinOptions {
         freeCompilerArgs += "-Xopt-in=kotlin.RequiresOptIn"
         jvmTarget = project.hasProperty("edgeDepsTest") ? JavaVersion.VERSION_21 : JavaVersion.VERSION_1_8
-        languageVersion = "${project.hasProperty("edgeDepsTest") ? '1.8' : '1.5'}"
+        languageVersion = "${project.hasProperty("edgeDepsTest") ? '2.4' : '1.5'}"
     }
 }
 
@@ -42,4 +42,3 @@ task registerNamespace(type: JavaExec) {
 }
 
 test.dependsOn 'registerNamespace'
-

temporal-kotlin/src/main/kotlin/io/temporal/internal/async/KotlinMethodReferenceDisassemblyService.kt

@@ -2,20 +2,25 @@
 package io.temporal.internal.async
 
 import io.temporal.internal.async.spi.MethodReferenceDisassemblyService
+import io.temporal.internal.common.JavaLambdaUtils
+import io.temporal.internal.common.kotlin.KotlinDetector
 import io.temporal.workflow.Functions
 import kotlin.jvm.internal.CallableReference
 import kotlin.jvm.internal.Lambda
 
 class KotlinMethodReferenceDisassemblyService : MethodReferenceDisassemblyService {
   override fun getMethodReferenceTarget(methodReference: Any): Any? {
-    return when (methodReference) {
-      is Functions.TemporalFunctionalInterfaceMarker -> {
-        unwrapTemporalFunctionalInterfaceInKotlin(methodReference)
-      }
-      else -> {
-        unwrapIfCallableReference(methodReference)
+    val callableTarget = unwrapIfCallableReference(methodReference)
+    if (callableTarget != null) {
+      return callableTarget
+    }
+    if (methodReference is Functions.TemporalFunctionalInterfaceMarker) {
+      val wrappedTarget = unwrapTemporalFunctionalInterfaceInKotlin(methodReference)
+      if (wrappedTarget != null) {
+        return wrappedTarget
       }
     }
+    return unwrapKotlinStaticAdapter(methodReference)
   }
 
   private fun unwrapIfCallableReference(callableReference: Any): Any? {
@@ -61,7 +66,7 @@ class KotlinMethodReferenceDisassemblyService : MethodReferenceDisassemblyServic
        * We also end up here in any version of Kotlin if wrapped lambda is passed and this case is handled in unwrapCallableReference
        */
       return unwrapIfCallableReference(proxiedValue)
-    } else {
+    } else if (KotlinDetector.isKotlinType(temporalFunction.javaClass)) {
       /**
        * Strategy 2.2
        * Kotlin 1.5 generates one of [io.temporal.workflow.Functions] directly over the target
@@ -70,6 +75,23 @@ class KotlinMethodReferenceDisassemblyService : MethodReferenceDisassemblyServic
        */
       return proxiedValue
     }
+    return null
+  }
+
+  /**
+   * Strategy 3
+   * Kotlin 2.4 uses a serialized lambda with a static adapter for a method reference.
+   * A lambda that captures the same target also has one field, so verify the adapter's method.
+   */
+  private fun unwrapKotlinStaticAdapter(methodReference: Any): Any? {
+    if (System.getProperty("temporal.kotlin.disableStaticAdapterUnwrapping") != null) {
+      return null
+    }
+    if (methodReference !is Functions.TemporalFunctionalInterfaceMarker) {
+      return null
+    }
+    val serializedLambda = JavaLambdaUtils.toSerializedLambda(methodReference) ?: return null
+    return KotlinDetector.getKotlinStaticAdapterTarget(serializedLambda, methodReference.javaClass.classLoader)
   }
 
   override fun getLanguageName(): String {

temporal-kotlin/src/test/kotlin/io/temporal/internal/async/KotlinAsyncMarkerTest.kt

@@ -0,0 +1,69 @@
+package io.temporal.internal.async
+
+import io.temporal.internal.async.spi.MethodReferenceDisassemblyService
+import io.temporal.internal.sync.AsyncInternal
+import io.temporal.workflow.Functions
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+class KotlinAsyncMarkerTest {
+  private interface Stub {
+    fun zero(): Int
+    fun execute(value: Int): Int
+    fun fire()
+    fun fireWith(value: Int)
+  }
+
+  private class MarkedStub : Stub, AsyncInternal.AsyncMarker {
+    override fun zero() = 0
+    override fun execute(value: Int) = value
+    override fun fire() {}
+    override fun fireWith(value: Int) {}
+  }
+
+  private fun <R> asFunc(function: Functions.Func<R>): Functions.Func<R> = function
+
+  private fun <A, R> asFunc1(function: Functions.Func1<A, R>): Functions.Func1<A, R> = function
+
+  private fun asProc(procedure: Functions.Proc): Functions.Proc = procedure
+
+  private fun <A> asProc1(procedure: Functions.Proc1<A>): Functions.Proc1<A> = procedure
+
+  @Test
+  fun methodReferencesToMarkedStubsAreAsync() {
+    val stub: Stub = MarkedStub()
+
+    assertTrue(AsyncInternal.isAsync(stub::zero))
+    assertTrue(AsyncInternal.isAsync(asFunc(stub::zero)))
+    assertTrue(AsyncInternal.isAsync(asFunc1(stub::execute)))
+    assertTrue(AsyncInternal.isAsync(asProc(stub::fire)))
+    assertTrue(AsyncInternal.isAsync(asProc1(stub::fireWith)))
+  }
+
+  @Test
+  fun lambdasCapturingMarkedStubsAreNotAsync() {
+    val stub: Stub = MarkedStub()
+
+    assertFalse(AsyncInternal.isAsync(asFunc { stub.zero() }))
+    assertFalse(AsyncInternal.isAsync(asFunc1<Int, Int> { value -> stub.execute(value) }))
+    assertFalse(AsyncInternal.isAsync(asProc { stub.fire() }))
+    assertFalse(AsyncInternal.isAsync(asProc1<Int> { value -> stub.fireWith(value) }))
+  }
+
+  @Test
+  fun kotlinLambdasDoNotRequireKotlinDisassemblyService() {
+    val kotlinService = MethodReferenceDisassembler.services.remove(MethodReferenceDisassemblyService.KOTLIN)
+    try {
+      val stub: Stub = MarkedStub()
+      assertFalse(AsyncInternal.isAsync(asFunc { stub.zero() }))
+      assertFalse(AsyncInternal.isAsync(asFunc1<Int, Int> { value -> stub.execute(value) }))
+      assertFalse(AsyncInternal.isAsync(asProc { stub.fire() }))
+      assertFalse(AsyncInternal.isAsync(asProc1<Int> { value -> stub.fireWith(value) }))
+    } finally {
+      if (kotlinService != null) {
+        MethodReferenceDisassembler.services[MethodReferenceDisassemblyService.KOTLIN] = kotlinService
+      }
+    }
+  }
+}

temporal-sdk/src/main/java/io/temporal/internal/async/MethodReferenceDisassembler.java

@@ -38,7 +38,9 @@ private static boolean isAsyncJava(Object func) {
   }
 
   private static boolean isAsyncKotlin(Object func) {
-    if (KotlinDetector.isKotlinType(func.getClass())) {
+    // Kotlin 2.4 SAM adapters have no Kotlin metadata on their generated class.
+    if (KotlinDetector.isKotlinType(func.getClass())
+        || KotlinDetector.isKotlinStaticAdapter(func)) {
       MethodReferenceDisassemblyService methodReferenceDisassemblyService =
           services.get(MethodReferenceDisassemblyService.KOTLIN);
       if (methodReferenceDisassemblyService == null) {

temporal-sdk/src/main/java/io/temporal/internal/common/kotlin/KotlinDetector.java

@@ -1,8 +1,14 @@
 package io.temporal.internal.common.kotlin;
 
+import io.temporal.internal.common.JavaLambdaUtils;
 import java.lang.annotation.Annotation;
+import java.lang.invoke.MethodHandleInfo;
+import java.lang.invoke.MethodType;
+import java.lang.invoke.SerializedLambda;
+import java.lang.reflect.Method;
+import java.util.Arrays;
 
-/** This class allows checking if the class is Kotlin class without using any Kotlin dependencies */
+/** Detects Kotlin classes and static method-reference adapters without a Kotlin dependency. */
 @SuppressWarnings("unchecked")
 public abstract class KotlinDetector {
 
@@ -57,4 +63,59 @@ public static boolean isKotlinReflectPresent() {
   public static boolean isKotlinType(Class<?> clazz) {
     return (kotlinMetadata != null && clazz.getDeclaredAnnotation(kotlinMetadata) != null);
   }
+
+  /** Determine whether the given lambda is a Kotlin static method-reference adapter. */
+  public static boolean isKotlinStaticAdapter(Object func) {
+    SerializedLambda lambda = JavaLambdaUtils.toSerializedLambda(func);
+    ClassLoader classLoader = func.getClass().getClassLoader();
+    if (getKotlinStaticAdapterTarget(lambda, classLoader) == null) {
+      return false;
+    }
+    try {
+      Class<?> capturingClass =
+          Class.forName(lambda.getCapturingClass().replace('/', '.'), false, classLoader);
+      return isKotlinType(capturingClass);
+    } catch (ClassNotFoundException e) {
+      return false;
+    }
+  }
+
+  /** Returns the target of a Kotlin static method-reference adapter, if the lambda has one. */
+  public static Object getKotlinStaticAdapterTarget(
+      SerializedLambda lambda, ClassLoader classLoader) {
+    if (lambda == null
+        || lambda.getImplMethodKind() != MethodHandleInfo.REF_invokeStatic
+        || lambda.getCapturedArgCount() != 1) {
+      return null;
+    }
+    Object target = JavaLambdaUtils.getTarget(lambda);
+    String adapterName = lambda.getImplMethodName();
+    int separator = adapterName.lastIndexOf('$');
+    if (target == null || separator < 0 || separator == adapterName.length() - 1) {
+      return null;
+    }
+
+    MethodType adapterType;
+    try {
+      adapterType =
+          MethodType.fromMethodDescriptorString(lambda.getImplMethodSignature(), classLoader);
+    } catch (IllegalArgumentException | TypeNotPresentException e) {
+      return null;
+    }
+    Class<?>[] adapterParameters = adapterType.parameterArray();
+    if (adapterParameters.length == 0 || !adapterParameters[0].isInstance(target)) {
+      return null;
+    }
+    String methodName = adapterName.substring(separator + 1);
+    Class<?>[] methodParameters =
+        Arrays.copyOfRange(adapterParameters, 1, adapterParameters.length);
+    for (Method method : target.getClass().getMethods()) {
+      if (method.getName().equals(methodName)
+          && method.getReturnType() == adapterType.returnType()
+          && Arrays.equals(method.getParameterTypes(), methodParameters)) {
+        return target;
+      }
+    }
+    return null;
+  }
 }