From 0a69278ee5ef729c4a4cc945a4cb0d405855efeb Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Tue, 7 Jul 2026 16:32:25 -0700 Subject: [PATCH 1/4] Add preferred version provider to worker options --- .../replay/ReplayWorkflowContextImpl.java | 16 +- .../{sync => replay}/WorkflowInfoImpl.java | 7 +- .../statemachines/VersionStateMachine.java | 54 +++- .../statemachines/WorkflowStateMachines.java | 12 + .../internal/sync/SyncWorkflowContext.java | 35 +-- .../internal/sync/WorkflowInternal.java | 1 + .../internal/worker/SingleWorkerOptions.java | 20 +- .../worker/PreferredVersionProvider.java | 30 +++ .../worker/PreferredVersionProviderInput.java | 63 +++++ .../io/temporal/worker/VersionPreference.java | 62 +++++ .../main/java/io/temporal/worker/Worker.java | 3 +- .../io/temporal/worker/WorkerOptions.java | 39 ++- .../java/io/temporal/workflow/Workflow.java | 7 +- .../VersionStateMachineTest.java | 235 ++++++++++++++++++ .../io/temporal/worker/WorkerOptionsTest.java | 7 +- .../PreferredVersionProviderTest.java | 78 ++++++ 16 files changed, 630 insertions(+), 39 deletions(-) rename temporal-sdk/src/main/java/io/temporal/internal/{sync => replay}/WorkflowInfoImpl.java (96%) create mode 100644 temporal-sdk/src/main/java/io/temporal/worker/PreferredVersionProvider.java create mode 100644 temporal-sdk/src/main/java/io/temporal/worker/PreferredVersionProviderInput.java create mode 100644 temporal-sdk/src/main/java/io/temporal/worker/VersionPreference.java create mode 100644 temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderTest.java diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java index cf704544bd..17cea576d5 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java @@ -16,6 +16,8 @@ import io.temporal.internal.common.SdkFlag; import io.temporal.internal.statemachines.*; import io.temporal.internal.worker.SingleWorkerOptions; +import io.temporal.worker.PreferredVersionProvider; +import io.temporal.worker.PreferredVersionProviderInput; import io.temporal.workflow.Functions; import io.temporal.workflow.Functions.Func; import io.temporal.workflow.Functions.Func1; @@ -38,6 +40,7 @@ final class ReplayWorkflowContextImpl implements ReplayWorkflowContext { private final @Nullable String fullReplayDirectQueryName; private final Scope replayAwareWorkflowMetricsScope; private final SingleWorkerOptions workerOptions; + private final WorkflowInfoImpl workflowInfo; /** * @param fullReplayDirectQueryName query name if an execution is a full replay caused by a direct @@ -61,6 +64,7 @@ final class ReplayWorkflowContextImpl implements ReplayWorkflowContext { this.replayAwareWorkflowMetricsScope = new ReplayAwareScope(workflowMetricsScope, this, workflowStateMachines::currentTimeMillis); this.workerOptions = workerOptions; + this.workflowInfo = new WorkflowInfoImpl(this); } @Override @@ -337,7 +341,17 @@ public Integer getVersion( int minSupported, int maxSupported, Functions.Proc2 callback) { - return workflowStateMachines.getVersion(changeId, minSupported, maxSupported, callback); + PreferredVersionProvider preferredVersionProvider = workerOptions.getPreferredVersionProvider(); + return workflowStateMachines.getVersion( + changeId, + minSupported, + maxSupported, + preferredVersionProvider == null + ? null + : (min, max) -> + preferredVersionProvider.getPreferredVersion( + new PreferredVersionProviderInput(workflowInfo, changeId, min, max)), + callback); } @Override diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInfoImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/WorkflowInfoImpl.java similarity index 96% rename from temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInfoImpl.java rename to temporal-sdk/src/main/java/io/temporal/internal/replay/WorkflowInfoImpl.java index b30f13d458..d51df7bc25 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInfoImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/replay/WorkflowInfoImpl.java @@ -1,4 +1,4 @@ -package io.temporal.internal.sync; +package io.temporal.internal.replay; import io.temporal.api.common.v1.SearchAttributes; import io.temporal.api.common.v1.WorkflowExecution; @@ -6,7 +6,6 @@ import io.temporal.common.RetryOptions; import io.temporal.common.SuggestContinueAsNewReason; import io.temporal.internal.common.ProtoConverters; -import io.temporal.internal.replay.ReplayWorkflowContext; import io.temporal.workflow.WorkflowInfo; import java.time.Duration; import java.util.List; @@ -14,11 +13,11 @@ import javax.annotation.Nonnull; import javax.annotation.Nullable; -final class WorkflowInfoImpl implements WorkflowInfo { +public final class WorkflowInfoImpl implements WorkflowInfo { private final ReplayWorkflowContext context; - WorkflowInfoImpl(ReplayWorkflowContext context) { + public WorkflowInfoImpl(ReplayWorkflowContext context) { this.context = context; } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/VersionStateMachine.java b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/VersionStateMachine.java index c218a82dbe..73874df05c 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/VersionStateMachine.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/VersionStateMachine.java @@ -14,8 +14,11 @@ import io.temporal.api.history.v1.HistoryEvent; import io.temporal.internal.history.VersionMarkerUtils; import io.temporal.worker.NonDeterministicException; +import io.temporal.worker.VersionPreference; import io.temporal.workflow.Functions; import java.util.Objects; +import java.util.Optional; +import java.util.function.BiFunction; import javax.annotation.Nullable; final class VersionStateMachine { @@ -137,17 +140,24 @@ class InvocationStateMachine private final Functions.Func1 upsertSearchAttributeCallback; private final Functions.Proc2 resultCallback; + @Nullable + private final BiFunction> + preferredVersionProvider; + InvocationStateMachine( int minSupported, int maxSupported, boolean waitForMarkerRecordedReplaying, Functions.Func1 upsertSearchAttributeCallback, + @Nullable + BiFunction> preferredVersionProvider, Functions.Proc2 callback) { super(STATE_MACHINE_DEFINITION, VersionStateMachine.this.commandSink, stateMachineSink); this.minSupported = minSupported; this.maxSupported = maxSupported; this.waitForMarkerRecordedReplaying = waitForMarkerRecordedReplaying; this.upsertSearchAttributeCallback = upsertSearchAttributeCallback; + this.preferredVersionProvider = preferredVersionProvider; this.resultCallback = Objects.requireNonNull(callback); } @@ -210,13 +220,17 @@ private void validateVersionAndThrow(boolean preloaded) { throw new IllegalStateException((preloaded ? "preloaded " : "") + " version not set"); } if (versionToUse < minSupported || versionToUse > maxSupported) { - throw new UnsupportedVersion.UnsupportedVersionException( - String.format( - "Version %d of changeId %s is not supported. Supported v is between %d and %d.", - versionToUse, changeId, minSupported, maxSupported)); + throwUnsupportedVersion(versionToUse); } } + private void throwUnsupportedVersion(int unsupportedVersion) { + throw new UnsupportedVersion.UnsupportedVersionException( + String.format( + "Version %d of changeId %s is not supported. Supported v is between %d and %d.", + unsupportedVersion, changeId, minSupported, maxSupported)); + } + void notifyFromVersion(boolean preloaded) { Integer versionToUse = preloaded ? preloadedVersion : version; resultCallback.apply(versionToUse, null); @@ -227,8 +241,8 @@ void notifyFromException(RuntimeException ex) { } void notifyFromVersionExecuting() { - // the only case when we don't need to validate before notification because - // we just initialized the version with maxVersion + // No validation needed before notification here: resolveVersionExecuting() already guarantees + // the version it produces is within [minSupported, maxSupported]. notifyFromVersion(false); } @@ -238,7 +252,7 @@ State createMarkerExecuting() { addCommand(StateMachineCommandUtils.RECORD_MARKER_FAKE_COMMAND); return State.SKIPPED; } else { - version = maxSupported; + version = resolveVersionExecuting(); SearchAttributes sa = upsertSearchAttributeCallback.apply(version); writeVersionChangeSA = sa != null; RecordMarkerCommandAttributes markerAttributes = @@ -253,6 +267,30 @@ State createMarkerExecuting() { } } + private int resolveVersionExecuting() { + if (preferredVersionProvider == null) { + return maxSupported; + } + Optional preference = + Objects.requireNonNull( + preferredVersionProvider.apply(minSupported, maxSupported), + "PreferredVersionProvider returned null"); + if (!preference.isPresent()) { + return maxSupported; + } + + VersionPreference versionPreference = preference.get(); + int preferredVersion = versionPreference.getVersion(); + if (preferredVersion >= minSupported && preferredVersion <= maxSupported) { + return preferredVersion; + } + if (versionPreference.isClampToSupportedRange()) { + return Math.min(Math.max(preferredVersion, minSupported), maxSupported); + } + throwUnsupportedVersion(preferredVersion); + throw new IllegalStateException("unreachable"); + } + void notifySkippedExecuting() { cancelCommand(); try { @@ -411,6 +449,7 @@ public Integer getVersion( int maxSupported, boolean waitForMarkerRecordedReplaying, Functions.Func1 upsertSearchAttributeCallback, + @Nullable BiFunction> preferredVersionProvider, Functions.Proc2 callback) { InvocationStateMachine ism = new InvocationStateMachine( @@ -418,6 +457,7 @@ public Integer getVersion( maxSupported, waitForMarkerRecordedReplaying, upsertSearchAttributeCallback, + preferredVersionProvider, callback); ism.explicitEvent(ExplicitEvent.CHECK_EXECUTION_STATE); ism.explicitEvent(ExplicitEvent.SCHEDULE); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java index 884e6947d7..e5b0e7d701 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java @@ -30,12 +30,14 @@ import io.temporal.serviceclient.Version; import io.temporal.worker.MetricsType; import io.temporal.worker.NonDeterministicException; +import io.temporal.worker.VersionPreference; import io.temporal.worker.WorkflowImplementationOptions; import io.temporal.workflow.ChildWorkflowCancellationType; import io.temporal.workflow.Functions; import io.temporal.workflow.NexusOperationCancellationType; import java.nio.charset.StandardCharsets; import java.util.*; +import java.util.function.BiFunction; import javax.annotation.Nonnull; import javax.annotation.Nullable; import org.slf4j.Logger; @@ -1244,6 +1246,15 @@ public Integer getVersion( int minSupported, int maxSupported, Functions.Proc2 callback) { + return getVersion(changeId, minSupported, maxSupported, null, callback); + } + + public Integer getVersion( + String changeId, + int minSupported, + int maxSupported, + @Nullable BiFunction> preferredVersionProvider, + Functions.Proc2 callback) { VersionStateMachine stateMachine = versions.computeIfAbsent( changeId, @@ -1275,6 +1286,7 @@ public Integer getVersion( } return sa; }, + preferredVersionProvider, (v, e) -> { callback.apply(v, e); // without this getVersion call will trigger the end of WFT, diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java index 1517c9257e..a1d92fc971 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java @@ -1143,21 +1143,26 @@ private R mutableSideEffectImpl( @Override public int getVersion(String changeId, int minSupported, int maxSupported) { CompletablePromise result = Workflow.newPromise(); - Integer versionToUse = - replayContext.getVersion( - changeId, - minSupported, - maxSupported, - (v, e) -> - runner.executeInWorkflowThread( - "version-callback", - () -> { - if (v != null) { - result.complete(v); - } else { - result.completeExceptionally(e); - } - })); + Integer versionToUse; + try { + versionToUse = + replayContext.getVersion( + changeId, + minSupported, + maxSupported, + (v, e) -> + runner.executeInWorkflowThread( + "version-callback", + () -> { + if (v != null) { + result.complete(v); + } else { + result.completeExceptionally(e); + } + })); + } catch (UnsupportedVersion.UnsupportedVersionException ex) { + throw new UnsupportedVersion(ex); + } /* * If we are replaying a workflow and encounter a getVersion call it is possible that this call did not exist * on the original execution. If the call did not exist on the original execution then we cannot block on results diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java index 79495694b4..3656c29d80 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java @@ -27,6 +27,7 @@ import io.temporal.internal.common.SdkFlag; import io.temporal.internal.common.SearchAttributesUtil; import io.temporal.internal.logging.ReplayAwareLogger; +import io.temporal.internal.replay.WorkflowInfoImpl; import io.temporal.internal.statemachines.UnsupportedContinueAsNewRequest; import io.temporal.serviceclient.CheckedExceptionWrapper; import io.temporal.workflow.*; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java index 3e84dc750f..8e0288566e 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java @@ -7,6 +7,7 @@ import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.GlobalDataConverter; import io.temporal.common.interceptors.WorkerInterceptor; +import io.temporal.worker.PreferredVersionProvider; import io.temporal.worker.WorkerDeploymentOptions; import java.time.Duration; import java.util.List; @@ -43,6 +44,7 @@ public static final class Builder { private String workerInstanceKey; private boolean allowActivityHeartbeatDuringShutdown; private String workerControlTaskQueue; + private PreferredVersionProvider preferredVersionProvider; private Builder() {} @@ -70,6 +72,7 @@ private Builder(SingleWorkerOptions options) { this.workerInstanceKey = options.getWorkerInstanceKey(); this.allowActivityHeartbeatDuringShutdown = options.getAllowActivityHeartbeatDuringShutdown(); this.workerControlTaskQueue = options.getWorkerControlTaskQueue(); + this.preferredVersionProvider = options.getPreferredVersionProvider(); } public Builder setIdentity(String identity) { @@ -177,6 +180,11 @@ public Builder setWorkerControlTaskQueue(String workerControlTaskQueue) { return this; } + public Builder setPreferredVersionProvider(PreferredVersionProvider preferredVersionProvider) { + this.preferredVersionProvider = preferredVersionProvider; + return this; + } + public SingleWorkerOptions build() { PollerOptions pollerOptions = this.pollerOptions; if (pollerOptions == null) { @@ -218,7 +226,8 @@ public SingleWorkerOptions build() { this.deploymentOptions, this.workerInstanceKey, this.allowActivityHeartbeatDuringShutdown, - this.workerControlTaskQueue); + this.workerControlTaskQueue, + this.preferredVersionProvider); } } @@ -242,6 +251,7 @@ public SingleWorkerOptions build() { private final String workerInstanceKey; private final boolean allowActivityHeartbeatDuringShutdown; private final String workerControlTaskQueue; + private final PreferredVersionProvider preferredVersionProvider; private SingleWorkerOptions( String identity, @@ -263,7 +273,8 @@ private SingleWorkerOptions( WorkerDeploymentOptions deploymentOptions, String workerInstanceKey, boolean allowActivityHeartbeatDuringShutdown, - String workerControlTaskQueue) { + String workerControlTaskQueue, + PreferredVersionProvider preferredVersionProvider) { this.identity = identity; this.binaryChecksum = binaryChecksum; this.buildId = buildId; @@ -284,6 +295,7 @@ private SingleWorkerOptions( this.workerInstanceKey = workerInstanceKey; this.allowActivityHeartbeatDuringShutdown = allowActivityHeartbeatDuringShutdown; this.workerControlTaskQueue = workerControlTaskQueue; + this.preferredVersionProvider = preferredVersionProvider; } public String getIdentity() { @@ -377,6 +389,10 @@ public String getWorkerControlTaskQueue() { return workerControlTaskQueue; } + public PreferredVersionProvider getPreferredVersionProvider() { + return preferredVersionProvider; + } + public WorkerVersioningOptions getWorkerVersioningOptions() { return new WorkerVersioningOptions( this.getBuildId(), this.isUsingBuildIdForVersioning(), this.getDeploymentOptions()); diff --git a/temporal-sdk/src/main/java/io/temporal/worker/PreferredVersionProvider.java b/temporal-sdk/src/main/java/io/temporal/worker/PreferredVersionProvider.java new file mode 100644 index 0000000000..1412d0d22e --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/worker/PreferredVersionProvider.java @@ -0,0 +1,30 @@ +package io.temporal.worker; + +import io.temporal.common.Experimental; +import java.util.Optional; +import javax.annotation.Nonnull; + +/** + * Provides the version to record for a {@link io.temporal.workflow.Workflow#getVersion(String, int, + * int)} call when the version marker is created for the first time. + * + *

This provider is called only during non-replay workflow task execution, before the SDK records + * a version marker for the change ID. It is not called during replay or after a version was already + * memoized for the same change ID. + */ +@Experimental +@FunctionalInterface +public interface PreferredVersionProvider { + + /** + * Returns the preferred version for the supplied {@code getVersion} call, or {@link + * Optional#empty()} to use the SDK default of {@code maxSupported}. + * + *

This method is invoked on the workflow thread. Any exception it throws propagates out of the + * {@code getVersion} call and fails the current workflow task, which will be retried; a provider + * that always throws will therefore block the workflow. Implementations should be deterministic + * and side-effect free. + */ + @Nonnull + Optional getPreferredVersion(@Nonnull PreferredVersionProviderInput input); +} diff --git a/temporal-sdk/src/main/java/io/temporal/worker/PreferredVersionProviderInput.java b/temporal-sdk/src/main/java/io/temporal/worker/PreferredVersionProviderInput.java new file mode 100644 index 0000000000..f29173aa58 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/worker/PreferredVersionProviderInput.java @@ -0,0 +1,63 @@ +package io.temporal.worker; + +import io.temporal.common.Experimental; +import io.temporal.workflow.WorkflowInfo; +import java.util.Objects; +import javax.annotation.Nonnull; + +/** Input passed to {@link PreferredVersionProvider}. */ +@Experimental +public final class PreferredVersionProviderInput { + private final WorkflowInfo workflowInfo; + private final String changeId; + private final int minSupported; + private final int maxSupported; + + public PreferredVersionProviderInput( + @Nonnull WorkflowInfo workflowInfo, + @Nonnull String changeId, + int minSupported, + int maxSupported) { + this.workflowInfo = Objects.requireNonNull(workflowInfo); + this.changeId = Objects.requireNonNull(changeId); + this.minSupported = minSupported; + this.maxSupported = maxSupported; + } + + /** Returns information about the workflow execution handling the {@code getVersion} call. */ + @Nonnull + public WorkflowInfo getWorkflowInfo() { + return workflowInfo; + } + + /** Returns the change ID passed to {@code Workflow.getVersion}. */ + @Nonnull + public String getChangeId() { + return changeId; + } + + /** Returns the minimum supported version passed to {@code Workflow.getVersion}. */ + public int getMinSupported() { + return minSupported; + } + + /** Returns the maximum supported version passed to {@code Workflow.getVersion}. */ + public int getMaxSupported() { + return maxSupported; + } + + @Override + public String toString() { + return "PreferredVersionProviderInput{" + + "workflowInfo=" + + workflowInfo + + ", changeId='" + + changeId + + '\'' + + ", minSupported=" + + minSupported + + ", maxSupported=" + + maxSupported + + '}'; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/worker/VersionPreference.java b/temporal-sdk/src/main/java/io/temporal/worker/VersionPreference.java new file mode 100644 index 0000000000..eb612c8dd0 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/worker/VersionPreference.java @@ -0,0 +1,62 @@ +package io.temporal.worker; + +import io.temporal.common.Experimental; +import java.util.Objects; + +/** Version preference returned by {@link PreferredVersionProvider}. */ +@Experimental +public final class VersionPreference { + private final int version; + private final boolean clampToSupportedRange; + + private VersionPreference(int version, boolean clampToSupportedRange) { + this.version = version; + this.clampToSupportedRange = clampToSupportedRange; + } + + /** + * Creates a preference for {@code version}. If the version is outside the supported range for the + * call, the SDK fails the workflow task unless {@link #clampToSupportedRange()} is used. + */ + public static VersionPreference of(int version) { + return new VersionPreference(version, false); + } + + /** + * Returns a preference that clamps the version into the call's supported range before recording. + */ + public VersionPreference clampToSupportedRange() { + return new VersionPreference(version, true); + } + + public int getVersion() { + return version; + } + + public boolean isClampToSupportedRange() { + return clampToSupportedRange; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + VersionPreference that = (VersionPreference) o; + return version == that.version && clampToSupportedRange == that.clampToSupportedRange; + } + + @Override + public int hashCode() { + return Objects.hash(version, clampToSupportedRange); + } + + @Override + public String toString() { + return "VersionPreference{" + + "version=" + + version + + ", clampToSupportedRange=" + + clampToSupportedRange + + '}'; + } +} 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 5c77e93a6c..5ba7613865 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -1072,7 +1072,8 @@ private static SingleWorkerOptions.Builder toSingleWorkerOptions( .setDefaultHeartbeatThrottleInterval(options.getDefaultHeartbeatThrottleInterval()) .setDeploymentOptions(options.getDeploymentOptions()) .setWorkerInstanceKey(workerInstanceKey) - .setWorkerControlTaskQueue(workerControlTaskQueue); + .setWorkerControlTaskQueue(workerControlTaskQueue) + .setPreferredVersionProvider(options.getPreferredVersionProvider()); } /** diff --git a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java index 84db0b8d62..2f69517b6c 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java @@ -78,6 +78,7 @@ public static final class Builder { private PollerBehavior activityTaskPollersBehavior; private PollerBehavior nexusTaskPollersBehavior; private boolean allowActivityHeartbeatDuringShutdown; + private PreferredVersionProvider preferredVersionProvider; private Builder() {} @@ -114,6 +115,7 @@ private Builder(WorkerOptions o) { this.activityTaskPollersBehavior = o.activityTaskPollersBehavior; this.nexusTaskPollersBehavior = o.nexusTaskPollersBehavior; this.allowActivityHeartbeatDuringShutdown = o.allowActivityHeartbeatDuringShutdown; + this.preferredVersionProvider = o.preferredVersionProvider; } /** @@ -548,6 +550,19 @@ public Builder setAllowActivityHeartbeatDuringShutdown( return this; } + /** + * Sets a provider that can choose the version recorded by the first non-replay {@link + * io.temporal.workflow.Workflow#getVersion(String, int, int)} call for a change ID. + * + *

If unset, or if the provider returns {@link java.util.Optional#empty()}, the SDK keeps the + * existing behavior of recording {@code maxSupported}. + */ + @Experimental + public Builder setPreferredVersionProvider(PreferredVersionProvider preferredVersionProvider) { + this.preferredVersionProvider = preferredVersionProvider; + return this; + } + public WorkerOptions build() { return new WorkerOptions( maxWorkerActivitiesPerSecond, @@ -578,7 +593,8 @@ public WorkerOptions build() { workflowTaskPollersBehavior, activityTaskPollersBehavior, nexusTaskPollersBehavior, - allowActivityHeartbeatDuringShutdown); + allowActivityHeartbeatDuringShutdown, + preferredVersionProvider); } public WorkerOptions validateAndBuildWithDefaults() { @@ -711,7 +727,8 @@ public WorkerOptions validateAndBuildWithDefaults() { workflowTaskPollersBehavior, activityTaskPollersBehavior, nexusTaskPollersBehavior, - allowActivityHeartbeatDuringShutdown); + allowActivityHeartbeatDuringShutdown, + preferredVersionProvider); } } @@ -744,6 +761,7 @@ public WorkerOptions validateAndBuildWithDefaults() { private final PollerBehavior activityTaskPollersBehavior; private final PollerBehavior nexusTaskPollersBehavior; private final boolean allowActivityHeartbeatDuringShutdown; + private final PreferredVersionProvider preferredVersionProvider; private WorkerOptions( double maxWorkerActivitiesPerSecond, @@ -774,7 +792,8 @@ private WorkerOptions( PollerBehavior workflowTaskPollersBehavior, PollerBehavior activityTaskPollersBehavior, PollerBehavior nexusTaskPollersBehavior, - boolean allowActivityHeartbeatDuringShutdown) { + boolean allowActivityHeartbeatDuringShutdown, + PreferredVersionProvider preferredVersionProvider) { this.maxWorkerActivitiesPerSecond = maxWorkerActivitiesPerSecond; this.maxConcurrentActivityExecutionSize = maxConcurrentActivityExecutionSize; this.maxConcurrentWorkflowTaskExecutionSize = maxConcurrentWorkflowTaskExecutionSize; @@ -804,6 +823,7 @@ private WorkerOptions( this.activityTaskPollersBehavior = activityTaskPollersBehavior; this.nexusTaskPollersBehavior = nexusTaskPollersBehavior; this.allowActivityHeartbeatDuringShutdown = allowActivityHeartbeatDuringShutdown; + this.preferredVersionProvider = preferredVersionProvider; } public double getMaxWorkerActivitiesPerSecond() { @@ -946,6 +966,11 @@ public boolean getAllowActivityHeartbeatDuringShutdown() { return allowActivityHeartbeatDuringShutdown; } + @Experimental + public PreferredVersionProvider getPreferredVersionProvider() { + return preferredVersionProvider; + } + @Override public boolean equals(Object o) { if (this == o) return true; @@ -979,7 +1004,8 @@ && compare(maxTaskQueueActivitiesPerSecond, that.maxTaskQueueActivitiesPerSecond && Objects.equals(workflowTaskPollersBehavior, that.workflowTaskPollersBehavior) && Objects.equals(activityTaskPollersBehavior, that.activityTaskPollersBehavior) && Objects.equals(nexusTaskPollersBehavior, that.nexusTaskPollersBehavior) - && allowActivityHeartbeatDuringShutdown == that.allowActivityHeartbeatDuringShutdown; + && allowActivityHeartbeatDuringShutdown == that.allowActivityHeartbeatDuringShutdown + && Objects.equals(preferredVersionProvider, that.preferredVersionProvider); } @Override @@ -1013,7 +1039,8 @@ public int hashCode() { workflowTaskPollersBehavior, activityTaskPollersBehavior, nexusTaskPollersBehavior, - allowActivityHeartbeatDuringShutdown); + allowActivityHeartbeatDuringShutdown, + preferredVersionProvider); } @Override @@ -1078,6 +1105,8 @@ public String toString() { + nexusTaskPollersBehavior + ", allowActivityHeartbeatDuringShutdown=" + allowActivityHeartbeatDuringShutdown + + ", preferredVersionProvider=" + + preferredVersionProvider + '}'; } } diff --git a/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java b/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java index 0378b8b6c2..cc2c3e476f 100644 --- a/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java +++ b/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java @@ -1087,10 +1087,11 @@ public static R mutableSideEffect( * is going to break determinism. The solution is to have both old code that is used to replay * existing workflows as well as the new one that is used when it is executed for the first time.\ * - *

{@code getVersion} returns maxSupported version when is executed for the first time. This + *

{@code getVersion} returns maxSupported version when it is executed for the first time, + * unless the worker has a {@link io.temporal.worker.PreferredVersionProvider} configured. This * version is recorded into the workflow history as a marker event. Even if maxSupported version - * is changed the version that was recorded is returned on replay. DefaultVersion constant - * contains version of code that wasn't versioned before. + * or the worker preference is changed, the version that was recorded is returned on replay. + * DefaultVersion constant contains version of code that wasn't versioned before. * *

For example initially workflow has the following code: * diff --git a/temporal-sdk/src/test/java/io/temporal/internal/statemachines/VersionStateMachineTest.java b/temporal-sdk/src/test/java/io/temporal/internal/statemachines/VersionStateMachineTest.java index 62fcf501c4..d7c91892f6 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/statemachines/VersionStateMachineTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/statemachines/VersionStateMachineTest.java @@ -17,10 +17,12 @@ import io.temporal.common.converter.DefaultDataConverter; import io.temporal.internal.history.VersionMarkerUtils; import io.temporal.worker.NonDeterministicException; +import io.temporal.worker.VersionPreference; import io.temporal.workflow.Functions; import java.util.ArrayList; import java.util.List; import java.util.Optional; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.junit.AfterClass; import org.junit.Test; @@ -41,6 +43,239 @@ private WorkflowStateMachines newStateMachines(TestEntityManagerListenerBase lis return new WorkflowStateMachines(listener, stateMachineList::add); } + @Test + public void testPreferredVersionIsRecorded() { + final int preferredVersion = 7; + class TestListener extends TestEntityManagerListenerBase { + @Override + public void buildWorkflow(AsyncWorkflowBuilder builder) { + builder + .add2( + (v, c) -> + stateMachines.getVersion( + "id1", + DEFAULT_VERSION, + 12, + (min, max) -> Optional.of(VersionPreference.of(preferredVersion)), + c)) + .add((v) -> stateMachines.completeWorkflow(converter.toPayloads(v.getT1()))); + } + } + + TestHistoryBuilder h = + new TestHistoryBuilder() + .add(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED) + .addWorkflowTask(); + + stateMachines = newStateMachines(new TestListener()); + List commands = h.handleWorkflowTaskTakeCommands(stateMachines, 1); + + assertEquals(preferredVersion, getVersionFromMarker(commands.get(0))); + assertEquals( + preferredVersion, + (int) + converter.fromPayloads( + 0, + Optional.of( + commands.get(1).getCompleteWorkflowExecutionCommandAttributes().getResult()), + Integer.class, + Integer.class)); + } + + @Test + public void testPreferredVersionEmptyFallsBackToMaxSupported() { + final int maxSupported = 12; + class TestListener extends TestEntityManagerListenerBase { + @Override + public void buildWorkflow(AsyncWorkflowBuilder builder) { + builder + .add2( + (v, c) -> + stateMachines.getVersion( + "id1", DEFAULT_VERSION, maxSupported, (min, max) -> Optional.empty(), c)) + .add((v) -> stateMachines.completeWorkflow(converter.toPayloads(v.getT1()))); + } + } + + TestHistoryBuilder h = + new TestHistoryBuilder() + .add(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED) + .addWorkflowTask(); + + stateMachines = newStateMachines(new TestListener()); + List commands = h.handleWorkflowTaskTakeCommands(stateMachines, 1); + + assertEquals(maxSupported, getVersionFromMarker(commands.get(0))); + } + + @Test + public void testPreferredVersionCalledOnceForChangeId() { + AtomicInteger providerCalls = new AtomicInteger(); + class TestListener extends TestEntityManagerListenerBase { + @Override + public void buildWorkflow(AsyncWorkflowBuilder builder) { + builder + .add2( + (v, c) -> + stateMachines.getVersion( + "id1", + DEFAULT_VERSION, + 12, + (min, max) -> { + providerCalls.incrementAndGet(); + return Optional.of(VersionPreference.of(3)); + }, + c)) + .add2( + (v, c) -> + stateMachines.getVersion( + "id1", + DEFAULT_VERSION, + 12, + (min, max) -> { + providerCalls.incrementAndGet(); + return Optional.of(VersionPreference.of(4)); + }, + c)) + .add((v) -> stateMachines.completeWorkflow(converter.toPayloads(v.getT1()))); + } + } + + TestHistoryBuilder h = + new TestHistoryBuilder() + .add(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED) + .addWorkflowTask(); + + stateMachines = newStateMachines(new TestListener()); + List commands = h.handleWorkflowTaskTakeCommands(stateMachines, 1); + + assertEquals(1, providerCalls.get()); + assertEquals(3, getVersionFromMarker(commands.get(0))); + } + + @Test + public void testPreferredVersionOutOfRangeThrows() { + class TestListener extends TestEntityManagerListenerBase { + @Override + public void buildWorkflow(AsyncWorkflowBuilder builder) { + builder.add2( + (v, c) -> + stateMachines.getVersion( + "id1", + DEFAULT_VERSION, + 1, + (min, max) -> Optional.of(VersionPreference.of(2)), + c)); + } + } + + TestHistoryBuilder h = + new TestHistoryBuilder() + .add(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED) + .addWorkflowTask(); + + stateMachines = newStateMachines(new TestListener()); + RuntimeException exception = + assertThrows( + RuntimeException.class, () -> h.handleWorkflowTaskTakeCommands(stateMachines, 1)); + + assertEquals( + "Version 2 of changeId id1 is not supported. Supported v is between -1 and 1.", + rootCause(exception).getMessage()); + } + + private Throwable rootCause(Throwable throwable) { + Throwable result = throwable; + while (result.getCause() != null) { + result = result.getCause(); + } + return result; + } + + @Test + public void testPreferredVersionCanClampToSupportedRange() { + class TestListener extends TestEntityManagerListenerBase { + @Override + public void buildWorkflow(AsyncWorkflowBuilder builder) { + builder + .add2( + (v, c) -> + stateMachines.getVersion( + "id1", + DEFAULT_VERSION, + 1, + (min, max) -> Optional.of(VersionPreference.of(2).clampToSupportedRange()), + c)) + .add((v) -> stateMachines.completeWorkflow(converter.toPayloads(v.getT1()))); + } + } + + TestHistoryBuilder h = + new TestHistoryBuilder() + .add(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED) + .addWorkflowTask(); + + stateMachines = newStateMachines(new TestListener()); + List commands = h.handleWorkflowTaskTakeCommands(stateMachines, 1); + + assertEquals(1, getVersionFromMarker(commands.get(0))); + } + + @Test + public void testPreferredVersionProviderNotCalledDuringReplay() { + AtomicInteger providerCalls = new AtomicInteger(); + class TestListener extends TestEntityManagerListenerBase { + @Override + public void buildWorkflow(AsyncWorkflowBuilder builder) { + builder + .add2( + (v, c) -> + stateMachines.getVersion( + "id1", + DEFAULT_VERSION, + 12, + (min, max) -> { + providerCalls.incrementAndGet(); + return Optional.of(VersionPreference.of(4)); + }, + c)) + .add((v) -> stateMachines.completeWorkflow(converter.toPayloads(v.getT1()))); + } + } + + TestHistoryBuilder h = + historyWithVersionMarker("id1", 3).add(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED); + + stateMachines = newStateMachines(new TestListener()); + h.handleWorkflowTaskTakeCommands(stateMachines); + + assertEquals(0, providerCalls.get()); + } + + private int getVersionFromMarker(Command command) { + return converter.fromPayloads( + 0, + Optional.ofNullable( + command + .getRecordMarkerCommandAttributes() + .getDetailsOrThrow(VersionMarkerUtils.MARKER_VERSION_KEY)), + Integer.class, + Integer.class); + } + + private TestHistoryBuilder historyWithVersionMarker(String changeId, int version) { + MarkerRecordedEventAttributes.Builder markerBuilder = + MarkerRecordedEventAttributes.newBuilder() + .setMarkerName(VersionMarkerUtils.MARKER_NAME) + .putDetails( + VersionMarkerUtils.MARKER_CHANGE_ID_KEY, converter.toPayloads(changeId).get()) + .putDetails(VersionMarkerUtils.MARKER_VERSION_KEY, converter.toPayloads(version).get()); + return new TestHistoryBuilder() + .add(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED) + .addWorkflowTask() + .add(EventType.EVENT_TYPE_MARKER_RECORDED, markerBuilder.build()); + } + @AfterClass public static void generateCoverage() { List>> diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java index 9bde963162..9787c45ace 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java @@ -13,7 +13,8 @@ public void build() { WorkerOptions.Builder builder = WorkerOptions.newBuilder() .setMaxConcurrentActivityExecutionSize(10) - .setMaxConcurrentLocalActivityExecutionSize(11); + .setMaxConcurrentLocalActivityExecutionSize(11) + .setPreferredVersionProvider((input) -> java.util.Optional.empty()); verifyBuild(builder.build()); verifyBuild(builder.validateAndBuildWithDefaults()); @@ -22,6 +23,7 @@ public void build() { private void verifyBuild(WorkerOptions options) { assertEquals(10, options.getMaxConcurrentActivityExecutionSize()); assertEquals(11, options.getMaxConcurrentLocalActivityExecutionSize()); + assertNotNull(options.getPreferredVersionProvider()); } @Test @@ -33,6 +35,7 @@ public void verifyWorkerOptionsEquality() { @Test public void verifyNewBuilderFromExistingWorkerOptions() { + PreferredVersionProvider preferredVersionProvider = mock(PreferredVersionProvider.class); @SuppressWarnings("deprecation") WorkerOptions w1 = WorkerOptions.newBuilder() @@ -57,6 +60,7 @@ public void verifyNewBuilderFromExistingWorkerOptions() { .setStickyTaskQueueDrainTimeout(Duration.ofSeconds(15)) .setIdentity("worker-identity") .setAllowActivityHeartbeatDuringShutdown(true) + .setPreferredVersionProvider(preferredVersionProvider) .build(); WorkerOptions w2 = WorkerOptions.newBuilder(w1).build(); @@ -92,6 +96,7 @@ public void verifyNewBuilderFromExistingWorkerOptions() { assertEquals(w1.getIdentity(), w2.getIdentity()); assertEquals( w1.getAllowActivityHeartbeatDuringShutdown(), w2.getAllowActivityHeartbeatDuringShutdown()); + assertSame(w1.getPreferredVersionProvider(), w2.getPreferredVersionProvider()); } @Test diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderTest.java new file mode 100644 index 0000000000..ea85afab7c --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderTest.java @@ -0,0 +1,78 @@ +package io.temporal.workflow.versionTests; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import io.temporal.client.WorkflowStub; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.worker.PreferredVersionProviderInput; +import io.temporal.worker.VersionPreference; +import io.temporal.worker.WorkerOptions; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.shared.TestWorkflows.TestWorkflowReturnString; +import java.time.Duration; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; + +public class PreferredVersionProviderTest { + private static final String CHANGE_ID = "preferred-change"; + private static final AtomicInteger providerCalls = new AtomicInteger(); + private static final AtomicReference providerInput = + new AtomicReference<>(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(TestPreferredVersionWorkflow.class) + .setWorkerOptions( + WorkerOptions.newBuilder() + .setStickyQueueScheduleToStartTimeout(Duration.ZERO) + .setPreferredVersionProvider( + (input) -> { + providerCalls.incrementAndGet(); + providerInput.set(input); + return Optional.of(VersionPreference.of(Workflow.DEFAULT_VERSION)); + }) + .build()) + .build(); + + @Before + public void setUp() { + providerCalls.set(0); + providerInput.set(null); + } + + @Test + public void providerReceivesGetVersionInputAndIsNotCalledOnReplay() { + TestWorkflowReturnString workflowStub = + testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflowReturnString.class); + + String result = workflowStub.execute(); + + assertEquals("old", result); + assertEquals(1, providerCalls.get()); + + PreferredVersionProviderInput input = providerInput.get(); + assertNotNull(input); + assertEquals(CHANGE_ID, input.getChangeId()); + assertEquals(Workflow.DEFAULT_VERSION, input.getMinSupported()); + assertEquals(1, input.getMaxSupported()); + assertEquals( + WorkflowStub.fromTyped(workflowStub).getExecution().getWorkflowId(), + input.getWorkflowInfo().getWorkflowId()); + } + + public static class TestPreferredVersionWorkflow implements TestWorkflowReturnString { + @Override + public String execute() { + int version = Workflow.getVersion(CHANGE_ID, Workflow.DEFAULT_VERSION, 1); + Workflow.sleep(Duration.ofMillis(1)); + assertEquals(version, Workflow.getVersion(CHANGE_ID, Workflow.DEFAULT_VERSION, 1)); + return version == Workflow.DEFAULT_VERSION ? "old" : "new"; + } + } +} From cd39796d2d81c507a846eb524ea4f9fca9099805 Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Tue, 7 Jul 2026 17:27:22 -0700 Subject: [PATCH 2/4] Add more complete integration scenario --- .../PreferredVersionProviderRolloutTest.java | 240 ++++++++++++++++++ 1 file changed, 240 insertions(+) create mode 100644 temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderRolloutTest.java diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderRolloutTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderRolloutTest.java new file mode 100644 index 0000000000..d54c7f803a --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderRolloutTest.java @@ -0,0 +1,240 @@ +package io.temporal.workflow.versionTests; + +import static io.temporal.testUtils.Eventually.assertEventually; +import static org.junit.Assert.assertEquals; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.testing.TestWorkflowEnvironment; +import io.temporal.worker.PreferredVersionProvider; +import io.temporal.worker.VersionPreference; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; +import io.temporal.worker.WorkerOptions; +import io.temporal.workflow.QueryMethod; +import io.temporal.workflow.SignalMethod; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.time.Duration; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Before; +import org.junit.Test; + +public class PreferredVersionProviderRolloutTest { + private static final String CHANGE_ID = "preferred-change"; + private static final AtomicInteger unactivatedProviderCalls = new AtomicInteger(); + private static final AtomicInteger activatedProviderCalls = new AtomicInteger(); + private static final AtomicInteger oldWorkerSignals = new AtomicInteger(); + private static final AtomicInteger unactivatedNewWorkerSignals = new AtomicInteger(); + + @Before + public void setUp() { + unactivatedProviderCalls.set(0); + activatedProviderCalls.set(0); + oldWorkerSignals.set(0); + unactivatedNewWorkerSignals.set(0); + } + + @Test + public void unactivatedNewGetVersionCallReplaysOnOldWorkerWithoutTheCall() throws Exception { + String taskQueue = "preferred-version-provider-rollout-" + UUID.randomUUID(); + try (TestWorkflowEnvironment testEnvironment = TestWorkflowEnvironment.newInstance()) { + WorkerFactory newWorkerFactory = + WorkerFactory.newInstance(newClient(testEnvironment, "new-rollout-worker")); + Worker newWorker = + newWorkerFactory.newWorker( + taskQueue, + workerOptions( + (input) -> { + unactivatedProviderCalls.incrementAndGet(); + return Optional.of(VersionPreference.of(Workflow.DEFAULT_VERSION)); + })); + newWorker.registerWorkflowImplementationTypes(NewRolloutWorkflowImpl.class); + + testEnvironment.start(); + newWorkerFactory.start(); + + RolloutWorkflow workflow = newWorkflowStub(testEnvironment, taskQueue); + WorkflowClient.start(workflow::run); + + assertEventually( + Duration.ofSeconds(5), () -> assertEquals(1, unactivatedProviderCalls.get())); + assertEquals("new-0", workflow.state()); + + shutdown(newWorkerFactory); + + WorkerFactory oldWorkerFactory = + WorkerFactory.newInstance(newClient(testEnvironment, "old-rollout-worker")); + Worker oldWorker = oldWorkerFactory.newWorker(taskQueue, workerOptions(null)); + oldWorker.registerWorkflowImplementationTypes(OldRolloutWorkflowImpl.class); + oldWorkerFactory.start(); + + workflow.release(); + testEnvironment.sleep(Duration.ofSeconds(1)); + + assertEventually(Duration.ofSeconds(5), () -> assertEquals(1, oldWorkerSignals.get())); + + workflow.release(); + assertEquals("old", WorkflowStub.fromTyped(workflow).getResult(String.class)); + shutdown(oldWorkerFactory); + } + } + + @Test + public void activatedGetVersionCallReplaysOnUnactivatedNewWorker() throws Exception { + String taskQueue = "preferred-version-provider-rollout-" + UUID.randomUUID(); + try (TestWorkflowEnvironment testEnvironment = TestWorkflowEnvironment.newInstance()) { + WorkerFactory activatedWorkerFactory = + WorkerFactory.newInstance(newClient(testEnvironment, "activated-rollout-worker")); + Worker activatedWorker = + activatedWorkerFactory.newWorker( + taskQueue, + workerOptions( + (input) -> { + activatedProviderCalls.incrementAndGet(); + return Optional.of(VersionPreference.of(1)); + })); + activatedWorker.registerWorkflowImplementationTypes(NewRolloutWorkflowImpl.class); + + testEnvironment.start(); + activatedWorkerFactory.start(); + + RolloutWorkflow workflow = newWorkflowStub(testEnvironment, taskQueue); + WorkflowClient.start(workflow::run); + + assertEventually(Duration.ofSeconds(5), () -> assertEquals(1, activatedProviderCalls.get())); + assertEquals("new-0", workflow.state()); + + shutdown(activatedWorkerFactory); + + WorkerFactory unactivatedWorkerFactory = + WorkerFactory.newInstance(newClient(testEnvironment, "unactivated-rollout-worker")); + Worker unactivatedWorker = + unactivatedWorkerFactory.newWorker( + taskQueue, + workerOptions( + (input) -> { + unactivatedProviderCalls.incrementAndGet(); + return Optional.of(VersionPreference.of(Workflow.DEFAULT_VERSION)); + })); + unactivatedWorker.registerWorkflowImplementationTypes( + UnactivatedNewRolloutWorkflowImpl.class); + unactivatedWorkerFactory.start(); + + workflow.release(); + testEnvironment.sleep(Duration.ofSeconds(1)); + + assertEventually( + Duration.ofSeconds(5), () -> assertEquals(1, unactivatedNewWorkerSignals.get())); + + workflow.release(); + assertEquals("new", WorkflowStub.fromTyped(workflow).getResult(String.class)); + assertEquals(0, unactivatedProviderCalls.get()); + shutdown(unactivatedWorkerFactory); + } + } + + private static RolloutWorkflow newWorkflowStub( + TestWorkflowEnvironment testEnvironment, String taskQueue) { + return testEnvironment + .getWorkflowClient() + .newWorkflowStub( + RolloutWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(taskQueue) + .setWorkflowId(UUID.randomUUID().toString()) + .build()); + } + + private static WorkerOptions workerOptions(PreferredVersionProvider preferredVersionProvider) { + return WorkerOptions.newBuilder() + .setStickyQueueScheduleToStartTimeout(Duration.ZERO) + .setPreferredVersionProvider(preferredVersionProvider) + .build(); + } + + private static WorkflowClient newClient( + TestWorkflowEnvironment testEnvironment, String identity) { + WorkflowClientOptions clientOptions = + testEnvironment.getWorkflowClient().getOptions().toBuilder() + .setIdentity(identity + "-" + UUID.randomUUID()) + .build(); + return WorkflowClient.newInstance(testEnvironment.getWorkflowServiceStubs(), clientOptions); + } + + private static void shutdown(WorkerFactory workerFactory) throws InterruptedException { + workerFactory.shutdownNow(); + workerFactory.awaitTermination(10, TimeUnit.SECONDS); + } + + @WorkflowInterface + public interface RolloutWorkflow { + @WorkflowMethod + String run(); + + @SignalMethod + void release(); + + @QueryMethod + String state(); + } + + public static class OldRolloutWorkflowImpl implements RolloutWorkflow { + private int releases; + + @Override + public String run() { + Workflow.await(() -> releases >= 1); + Workflow.await(() -> releases >= 2); + return "old"; + } + + @Override + public void release() { + oldWorkerSignals.incrementAndGet(); + releases++; + } + + @Override + public String state() { + return "old-" + releases; + } + } + + public static class NewRolloutWorkflowImpl implements RolloutWorkflow { + private int releases; + + @Override + public String run() { + int version = Workflow.getVersion(CHANGE_ID, Workflow.DEFAULT_VERSION, 1); + Workflow.await(() -> releases >= 1); + assertEquals(version, Workflow.getVersion(CHANGE_ID, Workflow.DEFAULT_VERSION, 1)); + Workflow.await(() -> releases >= 2); + return version == Workflow.DEFAULT_VERSION ? "old" : "new"; + } + + @Override + public void release() { + releases++; + } + + @Override + public String state() { + return "new-" + releases; + } + } + + public static class UnactivatedNewRolloutWorkflowImpl extends NewRolloutWorkflowImpl { + @Override + public void release() { + unactivatedNewWorkerSignals.incrementAndGet(); + super.release(); + } + } +} From 26705eb5788e07f40bec9ff108e928c6bca4c22b Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Wed, 8 Jul 2026 10:17:03 -0700 Subject: [PATCH 3/4] Address review comments --- .../replay/ReplayWorkflowContextImpl.java | 10 ++-- .../statemachines/VersionStateMachine.java | 18 +++---- .../statemachines/WorkflowStateMachines.java | 2 +- .../internal/sync/WorkflowInternal.java | 11 +++++ .../worker/PreferredVersionProvider.java | 10 ++-- .../VersionStateMachineTest.java | 18 +++---- .../io/temporal/worker/WorkerOptionsTest.java | 2 +- .../PreferredVersionProviderRolloutTest.java | 7 ++- .../PreferredVersionProviderTest.java | 47 ++++++++++++++++++- 9 files changed, 85 insertions(+), 40 deletions(-) diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java index 17cea576d5..2f600b20aa 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java @@ -15,6 +15,7 @@ import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.common.SdkFlag; import io.temporal.internal.statemachines.*; +import io.temporal.internal.sync.WorkflowInternal; import io.temporal.internal.worker.SingleWorkerOptions; import io.temporal.worker.PreferredVersionProvider; import io.temporal.worker.PreferredVersionProviderInput; @@ -40,7 +41,6 @@ final class ReplayWorkflowContextImpl implements ReplayWorkflowContext { private final @Nullable String fullReplayDirectQueryName; private final Scope replayAwareWorkflowMetricsScope; private final SingleWorkerOptions workerOptions; - private final WorkflowInfoImpl workflowInfo; /** * @param fullReplayDirectQueryName query name if an execution is a full replay caused by a direct @@ -64,7 +64,6 @@ final class ReplayWorkflowContextImpl implements ReplayWorkflowContext { this.replayAwareWorkflowMetricsScope = new ReplayAwareScope(workflowMetricsScope, this, workflowStateMachines::currentTimeMillis); this.workerOptions = workerOptions; - this.workflowInfo = new WorkflowInfoImpl(this); } @Override @@ -349,8 +348,11 @@ public Integer getVersion( preferredVersionProvider == null ? null : (min, max) -> - preferredVersionProvider.getPreferredVersion( - new PreferredVersionProviderInput(workflowInfo, changeId, min, max)), + WorkflowInternal.readOnly( + () -> + preferredVersionProvider.getPreferredVersion( + new PreferredVersionProviderInput( + WorkflowInternal.getWorkflowInfo(), changeId, min, max))), callback); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/VersionStateMachine.java b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/VersionStateMachine.java index 73874df05c..addb326482 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/VersionStateMachine.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/VersionStateMachine.java @@ -17,7 +17,6 @@ import io.temporal.worker.VersionPreference; import io.temporal.workflow.Functions; import java.util.Objects; -import java.util.Optional; import java.util.function.BiFunction; import javax.annotation.Nullable; @@ -141,16 +140,14 @@ class InvocationStateMachine private final Functions.Proc2 resultCallback; @Nullable - private final BiFunction> - preferredVersionProvider; + private final BiFunction preferredVersionProvider; InvocationStateMachine( int minSupported, int maxSupported, boolean waitForMarkerRecordedReplaying, Functions.Func1 upsertSearchAttributeCallback, - @Nullable - BiFunction> preferredVersionProvider, + @Nullable BiFunction preferredVersionProvider, Functions.Proc2 callback) { super(STATE_MACHINE_DEFINITION, VersionStateMachine.this.commandSink, stateMachineSink); this.minSupported = minSupported; @@ -271,15 +268,12 @@ private int resolveVersionExecuting() { if (preferredVersionProvider == null) { return maxSupported; } - Optional preference = - Objects.requireNonNull( - preferredVersionProvider.apply(minSupported, maxSupported), - "PreferredVersionProvider returned null"); - if (!preference.isPresent()) { + VersionPreference versionPreference = + preferredVersionProvider.apply(minSupported, maxSupported); + if (versionPreference == null) { return maxSupported; } - VersionPreference versionPreference = preference.get(); int preferredVersion = versionPreference.getVersion(); if (preferredVersion >= minSupported && preferredVersion <= maxSupported) { return preferredVersion; @@ -449,7 +443,7 @@ public Integer getVersion( int maxSupported, boolean waitForMarkerRecordedReplaying, Functions.Func1 upsertSearchAttributeCallback, - @Nullable BiFunction> preferredVersionProvider, + @Nullable BiFunction preferredVersionProvider, Functions.Proc2 callback) { InvocationStateMachine ism = new InvocationStateMachine( diff --git a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java index e5b0e7d701..2de6b6ea15 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java @@ -1253,7 +1253,7 @@ public Integer getVersion( String changeId, int minSupported, int maxSupported, - @Nullable BiFunction> preferredVersionProvider, + @Nullable BiFunction preferredVersionProvider, Functions.Proc2 callback) { VersionStateMachine stateMachine = versions.computeIfAbsent( diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java index 3656c29d80..92429c21ef 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java @@ -675,6 +675,17 @@ public static T deadlockDetectorOff(Functions.Func func) { } } + public static T readOnly(Functions.Func func) { + SyncWorkflowContext workflowContext = getRootWorkflowContext(); + boolean previousReadOnly = workflowContext.isReadOnly(); + workflowContext.setReadOnly(true); + try { + return func.apply(); + } finally { + workflowContext.setReadOnly(previousReadOnly); + } + } + public static WorkflowInfo getWorkflowInfo() { return new WorkflowInfoImpl(getRootWorkflowContext().getReplayContext()); } diff --git a/temporal-sdk/src/main/java/io/temporal/worker/PreferredVersionProvider.java b/temporal-sdk/src/main/java/io/temporal/worker/PreferredVersionProvider.java index 1412d0d22e..cd8351c5af 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/PreferredVersionProvider.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/PreferredVersionProvider.java @@ -1,8 +1,8 @@ package io.temporal.worker; import io.temporal.common.Experimental; -import java.util.Optional; import javax.annotation.Nonnull; +import javax.annotation.Nullable; /** * Provides the version to record for a {@link io.temporal.workflow.Workflow#getVersion(String, int, @@ -17,14 +17,14 @@ public interface PreferredVersionProvider { /** - * Returns the preferred version for the supplied {@code getVersion} call, or {@link - * Optional#empty()} to use the SDK default of {@code maxSupported}. + * Returns the preferred version for the supplied {@code getVersion} call, or {@code null} to use + * the SDK default of {@code maxSupported}. * *

This method is invoked on the workflow thread. Any exception it throws propagates out of the * {@code getVersion} call and fails the current workflow task, which will be retried; a provider * that always throws will therefore block the workflow. Implementations should be deterministic * and side-effect free. */ - @Nonnull - Optional getPreferredVersion(@Nonnull PreferredVersionProviderInput input); + @Nullable + VersionPreference getPreferredVersion(@Nonnull PreferredVersionProviderInput input); } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/statemachines/VersionStateMachineTest.java b/temporal-sdk/src/test/java/io/temporal/internal/statemachines/VersionStateMachineTest.java index d7c91892f6..5a6d54f625 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/statemachines/VersionStateMachineTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/statemachines/VersionStateMachineTest.java @@ -56,7 +56,7 @@ public void buildWorkflow(AsyncWorkflowBuilder builder) { "id1", DEFAULT_VERSION, 12, - (min, max) -> Optional.of(VersionPreference.of(preferredVersion)), + (min, max) -> VersionPreference.of(preferredVersion), c)) .add((v) -> stateMachines.completeWorkflow(converter.toPayloads(v.getT1()))); } @@ -92,7 +92,7 @@ public void buildWorkflow(AsyncWorkflowBuilder builder) { .add2( (v, c) -> stateMachines.getVersion( - "id1", DEFAULT_VERSION, maxSupported, (min, max) -> Optional.empty(), c)) + "id1", DEFAULT_VERSION, maxSupported, (min, max) -> null, c)) .add((v) -> stateMachines.completeWorkflow(converter.toPayloads(v.getT1()))); } } @@ -123,7 +123,7 @@ public void buildWorkflow(AsyncWorkflowBuilder builder) { 12, (min, max) -> { providerCalls.incrementAndGet(); - return Optional.of(VersionPreference.of(3)); + return VersionPreference.of(3); }, c)) .add2( @@ -134,7 +134,7 @@ public void buildWorkflow(AsyncWorkflowBuilder builder) { 12, (min, max) -> { providerCalls.incrementAndGet(); - return Optional.of(VersionPreference.of(4)); + return VersionPreference.of(4); }, c)) .add((v) -> stateMachines.completeWorkflow(converter.toPayloads(v.getT1()))); @@ -161,11 +161,7 @@ public void buildWorkflow(AsyncWorkflowBuilder builder) { builder.add2( (v, c) -> stateMachines.getVersion( - "id1", - DEFAULT_VERSION, - 1, - (min, max) -> Optional.of(VersionPreference.of(2)), - c)); + "id1", DEFAULT_VERSION, 1, (min, max) -> VersionPreference.of(2), c)); } } @@ -204,7 +200,7 @@ public void buildWorkflow(AsyncWorkflowBuilder builder) { "id1", DEFAULT_VERSION, 1, - (min, max) -> Optional.of(VersionPreference.of(2).clampToSupportedRange()), + (min, max) -> VersionPreference.of(2).clampToSupportedRange(), c)) .add((v) -> stateMachines.completeWorkflow(converter.toPayloads(v.getT1()))); } @@ -236,7 +232,7 @@ public void buildWorkflow(AsyncWorkflowBuilder builder) { 12, (min, max) -> { providerCalls.incrementAndGet(); - return Optional.of(VersionPreference.of(4)); + return VersionPreference.of(4); }, c)) .add((v) -> stateMachines.completeWorkflow(converter.toPayloads(v.getT1()))); diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java index 9787c45ace..877f6fdce8 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java @@ -14,7 +14,7 @@ public void build() { WorkerOptions.newBuilder() .setMaxConcurrentActivityExecutionSize(10) .setMaxConcurrentLocalActivityExecutionSize(11) - .setPreferredVersionProvider((input) -> java.util.Optional.empty()); + .setPreferredVersionProvider((input) -> null); verifyBuild(builder.build()); verifyBuild(builder.validateAndBuildWithDefaults()); diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderRolloutTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderRolloutTest.java index d54c7f803a..cfdb1b9b5f 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderRolloutTest.java +++ b/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderRolloutTest.java @@ -19,7 +19,6 @@ import io.temporal.workflow.WorkflowInterface; import io.temporal.workflow.WorkflowMethod; import java.time.Duration; -import java.util.Optional; import java.util.UUID; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -53,7 +52,7 @@ public void unactivatedNewGetVersionCallReplaysOnOldWorkerWithoutTheCall() throw workerOptions( (input) -> { unactivatedProviderCalls.incrementAndGet(); - return Optional.of(VersionPreference.of(Workflow.DEFAULT_VERSION)); + return VersionPreference.of(Workflow.DEFAULT_VERSION); })); newWorker.registerWorkflowImplementationTypes(NewRolloutWorkflowImpl.class); @@ -98,7 +97,7 @@ public void activatedGetVersionCallReplaysOnUnactivatedNewWorker() throws Except workerOptions( (input) -> { activatedProviderCalls.incrementAndGet(); - return Optional.of(VersionPreference.of(1)); + return VersionPreference.of(1); })); activatedWorker.registerWorkflowImplementationTypes(NewRolloutWorkflowImpl.class); @@ -121,7 +120,7 @@ public void activatedGetVersionCallReplaysOnUnactivatedNewWorker() throws Except workerOptions( (input) -> { unactivatedProviderCalls.incrementAndGet(); - return Optional.of(VersionPreference.of(Workflow.DEFAULT_VERSION)); + return VersionPreference.of(Workflow.DEFAULT_VERSION); })); unactivatedWorker.registerWorkflowImplementationTypes( UnactivatedNewRolloutWorkflowImpl.class); diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderTest.java index ea85afab7c..f3da4f7e76 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderTest.java +++ b/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderTest.java @@ -3,15 +3,19 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import io.temporal.client.WorkflowOptions; import io.temporal.client.WorkflowStub; +import io.temporal.internal.sync.ReadOnlyException; +import io.temporal.testing.TestWorkflowEnvironment; import io.temporal.testing.internal.SDKTestWorkflowRule; import io.temporal.worker.PreferredVersionProviderInput; import io.temporal.worker.VersionPreference; +import io.temporal.worker.Worker; import io.temporal.worker.WorkerOptions; import io.temporal.workflow.Workflow; import io.temporal.workflow.shared.TestWorkflows.TestWorkflowReturnString; import java.time.Duration; -import java.util.Optional; +import java.util.UUID; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.junit.Before; @@ -35,7 +39,7 @@ public class PreferredVersionProviderTest { (input) -> { providerCalls.incrementAndGet(); providerInput.set(input); - return Optional.of(VersionPreference.of(Workflow.DEFAULT_VERSION)); + return VersionPreference.of(Workflow.DEFAULT_VERSION); }) .build()) .build(); @@ -66,6 +70,45 @@ public void providerReceivesGetVersionInputAndIsNotCalledOnReplay() { input.getWorkflowInfo().getWorkflowId()); } + @Test + public void providerRunsInReadOnlyContext() { + String taskQueue = "preferred-version-provider-read-only-" + UUID.randomUUID(); + AtomicReference readOnlyException = new AtomicReference<>(); + try (TestWorkflowEnvironment testEnvironment = TestWorkflowEnvironment.newInstance()) { + Worker worker = + testEnvironment.newWorker( + taskQueue, + WorkerOptions.newBuilder() + .setPreferredVersionProvider( + (input) -> { + try { + Workflow.randomUUID(); + } catch (ReadOnlyException e) { + readOnlyException.set(e); + return VersionPreference.of(Workflow.DEFAULT_VERSION); + } + throw new AssertionError("provider should run in a read-only context"); + }) + .build()); + worker.registerWorkflowImplementationTypes(TestPreferredVersionWorkflow.class); + testEnvironment.start(); + + TestWorkflowReturnString workflow = + testEnvironment + .getWorkflowClient() + .newWorkflowStub( + TestWorkflowReturnString.class, + WorkflowOptions.newBuilder().setTaskQueue(taskQueue).build()); + + assertEquals("old", workflow.execute()); + } + + assertNotNull(readOnlyException.get()); + assertEquals( + "While in read-only function, action attempted: random UUID", + readOnlyException.get().getMessage()); + } + public static class TestPreferredVersionWorkflow implements TestWorkflowReturnString { @Override public String execute() { From 94c226f31c6967ec184e941897663ecd2bd536f7 Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Wed, 8 Jul 2026 11:01:36 -0700 Subject: [PATCH 4/4] Undo pubify of info impl --- .../internal/{replay => sync}/WorkflowInfoImpl.java | 7 ++++--- .../java/io/temporal/internal/sync/WorkflowInternal.java | 1 - 2 files changed, 4 insertions(+), 4 deletions(-) rename temporal-sdk/src/main/java/io/temporal/internal/{replay => sync}/WorkflowInfoImpl.java (96%) diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/WorkflowInfoImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInfoImpl.java similarity index 96% rename from temporal-sdk/src/main/java/io/temporal/internal/replay/WorkflowInfoImpl.java rename to temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInfoImpl.java index d51df7bc25..b30f13d458 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/WorkflowInfoImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInfoImpl.java @@ -1,4 +1,4 @@ -package io.temporal.internal.replay; +package io.temporal.internal.sync; import io.temporal.api.common.v1.SearchAttributes; import io.temporal.api.common.v1.WorkflowExecution; @@ -6,6 +6,7 @@ import io.temporal.common.RetryOptions; import io.temporal.common.SuggestContinueAsNewReason; import io.temporal.internal.common.ProtoConverters; +import io.temporal.internal.replay.ReplayWorkflowContext; import io.temporal.workflow.WorkflowInfo; import java.time.Duration; import java.util.List; @@ -13,11 +14,11 @@ import javax.annotation.Nonnull; import javax.annotation.Nullable; -public final class WorkflowInfoImpl implements WorkflowInfo { +final class WorkflowInfoImpl implements WorkflowInfo { private final ReplayWorkflowContext context; - public WorkflowInfoImpl(ReplayWorkflowContext context) { + WorkflowInfoImpl(ReplayWorkflowContext context) { this.context = context; } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java index 92429c21ef..3ea979e8db 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java @@ -27,7 +27,6 @@ import io.temporal.internal.common.SdkFlag; import io.temporal.internal.common.SearchAttributesUtil; import io.temporal.internal.logging.ReplayAwareLogger; -import io.temporal.internal.replay.WorkflowInfoImpl; import io.temporal.internal.statemachines.UnsupportedContinueAsNewRequest; import io.temporal.serviceclient.CheckedExceptionWrapper; import io.temporal.workflow.*;