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..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,7 +15,10 @@ 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; import io.temporal.workflow.Functions; import io.temporal.workflow.Functions.Func; import io.temporal.workflow.Functions.Func1; @@ -337,7 +340,20 @@ 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) -> + WorkflowInternal.readOnly( + () -> + preferredVersionProvider.getPreferredVersion( + new PreferredVersionProviderInput( + WorkflowInternal.getWorkflowInfo(), changeId, min, max))), + callback); } @Override 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..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 @@ -14,8 +14,10 @@ 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.function.BiFunction; import javax.annotation.Nullable; final class VersionStateMachine { @@ -137,17 +139,22 @@ 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 +217,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 +238,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 +249,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 +264,27 @@ State createMarkerExecuting() { } } + private int resolveVersionExecuting() { + if (preferredVersionProvider == null) { + return maxSupported; + } + VersionPreference versionPreference = + preferredVersionProvider.apply(minSupported, maxSupported); + if (versionPreference == null) { + return maxSupported; + } + + 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 +443,7 @@ public Integer getVersion( int maxSupported, boolean waitForMarkerRecordedReplaying, Functions.Func1 upsertSearchAttributeCallback, + @Nullable BiFunction preferredVersionProvider, Functions.Proc2 callback) { InvocationStateMachine ism = new InvocationStateMachine( @@ -418,6 +451,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..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 @@ -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..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 @@ -674,6 +674,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/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..cd8351c5af --- /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 javax.annotation.Nonnull; +import javax.annotation.Nullable; + +/** + * 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 {@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. + */ + @Nullable + VersionPreference 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..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 @@ -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,235 @@ 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) -> 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) -> null, 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 VersionPreference.of(3); + }, + c)) + .add2( + (v, c) -> + stateMachines.getVersion( + "id1", + DEFAULT_VERSION, + 12, + (min, max) -> { + providerCalls.incrementAndGet(); + return 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) -> 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) -> 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 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..877f6fdce8 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) -> null); 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/PreferredVersionProviderRolloutTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderRolloutTest.java new file mode 100644 index 0000000000..cfdb1b9b5f --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderRolloutTest.java @@ -0,0 +1,239 @@ +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.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 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 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 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(); + } + } +} 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..f3da4f7e76 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/versionTests/PreferredVersionProviderTest.java @@ -0,0 +1,121 @@ +package io.temporal.workflow.versionTests; + +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.UUID; +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 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()); + } + + @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() { + 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"; + } + } +}