From 1c7010fff944c716ee6402a63e80126563515579 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Mon, 5 Oct 2026 09:11:40 -0700 Subject: [PATCH 1/4] Update API protos to v1.63.8 Pin the API submodule to the v1.63.8 release (1a8f67faead00aceb8ad90a3dd16b04e73415648). Keep the API dependency update separate from the SDK runtime reporting changes. --- temporal-serviceclient/src/main/proto | 2 +- .../payload/limits/gen/PayloadLimitValidatorGenerator.java | 2 ++ .../internal/payload/limits/PayloadLimitValidatorTest.java | 6 ++++++ 3 files changed, 9 insertions(+), 1 deletion(-) diff --git a/temporal-serviceclient/src/main/proto b/temporal-serviceclient/src/main/proto index 3ebdff42a9..1a8f67faea 160000 --- a/temporal-serviceclient/src/main/proto +++ b/temporal-serviceclient/src/main/proto @@ -1 +1 @@ -Subproject commit 3ebdff42a9f07ac484b415fe8ff0b483b4ce3340 +Subproject commit 1a8f67faead00aceb8ad90a3dd16b04e73415648 diff --git a/temporal-serviceclient/src/payloadGenerators/java/io/temporal/internal/payload/limits/gen/PayloadLimitValidatorGenerator.java b/temporal-serviceclient/src/payloadGenerators/java/io/temporal/internal/payload/limits/gen/PayloadLimitValidatorGenerator.java index 4802f21e98..b57e52554d 100644 --- a/temporal-serviceclient/src/payloadGenerators/java/io/temporal/internal/payload/limits/gen/PayloadLimitValidatorGenerator.java +++ b/temporal-serviceclient/src/payloadGenerators/java/io/temporal/internal/payload/limits/gen/PayloadLimitValidatorGenerator.java @@ -173,6 +173,8 @@ public final class PayloadLimitValidatorGenerator { "temporal.api.sdk.v1.UserMetadata.details", "temporal.api.sdk.v1.UserMetadata.summary", "temporal.api.nexus.v1.EndpointSpec.description", + // Callback context has dedicated per-callback and aggregate limits, not blob/memo limits. + "temporal.api.common.v1.Callback.NexusHandler.source_context", // Event group marker label: custom 400-byte server-side limit, not blob/memo. "temporal.api.sdk.v1.EventGroupMarker.Label.label", // Update input args: frontend records a metric only — enforced on delivery via Message.body. diff --git a/temporal-serviceclient/src/test/java/io/temporal/internal/payload/limits/PayloadLimitValidatorTest.java b/temporal-serviceclient/src/test/java/io/temporal/internal/payload/limits/PayloadLimitValidatorTest.java index 2d63f5d6dd..a9c283dbde 100644 --- a/temporal-serviceclient/src/test/java/io/temporal/internal/payload/limits/PayloadLimitValidatorTest.java +++ b/temporal-serviceclient/src/test/java/io/temporal/internal/payload/limits/PayloadLimitValidatorTest.java @@ -14,6 +14,7 @@ import io.temporal.api.command.v1.RecordMarkerCommandAttributes; import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributes; import io.temporal.api.command.v1.ScheduleNexusOperationCommandAttributes; +import io.temporal.api.common.v1.Callback; import io.temporal.api.common.v1.Header; import io.temporal.api.common.v1.Memo; import io.temporal.api.common.v1.Payload; @@ -315,6 +316,11 @@ public void startWorkflowVisitsOnlyItsValidatedFields() { UserMetadata.newBuilder() .setSummary(payload(1)) .setDetails(payload(1))) // dedicated non-fetchable limits + .addCompletionCallbacks( + Callback.newBuilder() + .setNexusHandler( + Callback.NexusHandler.newBuilder() + .setSourceContext(payload(1)))) // dedicated callback limits .build(); RecordingSink sink = new RecordingSink(); GeneratedPayloadLimitValidator.dispatch(sink, req); From 1bd193e7c80b0cda590d7ca6b321ce2f9d807ff1 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Mon, 5 Oct 2026 09:11:40 -0700 Subject: [PATCH 2/4] Report Kotlin and Graal runtimes in worker heartbeats Detect directly registered Kotlin workflow and activity implementations per worker and include the Kotlin version alongside the JVM runtime. Report Graal native-image execution when detected. Keep detection best effort and preserve telemetry opt-out and heartbeat acceptance behavior. --- CHANGELOG.md | 2 + .../worker/JavaHeartbeatImplementations.java | 31 +++ .../KotlinWorkerHeartbeatIntegrationTest.kt | 144 +++++++++++++ .../worker/KotlinWorkerHeartbeatTest.kt | 161 ++++++++++++++ .../client/WorkflowClientOptions.java | 7 +- .../worker/WorkerEnvironmentInfo.java | 102 ++++++++- .../main/java/io/temporal/worker/Worker.java | 22 +- .../worker/WorkerEnvironmentInfoTest.java | 200 +++++++++++++++++- 8 files changed, 655 insertions(+), 14 deletions(-) create mode 100644 temporal-kotlin/src/test/java/io/temporal/worker/JavaHeartbeatImplementations.java create mode 100644 temporal-kotlin/src/test/kotlin/io/temporal/worker/KotlinWorkerHeartbeatIntegrationTest.kt create mode 100644 temporal-kotlin/src/test/kotlin/io/temporal/worker/KotlinWorkerHeartbeatTest.kt diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a35074873..2f3e17b042 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,8 @@ to docs, or any other relevant information. applies this conversion to workflow, activity, Nexus, and schedule payloads, including failure details. `WorkflowClient.getOptions().getDataConverter()` returns the transfer-aware converter that the client uses. +- Worker heartbeats now report detected Graal native-image and Kotlin runtimes alongside the JVM + version. Detection is best effort and respects the worker environment information opt-out. - Added experimental `ChildWorkflowOptions.Builder.setVersioningOverride` and `VersioningOverride.OneTimeVersioningOverride` for explicit pinned, auto-upgrade, and one-time child workflow routing. Invalid child overrides are reported as `InvalidVersioningOverrideFailure` diff --git a/temporal-kotlin/src/test/java/io/temporal/worker/JavaHeartbeatImplementations.java b/temporal-kotlin/src/test/java/io/temporal/worker/JavaHeartbeatImplementations.java new file mode 100644 index 0000000000..e0363c676e --- /dev/null +++ b/temporal-kotlin/src/test/java/io/temporal/worker/JavaHeartbeatImplementations.java @@ -0,0 +1,31 @@ +package io.temporal.worker; + +import io.temporal.activity.ActivityInterface; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +/** Java implementations without Kotlin metadata, even when Kotlin is on the classpath. */ +public final class JavaHeartbeatImplementations { + private JavaHeartbeatImplementations() {} + + @WorkflowInterface + public interface JavaWorkflow { + @WorkflowMethod + void execute(); + } + + public static class JavaWorkflowImpl implements JavaWorkflow { + @Override + public void execute() {} + } + + @ActivityInterface + public interface JavaActivity { + void executeActivity(); + } + + public static class JavaActivityImpl implements JavaActivity { + @Override + public void executeActivity() {} + } +} diff --git a/temporal-kotlin/src/test/kotlin/io/temporal/worker/KotlinWorkerHeartbeatIntegrationTest.kt b/temporal-kotlin/src/test/kotlin/io/temporal/worker/KotlinWorkerHeartbeatIntegrationTest.kt new file mode 100644 index 0000000000..e00daf74da --- /dev/null +++ b/temporal-kotlin/src/test/kotlin/io/temporal/worker/KotlinWorkerHeartbeatIntegrationTest.kt @@ -0,0 +1,144 @@ +package io.temporal.worker + +import io.grpc.Status +import io.grpc.StatusRuntimeException +import io.temporal.api.worker.v1.EnvironmentInfo.Runtime.RuntimeType +import io.temporal.api.worker.v1.WorkerHeartbeat +import io.temporal.api.workflowservice.v1.DescribeNamespaceRequest +import io.temporal.api.workflowservice.v1.DescribeWorkerRequest +import io.temporal.api.workflowservice.v1.ListWorkersRequest +import io.temporal.client.WorkflowClientOptions +import io.temporal.testing.internal.SDKTestWorkflowRule +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNotNull +import org.junit.Assume.assumeTrue +import org.junit.Before +import org.junit.Rule +import org.junit.Test +import java.time.Duration + +class KotlinWorkerHeartbeatIntegrationTest { + @Rule + @JvmField + val testWorkflowRule = SDKTestWorkflowRule.newBuilder() + .setTestTimeoutSeconds(20) + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder() + .setWorkerHeartbeatInterval(Duration.ofSeconds(1)) + .build() + ) + .setWorkflowTypes(KotlinWorkerHeartbeatTest.KotlinWorkflowImpl::class.java) + .setDoNotStart(true) + .build() + + private val client get() = testWorkflowRule.workflowClient + private val service get() = client.workflowServiceStubs.blockingStub() + private val namespace get() = client.options.namespace + + @Before + fun checkServerSupportsHeartbeats() { + assumeTrue( + "Requires real server with worker heartbeat support", + SDKTestWorkflowRule.useExternalService + ) + assumeTrue( + "Server does not support worker heartbeats", + service.describeNamespace( + DescribeNamespaceRequest.newBuilder().setNamespace(namespace).build() + ).namespaceInfo.capabilities.workerHeartbeats + ) + } + + @Test + fun reportsRuntimeVersionsForKotlinWorkflowAndActivityButNotJava() { + val workflowQueue = testWorkflowRule.taskQueue + val activityQueue = "$workflowQueue-kotlin-activity" + val javaQueue = "$workflowQueue-java" + val environment = testWorkflowRule.testEnvironment + environment.newWorker(activityQueue) + .registerActivitiesImplementations(KotlinWorkerHeartbeatTest.KotlinActivityImpl()) + val javaWorker = environment.newWorker(javaQueue) + javaWorker.registerWorkflowImplementationTypes(JavaHeartbeatImplementations.JavaWorkflowImpl::class.java) + javaWorker.registerActivitiesImplementations(JavaHeartbeatImplementations.JavaActivityImpl()) + environment.start() + + assertEventually(Duration.ofSeconds(10)) { + assertServerRuntimes(workflowQueue, kotlin = true) + assertServerRuntimes(activityQueue, kotlin = true) + assertServerRuntimes(javaQueue, kotlin = false) + } + } + + private fun assertServerRuntimes(taskQueue: String, kotlin: Boolean) { + val workers = listWorkersForQueue(taskQueue) + assertFalse("No workers found via ListWorkers for queue: $taskQueue", workers.isEmpty()) + val listed = workers.single() + assertFalse("Worker instance key should not be empty", listed.workerInstanceKey.isEmpty()) + val described = describeWorker(listed.workerInstanceKey) + assertNotNull("DescribeWorker should return a stored heartbeat", described) + assertEquals(taskQueue, described!!.taskQueue) + assertEquals(listed.workerInstanceKey, described.workerInstanceKey) + assertRuntimes(described, kotlin) + } + + private fun assertRuntimes(heartbeat: WorkerHeartbeat, kotlin: Boolean) { + val expected = mutableMapOf(RuntimeType.RUNTIME_TYPE_JVM to System.getProperty("java.version")) + if (kotlin) { + expected[RuntimeType.RUNTIME_TYPE_KOTLIN] = KotlinVersion.CURRENT.toString() + } + val runtimes = heartbeat.environment.runtimesList + assertEquals("Runtime entries should not be duplicated", expected.size, runtimes.size) + assertEquals(expected, runtimes.associate { it.type to it.version }) + } + + // WorkersInfo is deprecated, but its replacement is not yet populated by the server. + @Suppress("DEPRECATION") + private fun listWorkersForQueue(taskQueue: String): List { + try { + return service.listWorkers( + ListWorkersRequest.newBuilder() + .setNamespace(namespace) + .setQuery("TaskQueue = \"$taskQueue\"") + .setPageSize(200) + .build() + ).workersInfoList.map { it.workerHeartbeat } + } catch (e: StatusRuntimeException) { + if (e.status.code == Status.Code.RESOURCE_EXHAUSTED) { + return emptyList() + } + throw e + } + } + + private fun describeWorker(workerInstanceKey: String): WorkerHeartbeat? { + try { + return service.describeWorker( + DescribeWorkerRequest.newBuilder() + .setNamespace(namespace) + .setWorkerInstanceKey(workerInstanceKey) + .build() + ).workerInfo.workerHeartbeat + } catch (e: StatusRuntimeException) { + if (e.status.code == Status.Code.NOT_FOUND || e.status.code == Status.Code.RESOURCE_EXHAUSTED) { + return null + } + throw e + } + } + + private fun assertEventually(timeout: Duration, assertion: () -> Unit) { + val deadline = System.nanoTime() + timeout.toNanos() + while (true) { + try { + assertion() + return + } catch (e: AssertionError) { + if (System.nanoTime() >= deadline) { + throw e + } + Thread.sleep(100) + } + } + } +} diff --git a/temporal-kotlin/src/test/kotlin/io/temporal/worker/KotlinWorkerHeartbeatTest.kt b/temporal-kotlin/src/test/kotlin/io/temporal/worker/KotlinWorkerHeartbeatTest.kt new file mode 100644 index 0000000000..7783656294 --- /dev/null +++ b/temporal-kotlin/src/test/kotlin/io/temporal/worker/KotlinWorkerHeartbeatTest.kt @@ -0,0 +1,161 @@ +package io.temporal.worker + +import io.temporal.activity.ActivityInterface +import io.temporal.api.worker.v1.EnvironmentInfo +import io.temporal.api.worker.v1.EnvironmentInfo.Runtime.RuntimeType +import io.temporal.client.WorkflowClientOptions +import io.temporal.testing.TestEnvironmentOptions +import io.temporal.testing.TestWorkflowEnvironment +import io.temporal.workflow.WorkflowInterface +import io.temporal.workflow.WorkflowMethod +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test +import java.time.Duration + +class KotlinWorkerHeartbeatTest { + private lateinit var testEnvironment: TestWorkflowEnvironment + + private val jvmRuntime = EnvironmentInfo.Runtime.newBuilder() + .setType(RuntimeType.RUNTIME_TYPE_JVM) + .setVersion("test-jvm-version") + .build() + private val environment = EnvironmentInfo.newBuilder() + .addRuntimes(jvmRuntime) + .build() + + @Before + fun setUp() { + // Exercise callbacks directly without a background heartbeat accepting the environment. + testEnvironment = TestWorkflowEnvironment.newInstance( + TestEnvironmentOptions.newBuilder() + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder() + .setWorkerHeartbeatInterval(Duration.ofSeconds(-1)) + .build() + ) + .build() + ) + } + + @After + fun tearDown() { + testEnvironment.close() + } + + @Test + fun kotlinWorkflowRegistrationReportsKotlin() { + val worker = testEnvironment.newWorker("kotlin-workflow") + worker.registerWorkflowImplementationTypes(KotlinWorkflowImpl::class.java) + testEnvironment.start() + + assertKotlinEnvironment(worker.buildHeartbeatCallback("group", environment).get().environment) + } + + @Test + fun kotlinWorkflowRegistrationWithOptionsReportsKotlin() { + val worker = testEnvironment.newWorker("kotlin-workflow-options") + worker.registerWorkflowImplementationTypes( + WorkflowImplementationOptions.newBuilder().build(), + KotlinWorkflowImpl::class.java + ) + testEnvironment.start() + + assertKotlinEnvironment(worker.buildHeartbeatCallback("group", environment).get().environment) + } + + @Test + fun kotlinActivityRegistrationReportsKotlinForLocalOnlyWorker() { + val worker = testEnvironment.newWorker( + "kotlin-local-activity", + WorkerOptions.newBuilder().setLocalActivityWorkerOnly(true).build() + ) + worker.registerActivitiesImplementations(KotlinActivityImpl()) + testEnvironment.start() + + assertKotlinEnvironment(worker.buildHeartbeatCallback("group", environment).get().environment) + } + + @Test + fun kotlinDetectionIsPerWorkerNotPerClasspathOrFactory() { + val kotlinWorker = testEnvironment.newWorker("kotlin") + kotlinWorker.registerWorkflowImplementationTypes(KotlinWorkflowImpl::class.java) + val javaWorker = testEnvironment.newWorker("java") + javaWorker.registerWorkflowImplementationTypes(JavaHeartbeatImplementations.JavaWorkflowImpl::class.java) + javaWorker.registerActivitiesImplementations(JavaHeartbeatImplementations.JavaActivityImpl()) + val unregisteredWorker = testEnvironment.newWorker("unregistered") + testEnvironment.start() + + assertKotlinEnvironment(kotlinWorker.buildHeartbeatCallback("group", environment).get().environment) + assertEquals(environment, javaWorker.buildHeartbeatCallback("group", environment).get().environment) + assertEquals(environment, unregisteredWorker.buildHeartbeatCallback("group", environment).get().environment) + } + + @Test + fun environmentIsRetriedUntilHeartbeatAccepted() { + val worker = testEnvironment.newWorker("retry") + worker.registerWorkflowImplementationTypes(KotlinWorkflowImpl::class.java) + testEnvironment.start() + val heartbeat = worker.buildHeartbeatCallback("group", environment) + + val first = heartbeat.get() + assertTrue(first.hasEnvironment()) + assertKotlinEnvironment(first.environment) + repeat(3) { + val retry = heartbeat.get() + assertTrue(retry.hasEnvironment()) + assertEquals(first.environment, retry.environment) + } + + worker.onHeartbeatAccepted() + repeat(3) { + assertFalse(heartbeat.get().hasEnvironment()) + } + } + + @Test + fun nullEnvironmentOptsOutEvenForKotlinImplementations() { + val worker = testEnvironment.newWorker("optout") + worker.registerWorkflowImplementationTypes(KotlinWorkflowImpl::class.java) + worker.registerActivitiesImplementations(KotlinActivityImpl()) + testEnvironment.start() + val heartbeat = worker.buildHeartbeatCallback("group", null) + + repeat(3) { + assertFalse(heartbeat.get().hasEnvironment()) + } + worker.onHeartbeatAccepted() + assertFalse(heartbeat.get().hasEnvironment()) + } + + private fun assertKotlinEnvironment(actual: EnvironmentInfo) { + assertEquals(2, actual.runtimesCount) + assertEquals(jvmRuntime, actual.runtimesList.single { it.type == RuntimeType.RUNTIME_TYPE_JVM }) + val kotlinRuntime = actual.runtimesList.single { it.type == RuntimeType.RUNTIME_TYPE_KOTLIN } + assertEquals(KotlinVersion.CURRENT.toString(), kotlinRuntime.version) + // Augmenting the heartbeat must not mutate the caller's environment. + assertEquals(listOf(jvmRuntime), environment.runtimesList) + } + + @WorkflowInterface + interface KotlinWorkflow { + @WorkflowMethod + fun execute() + } + + class KotlinWorkflowImpl : KotlinWorkflow { + override fun execute() {} + } + + @ActivityInterface + interface KotlinActivity { + fun executeActivity() + } + + class KotlinActivityImpl : KotlinActivity { + override fun executeActivity() {} + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java index 3015d20233..0450786b3d 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java @@ -190,9 +190,10 @@ public Builder setWorkerHeartbeatInterval(Duration workerHeartbeatInterval) { } /** - * Disables reporting the JVM version, detected hosting environments (Docker, Kubernetes, cloud - * platforms), and OS platform in worker heartbeats. This information is sent once per worker, - * with the first heartbeat accepted by the server. + * Disables reporting runtime versions (JVM, detected Graal native images, and Kotlin), detected + * hosting environments (Docker, Kubernetes, cloud platforms), and OS platform in worker + * heartbeats. This information is sent once per worker, with the first heartbeat accepted by + * the server. * * @param disableWorkerEnvironmentInfo true to omit environment information from heartbeats */ diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkerEnvironmentInfo.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkerEnvironmentInfo.java index cd33aad2fa..5e0277a1af 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkerEnvironmentInfo.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkerEnvironmentInfo.java @@ -11,6 +11,8 @@ import io.temporal.api.worker.v1.EnvironmentInfo.Runtime.RuntimeType; import io.temporal.api.worker.v1.EnvironmentInfo.WindowsPlatform; import java.io.IOException; +import java.lang.annotation.Annotation; +import java.lang.annotation.AnnotationFormatError; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; @@ -33,28 +35,110 @@ public final class WorkerEnvironmentInfo { private WorkerEnvironmentInfo() {} /** - * Never throws: this runs during client creation, and telemetry must not break it. System - * property, environment, and filesystem access can all fail under a security manager, in which - * case whatever was collected before the failure is returned. + * Best effort: this runs during client creation, and telemetry must not break it. Each probe is + * isolated so unavailable runtime information does not prevent platform or hosting detection. */ public static EnvironmentInfo detect() { + return detect(System::getProperty); + } + + static EnvironmentInfo detect(Function properties) { EnvironmentInfo.Builder builder = EnvironmentInfo.newBuilder(); + builder.addRuntimes( + Runtime.newBuilder() + .setType(RuntimeType.RUNTIME_TYPE_JVM) + .setVersion(propertyValue(properties, "java.version"))); + try { + if ("runtime".equals(properties.apply("org.graalvm.nativeimage.imagecode"))) { + builder.addRuntimes( + Runtime.newBuilder() + .setType(RuntimeType.RUNTIME_TYPE_GRAAL_AOT) + .setVersion(propertyValue(properties, "org.graalvm.version"))); + } + } catch (RuntimeException | LinkageError e) { + log.info("Failed to detect Graal native-image runtime", e); + } try { - builder.addRuntimes( - Runtime.newBuilder() - .setType(RuntimeType.RUNTIME_TYPE_JVM) - .setVersion(nullToEmpty(System.getProperty("java.version")))); builder.addAllHostingEnvironments(detectHostingEnvironments(System::getenv)); + } catch (RuntimeException | LinkageError e) { + log.info("Failed to detect worker hosting environments", e); + } + try { Platform platform = detectPlatform(); if (platform != null) { builder.setPlatform(platform); } - } catch (RuntimeException e) { - log.info("Failed to detect worker environment information, reporting partial results", e); + } catch (RuntimeException | LinkageError e) { + log.info("Failed to detect worker platform", e); } return builder.build(); } + private static String propertyValue(Function properties, String name) { + try { + return nullToEmpty(properties.apply(name)); + } catch (RuntimeException | LinkageError e) { + return ""; + } + } + + /** + * Detects Kotlin implementation classes without a Kotlin dependency or assuming a shared + * annotation class loader. + */ + public static boolean isKotlinImplementation(@Nullable Class implementationClass) { + if (implementationClass == null) { + return false; + } + try { + for (Annotation annotation : implementationClass.getDeclaredAnnotations()) { + if ("kotlin.Metadata".equals(annotation.annotationType().getName())) { + return true; + } + } + } catch (RuntimeException | LinkageError | AnnotationFormatError e) { + log.info("Failed to detect Kotlin implementation metadata", e); + } + return false; + } + + /** + * Returns immutable environment information augmented with Kotlin when the implementation has + * Kotlin metadata. A null environment (telemetry opt-out) is preserved, as are existing runtimes. + */ + @Nullable + public static EnvironmentInfo withKotlinRuntime( + @Nullable EnvironmentInfo info, @Nullable Class implementationClass) { + if (info == null) { + return null; + } + for (Runtime runtime : info.getRuntimesList()) { + if (runtime.getType() == RuntimeType.RUNTIME_TYPE_KOTLIN) { + return info; + } + } + if (!isKotlinImplementation(implementationClass)) { + return info; + } + return info.toBuilder() + .addRuntimes( + Runtime.newBuilder() + .setType(RuntimeType.RUNTIME_TYPE_KOTLIN) + .setVersion(kotlinVersion(implementationClass))) + .build(); + } + + private static String kotlinVersion(Class implementationClass) { + try { + Class version = + Class.forName("kotlin.KotlinVersion", true, implementationClass.getClassLoader()); + Object current = version.getField("CURRENT").get(null); + return current == null ? "" : current.toString(); + } catch (ReflectiveOperationException | RuntimeException | LinkageError e) { + return ""; + } + } + /** * Several environments may be detected at once, e.g. Docker inside Kubernetes or Azure Functions * inside Azure App Service. diff --git a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java index b1a5750078..6e29786265 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -84,6 +84,7 @@ public final class Worker { // Reported in every heartbeat (including the one embedded in ShutdownWorkerRequest) until the // server accepts one, then cleared so it is sent only once per worker. private final AtomicReference pendingEnvironmentInfo = new AtomicReference<>(); + private final AtomicReference> kotlinImplementationClass = new AtomicReference<>(); private static final class TaskSnapshot { final int processed; @@ -287,6 +288,7 @@ public void registerWorkflowImplementationTypes(Class... workflowImplementati workflowWorker.registerWorkflowImplementationTypes( WorkflowImplementationOptions.newBuilder().build(), workflowImplementationClasses); + recordKotlinImplementations(workflowImplementationClasses); } /** @@ -311,6 +313,7 @@ public void registerWorkflowImplementationTypes( "registerWorkflowImplementationTypes is not allowed after worker has started"); workflowWorker.registerWorkflowImplementationTypes(options, workflowImplementationClasses); + recordKotlinImplementations(workflowImplementationClasses); } /** @@ -487,6 +490,22 @@ public void registerActivitiesImplementations(Object... activityImplementations) activityWorker.registerActivityImplementations(activityImplementations); } workflowWorker.registerLocalActivityImplementations(activityImplementations); + for (Object implementation : activityImplementations) { + recordKotlinImplementations(implementation.getClass()); + } + } + + private void recordKotlinImplementations(Class... implementationClasses) { + if (clientOptions.isWorkerEnvironmentInfoDisabled() + || kotlinImplementationClass.get() != null) { + return; + } + for (Class implementationClass : implementationClasses) { + if (WorkerEnvironmentInfo.isKotlinImplementation(implementationClass)) { + kotlinImplementationClass.compareAndSet(null, implementationClass); + return; + } + } } /** @@ -609,7 +628,8 @@ void onHeartbeatAccepted() { Supplier buildHeartbeatCallback( String workerGroupingKey, @Nullable EnvironmentInfo environmentInfo) { - pendingEnvironmentInfo.set(environmentInfo); + pendingEnvironmentInfo.set( + WorkerEnvironmentInfo.withKotlinRuntime(environmentInfo, kotlinImplementationClass.get())); // The callback can be invoked concurrently from the heartbeat scheduler and the shutdown path final Object callbackLock = new Object(); final AtomicReference lastHeartbeatTime = new AtomicReference<>(null); diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkerEnvironmentInfoTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkerEnvironmentInfoTest.java index 1c84aa37b0..2e1f4812ea 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkerEnvironmentInfoTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkerEnvironmentInfoTest.java @@ -7,21 +7,33 @@ import io.temporal.api.worker.v1.EnvironmentInfo.HostingEnvironment; import io.temporal.api.worker.v1.EnvironmentInfo.HostingEnvironment.HostingEnvironmentType; import io.temporal.api.worker.v1.EnvironmentInfo.Runtime.RuntimeType; +import java.io.File; +import java.net.URI; +import java.net.URL; +import java.net.URLClassLoader; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.stream.Collectors; +import javax.tools.JavaCompiler; +import javax.tools.JavaFileObject; +import javax.tools.SimpleJavaFileObject; +import javax.tools.ToolProvider; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; public class WorkerEnvironmentInfoTest { + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + @Test public void detectReportsJvmRuntimeAndPlatform() { EnvironmentInfo info = WorkerEnvironmentInfo.detect(); - assertEquals(1, info.getRuntimesCount()); + assertTrue(info.getRuntimesCount() >= 1); assertEquals(RuntimeType.RUNTIME_TYPE_JVM, info.getRuntimes(0).getType()); assertEquals(System.getProperty("java.version"), info.getRuntimes(0).getVersion()); @@ -43,6 +55,192 @@ public void detectReportsJvmRuntimeAndPlatform() { } } + @Test + public void detectGraalNativeImageRuntime() { + Map properties = new HashMap<>(); + properties.put("java.version", "21.0.2"); + properties.put("org.graalvm.nativeimage.imagecode", "runtime"); + properties.put("org.graalvm.version", "24.0.1"); + EnvironmentInfo info = WorkerEnvironmentInfo.detect(properties::get); + assertEquals(2, info.getRuntimesCount()); + assertEquals(RuntimeType.RUNTIME_TYPE_JVM, info.getRuntimes(0).getType()); + assertEquals("21.0.2", info.getRuntimes(0).getVersion()); + assertEquals(RuntimeType.RUNTIME_TYPE_GRAAL_AOT, info.getRuntimes(1).getType()); + assertEquals("24.0.1", info.getRuntimes(1).getVersion()); + + properties.remove("org.graalvm.version"); + assertEquals("", WorkerEnvironmentInfo.detect(properties::get).getRuntimes(1).getVersion()); + properties.put("org.graalvm.nativeimage.imagecode", "buildtime"); + assertEquals(1, WorkerEnvironmentInfo.detect(properties::get).getRuntimesCount()); + properties.remove("org.graalvm.nativeimage.imagecode"); + assertEquals(1, WorkerEnvironmentInfo.detect(properties::get).getRuntimesCount()); + } + + @Test + public void runtimeProbeFailuresAreIsolated() { + for (boolean linkageError : new boolean[] {false, true}) { + EnvironmentInfo baseline = WorkerEnvironmentInfo.detect(); + EnvironmentInfo info = + WorkerEnvironmentInfo.detect( + name -> { + if ("org.graalvm.nativeimage.imagecode".equals(name)) { + throwProbeFailure(linkageError); + } + return System.getProperty(name); + }); + assertEquals(RuntimeType.RUNTIME_TYPE_JVM, info.getRuntimes(0).getType()); + assertEquals(System.getProperty("java.version"), info.getRuntimes(0).getVersion()); + assertEquals(baseline.getPlatform(), info.getPlatform()); + assertEquals(baseline.getHostingEnvironmentsList(), info.getHostingEnvironmentsList()); + + info = + WorkerEnvironmentInfo.detect( + name -> { + if ("org.graalvm.nativeimage.imagecode".equals(name)) { + return "runtime"; + } + throwProbeFailure(linkageError); + return null; + }); + assertEquals(2, info.getRuntimesCount()); + assertEquals("", info.getRuntimes(0).getVersion()); + assertEquals(RuntimeType.RUNTIME_TYPE_GRAAL_AOT, info.getRuntimes(1).getType()); + assertEquals("", info.getRuntimes(1).getVersion()); + assertEquals(baseline.getPlatform(), info.getPlatform()); + assertEquals(baseline.getHostingEnvironmentsList(), info.getHostingEnvironmentsList()); + } + } + + @Test + public void kotlinAugmentationPreservesOptOutAndJavaImplementations() { + EnvironmentInfo info = WorkerEnvironmentInfo.detect(); + assertFalse(WorkerEnvironmentInfo.isKotlinImplementation(null)); + assertFalse(WorkerEnvironmentInfo.isKotlinImplementation(getClass())); + assertNull(WorkerEnvironmentInfo.withKotlinRuntime(null, getClass())); + assertSame(info, WorkerEnvironmentInfo.withKotlinRuntime(info, null)); + assertSame(info, WorkerEnvironmentInfo.withKotlinRuntime(info, getClass())); + } + + @Test + public void kotlinMetadataDetectedAcrossClassLoaderBoundary() throws Exception { + File classes = compileKotlinFixtures(true); + try (URLClassLoader loader = fixtureLoader(classes, null, false)) { + Class implementation = loader.loadClass("fixtures.Implementation"); + assertTrue(WorkerEnvironmentInfo.isKotlinImplementation(implementation)); + EnvironmentInfo info = WorkerEnvironmentInfo.detect(); + EnvironmentInfo augmented = WorkerEnvironmentInfo.withKotlinRuntime(info, implementation); + assertEquals(info.getRuntimesCount() + 1, augmented.getRuntimesCount()); + assertEquals(info.getPlatform(), augmented.getPlatform()); + assertEquals(info.getHostingEnvironmentsList(), augmented.getHostingEnvironmentsList()); + assertEquals( + RuntimeType.RUNTIME_TYPE_KOTLIN, + augmented.getRuntimes(info.getRuntimesCount()).getType()); + assertEquals("2.1.0", augmented.getRuntimes(info.getRuntimesCount()).getVersion()); + assertSame(augmented, WorkerEnvironmentInfo.withKotlinRuntime(augmented, implementation)); + assertEquals( + info.getRuntimesList(), augmented.getRuntimesList().subList(0, info.getRuntimesCount())); + assertNull(WorkerEnvironmentInfo.withKotlinRuntime(null, implementation)); + } + } + + @Test + public void kotlinVersionIsOptional() throws Exception { + File classes = compileKotlinFixtures(false); + try (URLClassLoader loader = fixtureLoader(classes, null, false)) { + EnvironmentInfo info = + WorkerEnvironmentInfo.withKotlinRuntime( + EnvironmentInfo.getDefaultInstance(), loader.loadClass("fixtures.Implementation")); + assertEquals(RuntimeType.RUNTIME_TYPE_KOTLIN, info.getRuntimes(0).getType()); + assertEquals("", info.getRuntimes(0).getVersion()); + } + } + + @Test + public void kotlinProbeFailuresAreBestEffort() throws Exception { + File classes = compileKotlinFixtures(true); + for (boolean linkageError : new boolean[] {false, true}) { + try (URLClassLoader loader = fixtureLoader(classes, "kotlin.Metadata", linkageError)) { + Class implementation = loader.loadClass("fixtures.Implementation"); + EnvironmentInfo info = EnvironmentInfo.getDefaultInstance(); + assertFalse(WorkerEnvironmentInfo.isKotlinImplementation(implementation)); + assertSame(info, WorkerEnvironmentInfo.withKotlinRuntime(info, implementation)); + } + try (URLClassLoader loader = fixtureLoader(classes, "kotlin.KotlinVersion", linkageError)) { + EnvironmentInfo info = + WorkerEnvironmentInfo.withKotlinRuntime( + EnvironmentInfo.getDefaultInstance(), loader.loadClass("fixtures.Implementation")); + assertEquals(RuntimeType.RUNTIME_TYPE_KOTLIN, info.getRuntimes(0).getType()); + assertEquals("", info.getRuntimes(0).getVersion()); + } + } + } + + private static void throwProbeFailure(boolean linkageError) { + if (linkageError) { + throw new LinkageError("Unavailable telemetry probe."); + } + throw new SecurityException("Unavailable telemetry probe."); + } + + private URLClassLoader fixtureLoader(File classes, String failingClass, boolean linkageError) + throws Exception { + // No parent: the fixtures neither need nor accidentally use a real Kotlin dependency. + return new URLClassLoader(new URL[] {classes.toURI().toURL()}, null) { + @Override + protected Class loadClass(String name, boolean resolve) throws ClassNotFoundException { + if (name.equals(failingClass)) { + throwProbeFailure(linkageError); + } + return super.loadClass(name, resolve); + } + }; + } + + private File compileKotlinFixtures(boolean includeVersion) throws Exception { + File classes = temporaryFolder.newFolder(); + List sources = + new java.util.ArrayList<>( + Arrays.asList( + source( + "kotlin.Metadata", + "package kotlin; @java.lang.annotation.Retention(" + + "java.lang.annotation.RetentionPolicy.RUNTIME) public @interface Metadata {}"), + source( + "fixtures.Implementation", + "package fixtures; @kotlin.Metadata public class Implementation {}"))); + if (includeVersion) { + sources.add( + source( + "kotlin.KotlinVersion", + "package kotlin; public class KotlinVersion {" + + " public static final KotlinVersion CURRENT = new KotlinVersion();" + + " public String toString() { return \"2.1.0\"; } }")); + } + JavaCompiler compiler = ToolProvider.getSystemJavaCompiler(); + assertNotNull("The fixture compiler requires a JDK.", compiler); + assertTrue( + compiler + .getTask( + null, + null, + null, + Arrays.asList("-source", "8", "-target", "8", "-d", classes.getAbsolutePath()), + null, + sources) + .call()); + return classes; + } + + private static JavaFileObject source(String name, String contents) { + return new SimpleJavaFileObject( + URI.create("string:///" + name.replace('.', '/') + ".java"), JavaFileObject.Kind.SOURCE) { + @Override + public CharSequence getCharContent(boolean ignoreEncodingErrors) { + return contents; + } + }; + } + @Test public void detectHostingEnvironments() { Map env = new HashMap<>(); From 24cb6ab88273233da73c03821af4e3a646e3a0e1 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Mon, 19 Oct 2026 09:17:16 -0700 Subject: [PATCH 3/4] Report runtimes for factory-registered workflows Add factory registration overloads that accept the implementation class for worker environment detection without invoking the factory. Pass the known class from Spring's ordinary, WorkflowInit, and dynamic workflow registration paths. --- CHANGELOG.md | 2 + .../main/java/io/temporal/worker/Worker.java | 47 +++++++++++++++++++ .../template/WorkersTemplate.java | 3 ++ 3 files changed, 52 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f3e17b042..dbc9d1e623 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,8 @@ to docs, or any other relevant information. that the client uses. - Worker heartbeats now report detected Graal native-image and Kotlin runtimes alongside the JVM version. Detection is best effort and respects the worker environment information opt-out. + Factory registration overloads accept an implementation class for runtime detection, including + Spring Boot workflow registrations. - Added experimental `ChildWorkflowOptions.Builder.setVersioningOverride` and `VersioningOverride.OneTimeVersioningOverride` for explicit pinned, auto-upgrade, and one-time child workflow routing. Invalid child overrides are reported as `InvalidVersioningOverrideFailure` diff --git a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java index 6e29786265..9538546587 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -421,6 +421,53 @@ public void registerWorkflowImplementationFactory( workflowWorker.registerWorkflowImplementationFactory(options, workflowInterface, factory); } + /** + * Registers a workflow factory with its implementation class for worker environment reporting. + * + *

The implementation class is used only for metadata detection. The factory remains + * responsible for creating workflow instances and is not invoked during registration. + * + * @param workflowInterface workflow interface implemented by the factory's instances + * @param implementationClass implementation class created by the factory + * @param factory factory that creates a new workflow instance each time it is called + * @param options custom workflow implementation options for a worker + * @see #registerWorkflowImplementationFactory(Class, Func, WorkflowImplementationOptions) + */ + public void registerWorkflowImplementationFactory( + Class workflowInterface, + Class implementationClass, + Func factory, + WorkflowImplementationOptions options) { + Objects.requireNonNull(implementationClass, "implementationClass"); + registerWorkflowImplementationFactory(workflowInterface, factory, options); + recordKotlinImplementations(implementationClass); + } + + /** + * Registers a workflow factory accepting workflow arguments with its implementation class for + * worker environment reporting. + * + *

The implementation class is used only for metadata detection. The factory remains + * responsible for creating workflow instances and is not invoked during registration. + * + * @param workflowInterface workflow interface implemented by the factory's instances + * @param implementationClass implementation class created by the factory + * @param factory factory that creates a new workflow instance from workflow arguments each time + * it is called + * @param options custom workflow implementation options for a worker + * @see #registerWorkflowImplementationFactory(Class, Functions.Func1, + * WorkflowImplementationOptions) + */ + public void registerWorkflowImplementationFactory( + Class workflowInterface, + Class implementationClass, + Functions.Func1 factory, + WorkflowImplementationOptions options) { + Objects.requireNonNull(implementationClass, "implementationClass"); + registerWorkflowImplementationFactory(workflowInterface, factory, options); + recordKotlinImplementations(implementationClass); + } + /** * Configures a factory to use when an instance of a workflow implementation is created. * diff --git a/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/template/WorkersTemplate.java b/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/template/WorkersTemplate.java index 46b0dc6dd5..be4e73f763 100644 --- a/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/template/WorkersTemplate.java +++ b/temporal-spring-boot-autoconfigure/src/main/java/io/temporal/spring/boot/autoconfigure/template/WorkersTemplate.java @@ -612,6 +612,7 @@ private void configureWorkflowImplementation(Worker worker, Class clazz) .createWorkflowImplementationOptions(worker, clazz, null); worker.registerWorkflowImplementationFactory( DynamicWorkflow.class, + (Class) clazz, (encodedValues) -> { if (ctor.isPresent()) { try { @@ -679,6 +680,7 @@ private void configureWorkflowImplementation(Worker worker, Class clazz) .createWorkflowImplementationOptions(worker, clazz, workflowMethod); worker.registerWorkflowImplementationFactory( (Class) workflowMethod.getWorkflowInterface(), + (Class) clazz, (encodedValues) -> { try { Constructor ctor = workflowMetadata.getWorkflowInit(); @@ -714,6 +716,7 @@ private void configureWorkflowImplementation(Worker worker, Class clazz) .createWorkflowImplementationOptions(worker, clazz, workflowMethod); worker.registerWorkflowImplementationFactory( (Class) workflowMethod.getWorkflowInterface(), + (Class) clazz, () -> (T) beanFactory.createBean(clazz), workflowImplementationOptions); addRegisteredWorkflowImpl( From e06adee3a49de7eec1e3f80a00cf9f75f5dae37d Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Mon, 19 Oct 2026 11:49:23 -0700 Subject: [PATCH 4/4] Fix one-shot environment assertion in heartbeat integration test Capture runtime metadata only from successfully acknowledged heartbeat RPCs. Verify a later accepted heartbeat omits the environment before asserting the captured values, rather than requiring metadata on the latest stored heartbeat. Keep ListWorkers and DescribeWorker round-trip checks. --- .../KotlinWorkerHeartbeatIntegrationTest.kt | 87 +++++++++++++++++-- 1 file changed, 81 insertions(+), 6 deletions(-) diff --git a/temporal-kotlin/src/test/kotlin/io/temporal/worker/KotlinWorkerHeartbeatIntegrationTest.kt b/temporal-kotlin/src/test/kotlin/io/temporal/worker/KotlinWorkerHeartbeatIntegrationTest.kt index e00daf74da..da095e7eea 100644 --- a/temporal-kotlin/src/test/kotlin/io/temporal/worker/KotlinWorkerHeartbeatIntegrationTest.kt +++ b/temporal-kotlin/src/test/kotlin/io/temporal/worker/KotlinWorkerHeartbeatIntegrationTest.kt @@ -1,5 +1,13 @@ package io.temporal.worker +import io.grpc.CallOptions +import io.grpc.Channel +import io.grpc.ClientCall +import io.grpc.ClientInterceptor +import io.grpc.ForwardingClientCall +import io.grpc.ForwardingClientCallListener +import io.grpc.Metadata +import io.grpc.MethodDescriptor import io.grpc.Status import io.grpc.StatusRuntimeException import io.temporal.api.worker.v1.EnvironmentInfo.Runtime.RuntimeType @@ -7,22 +15,76 @@ import io.temporal.api.worker.v1.WorkerHeartbeat import io.temporal.api.workflowservice.v1.DescribeNamespaceRequest import io.temporal.api.workflowservice.v1.DescribeWorkerRequest import io.temporal.api.workflowservice.v1.ListWorkersRequest +import io.temporal.api.workflowservice.v1.RecordWorkerHeartbeatRequest +import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc import io.temporal.client.WorkflowClientOptions +import io.temporal.serviceclient.WorkflowServiceStubsOptions import io.temporal.testing.internal.SDKTestWorkflowRule import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse import org.junit.Assert.assertNotNull +import org.junit.Assert.assertTrue import org.junit.Assume.assumeTrue import org.junit.Before import org.junit.Rule import org.junit.Test import java.time.Duration +import java.util.concurrent.ConcurrentHashMap class KotlinWorkerHeartbeatIntegrationTest { + private val acceptedEnvironments = ConcurrentHashMap() + private val workersWithLaterHeartbeats = ConcurrentHashMap.newKeySet() + private val heartbeatInterceptor = object : ClientInterceptor { + override fun interceptCall( + method: MethodDescriptor, + callOptions: CallOptions, + next: Channel + ): ClientCall { + val call = next.newCall(method, callOptions) + if (method != WorkflowServiceGrpc.getRecordWorkerHeartbeatMethod()) { + return call + } + return object : ForwardingClientCall.SimpleForwardingClientCall(call) { + @Volatile + private var request: RecordWorkerHeartbeatRequest? = null + + override fun sendMessage(message: ReqT) { + request = message as RecordWorkerHeartbeatRequest + super.sendMessage(message) + } + + override fun start(responseListener: Listener, headers: Metadata) { + super.start( + object : ForwardingClientCallListener.SimpleForwardingClientCallListener(responseListener) { + override fun onClose(status: Status, trailers: Metadata) { + if (status.isOk) { + request?.workerHeartbeatList?.forEach { + if (it.hasEnvironment()) { + acceptedEnvironments.putIfAbsent(it.workerInstanceKey, it) + } else { + workersWithLaterHeartbeats.add(it.workerInstanceKey) + } + } + } + super.onClose(status, trailers) + } + }, + headers + ) + } + } + } + } + @Rule @JvmField val testWorkflowRule = SDKTestWorkflowRule.newBuilder() - .setTestTimeoutSeconds(20) + .setTestTimeoutSeconds(45) + .setWorkflowServiceStubsOptions( + WorkflowServiceStubsOptions.newBuilder() + .setGrpcClientInterceptors(listOf(heartbeatInterceptor)) + .build() + ) .setWorkflowClientOptions( WorkflowClientOptions.newBuilder() .setWorkerHeartbeatInterval(Duration.ofSeconds(1)) @@ -63,10 +125,16 @@ class KotlinWorkerHeartbeatIntegrationTest { javaWorker.registerActivitiesImplementations(JavaHeartbeatImplementations.JavaActivityImpl()) environment.start() - assertEventually(Duration.ofSeconds(10)) { - assertServerRuntimes(workflowQueue, kotlin = true) - assertServerRuntimes(activityQueue, kotlin = true) - assertServerRuntimes(javaQueue, kotlin = false) + val pendingQueues = linkedMapOf( + workflowQueue to true, + activityQueue to true, + javaQueue to false + ) + assertEventually(Duration.ofSeconds(25)) { + for ((taskQueue, kotlin) in pendingQueues.toMap()) { + assertServerRuntimes(taskQueue, kotlin) + pendingQueues.remove(taskQueue) + } } } @@ -79,7 +147,14 @@ class KotlinWorkerHeartbeatIntegrationTest { assertNotNull("DescribeWorker should return a stored heartbeat", described) assertEquals(taskQueue, described!!.taskQueue) assertEquals(listed.workerInstanceKey, described.workerInstanceKey) - assertRuntimes(described, kotlin) + assertTrue( + "Waiting for a later accepted heartbeat without environment for queue: $taskQueue", + workersWithLaterHeartbeats.contains(listed.workerInstanceKey) + ) + // Later heartbeats omit the one-shot environment, so inspect a successfully accepted RPC. + val accepted = acceptedEnvironments[listed.workerInstanceKey] + assertNotNull("No accepted heartbeat with environment for queue: $taskQueue", accepted) + assertRuntimes(accepted!!, kotlin) } private fun assertRuntimes(heartbeat: WorkerHeartbeat, kotlin: Boolean) {