diff --git a/CHANGELOG.md b/CHANGELOG.md index 00493c2f93..ae777c2de2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,9 @@ to docs, or any other relevant information. to the given name. - Added experimental `WorkflowUnsafe.isSubjectToReplay()`, which reports whether the calling code is re-executed on replay. +- Added experimental `ActivityClientPlugin` and `NexusClientPlugin`, which plugins on the service stubs use to configure + `ActivityClient` and `NexusClient`. `SimplePlugin` implements both, and its builder gained + `addScheduleClientInterceptors`, `addActivityClientInterceptors`, and `addNexusClientInterceptors`. ### Changed - Release notes for all future releases are now in a single CHANGELOG.md file. `releases` directory with old release @@ -43,6 +46,8 @@ to docs, or any other relevant information. - `TemporalDevServer`, `TemporalDevServerOptions`, `TestWorkflowEnvironment.startLocal`, and the `useDevServer` methods of `TestWorkflowRule` and `TestWorkflowExtension` are no longer marked experimental. `TemporalDevServerOptions` remain specific to the Temporal CLI dev server implementation and may no longer be valid if that implementation changes. +- `SimplePlugin` now applies its data converter customizer to schedule and Nexus clients, and its context + propagators to schedule and activity clients. Previously `SimplePlugin.configureScheduleClient` did nothing. ### Fixed - Kotlin `Async` method references compiled with Kotlin 2.4 now retain their inline behavior, diff --git a/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java b/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java index 77bf2bd794..7cbddd9190 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java @@ -11,6 +11,7 @@ import io.temporal.internal.client.RootActivityClientInvoker; import io.temporal.internal.client.external.GenericWorkflowClientImpl; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; +import io.temporal.internal.common.PluginUtils; import io.temporal.internal.util.MethodExtractor; import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; @@ -23,6 +24,8 @@ import java.util.Map; import java.util.stream.Stream; import javax.annotation.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Implementation of {@link ActivityClient} that delegates calls through the activity interceptor @@ -30,6 +33,8 @@ */ class ActivityClientImpl implements ActivityClient, ActivityClientInternal { + private static final Logger log = LoggerFactory.getLogger(ActivityClientImpl.class); + private final WorkflowServiceStubs stubs; private final ActivityClientOptions options; private final ActivityClientCallsInterceptor invoker; @@ -37,6 +42,33 @@ class ActivityClientImpl implements ActivityClient, ActivityClientInternal { private final Scope metricsScope; ActivityClientImpl(WorkflowServiceStubs stubs, ActivityClientOptions options) { + // Extract the ActivityClientPlugins propagated from the service stubs plugins. + ActivityClientPlugin[] propagatedPlugins = + PluginUtils.extractPlugins( + stubs.getOptions().getPlugins(), + ActivityClientPlugin.class, + ActivityClientPlugin[]::new); + + // Merge propagated plugins with activity client-specified plugins. + ActivityClientPlugin[] mergedPlugins = + PluginUtils.mergePlugins( + propagatedPlugins, + options.getPlugins(), + ActivityClientPlugin::getName, + log, + "service stubs", + ActivityClientPlugin.class); + + // Apply plugin configuration phase (forward order) on user-provided options, + // so plugins see unmodified state before defaults and plugin merging. + ActivityClientOptions.Builder builder = ActivityClientOptions.newBuilder(options); + for (ActivityClientPlugin plugin : mergedPlugins) { + plugin.configureActivityClient(builder); + } + // Set merged plugins after configuration, then build. + builder.setPlugins(mergedPlugins); + options = builder.build(); + this.stubs = stubs; this.options = options; this.metricsScope = diff --git a/temporal-sdk/src/main/java/io/temporal/client/ActivityClientOptions.java b/temporal-sdk/src/main/java/io/temporal/client/ActivityClientOptions.java index 5298f81140..8962d2ab6c 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/ActivityClientOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/client/ActivityClientOptions.java @@ -1,10 +1,12 @@ package io.temporal.client; +import io.temporal.common.Experimental; import io.temporal.common.context.ContextPropagator; import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.GlobalDataConverter; import io.temporal.common.interceptors.ActivityClientInterceptor; import java.lang.management.ManagementFactory; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Objects; @@ -40,12 +42,14 @@ public static final class Builder { Collections.emptyList(); private static final List EMPTY_INTERCEPTORS = Collections.emptyList(); + private static final ActivityClientPlugin[] EMPTY_PLUGINS = new ActivityClientPlugin[0]; private String namespace; private DataConverter dataConverter; private String identity; private List contextPropagators; private List interceptors; + private ActivityClientPlugin[] plugins; private Builder() {} @@ -58,6 +62,7 @@ private Builder(ActivityClientOptions options) { identity = options.identity; contextPropagators = options.contextPropagators; interceptors = options.interceptors; + plugins = options.plugins; } /** Set the namespace this client will operate on. */ @@ -102,6 +107,17 @@ public Builder setInterceptors(List interceptors) { return this; } + /** + * Set the plugins for this client. + * + * @param plugins specifies the plugins to use with the client. + */ + @Experimental + public Builder setPlugins(ActivityClientPlugin... plugins) { + this.plugins = Objects.requireNonNull(plugins); + return this; + } + public ActivityClientOptions build() { String name = identity == null ? ManagementFactory.getRuntimeMXBean().getName() : identity; return new ActivityClientOptions( @@ -109,7 +125,8 @@ public ActivityClientOptions build() { dataConverter == null ? GlobalDataConverter.get() : dataConverter, name, contextPropagators == null ? EMPTY_CONTEXT_PROPAGATORS : contextPropagators, - interceptors == null ? EMPTY_INTERCEPTORS : interceptors); + interceptors == null ? EMPTY_INTERCEPTORS : interceptors, + plugins == null ? EMPTY_PLUGINS : plugins); } } @@ -118,18 +135,21 @@ public ActivityClientOptions build() { private final String identity; private final List contextPropagators; private final List interceptors; + private final ActivityClientPlugin[] plugins; private ActivityClientOptions( String namespace, DataConverter dataConverter, String identity, List contextPropagators, - List interceptors) { + List interceptors, + ActivityClientPlugin[] plugins) { this.namespace = namespace; this.dataConverter = dataConverter; this.identity = identity; this.contextPropagators = contextPropagators; this.interceptors = interceptors; + this.plugins = plugins; } /** @@ -177,6 +197,16 @@ public List getInterceptors() { return interceptors; } + /** + * Get the plugins of this client. + * + * @return The plugins to use with the client. + */ + @Experimental + public ActivityClientPlugin[] getPlugins() { + return plugins; + } + @Override public boolean equals(Object o) { if (this == o) return true; @@ -186,12 +216,19 @@ public boolean equals(Object o) { && Objects.equals(dataConverter, that.dataConverter) && Objects.equals(identity, that.identity) && Objects.equals(contextPropagators, that.contextPropagators) - && Objects.equals(interceptors, that.interceptors); + && Objects.equals(interceptors, that.interceptors) + && Arrays.equals(plugins, that.plugins); } @Override public int hashCode() { - return Objects.hash(namespace, dataConverter, identity, contextPropagators, interceptors); + return Objects.hash( + namespace, + dataConverter, + identity, + contextPropagators, + interceptors, + Arrays.hashCode(plugins)); } @Override @@ -209,6 +246,8 @@ public String toString() { + contextPropagators + ", interceptors=" + interceptors + + ", plugins=" + + Arrays.toString(plugins) + '}'; } } diff --git a/temporal-sdk/src/main/java/io/temporal/client/ActivityClientPlugin.java b/temporal-sdk/src/main/java/io/temporal/client/ActivityClientPlugin.java new file mode 100644 index 0000000000..a129464c00 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/client/ActivityClientPlugin.java @@ -0,0 +1,34 @@ +package io.temporal.client; + +import io.temporal.common.Experimental; +import javax.annotation.Nonnull; + +/** + * Plugin interface for customizing Temporal activity client configuration. + * + *

Plugins that implement both {@link io.temporal.serviceclient.WorkflowServiceStubsPlugin} and + * {@code ActivityClientPlugin} are automatically propagated from the service stubs to the activity + * client. + * + * @see io.temporal.serviceclient.WorkflowServiceStubsPlugin + */ +@Experimental +public interface ActivityClientPlugin { + + /** + * Returns a unique name for this plugin. Used for logging and duplicate detection. Recommended + * format: "organization.plugin-name" (e.g., "io.temporal.tracing") + * + * @return fully qualified plugin name + */ + @Nonnull + String getName(); + + /** + * Allows the plugin to modify activity client options before the client is created. Called during + * configuration phase in forward (registration) order. + * + * @param builder the options builder to modify + */ + void configureActivityClient(@Nonnull ActivityClientOptions.Builder builder); +} diff --git a/temporal-sdk/src/main/java/io/temporal/client/NexusClientImpl.java b/temporal-sdk/src/main/java/io/temporal/client/NexusClientImpl.java index a9248e80b1..685fc1aee1 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/NexusClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/NexusClientImpl.java @@ -17,6 +17,7 @@ import io.temporal.internal.client.RootNexusClientInvoker; import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.client.external.GenericWorkflowClientImpl; +import io.temporal.internal.common.PluginUtils; import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; import java.util.List; @@ -39,6 +40,32 @@ public class NexusClientImpl implements NexusClient { public static NexusClient newInstance(WorkflowServiceStubs service, NexusClientOptions options) { enforceNonWorkflowThread(); + + // Extract the NexusClientPlugins propagated from the service stubs plugins. + NexusClientPlugin[] propagatedPlugins = + PluginUtils.extractPlugins( + service.getOptions().getPlugins(), NexusClientPlugin.class, NexusClientPlugin[]::new); + + // Merge propagated plugins with Nexus client-specified plugins. + NexusClientPlugin[] mergedPlugins = + PluginUtils.mergePlugins( + propagatedPlugins, + options.getPlugins(), + NexusClientPlugin::getName, + log, + "service stubs", + NexusClientPlugin.class); + + // Apply plugin configuration phase (forward order) on user-provided options, + // so plugins see unmodified state before defaults and plugin merging. + NexusClientOptions.Builder builder = NexusClientOptions.newBuilder(options); + for (NexusClientPlugin plugin : mergedPlugins) { + plugin.configureNexusClient(builder); + } + // Set merged plugins after configuration, then build. + builder.setPlugins(mergedPlugins); + options = builder.build(); + return WorkflowThreadMarker.protectFromWorkflowThread( new NexusClientImpl(service, options.toResolvedOptions()), NexusClient.class); } diff --git a/temporal-sdk/src/main/java/io/temporal/client/NexusClientOptions.java b/temporal-sdk/src/main/java/io/temporal/client/NexusClientOptions.java index 414cd3c568..bf384a28ae 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/NexusClientOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/client/NexusClientOptions.java @@ -11,6 +11,7 @@ import java.lang.management.ManagementFactory; import java.util.Collections; import java.util.List; +import java.util.Objects; import javax.annotation.Nullable; /** @@ -42,18 +43,21 @@ public class NexusClientOptions { private final DataConverter dataConverter; private final String identity; private final @Nullable ExternalStorage externalStorage; + private final NexusClientPlugin[] plugins; private NexusClientOptions( String namespace, List interceptors, DataConverter dataConverter, String identity, - @Nullable ExternalStorage externalStorage) { + @Nullable ExternalStorage externalStorage, + NexusClientPlugin[] plugins) { this.namespace = namespace; this.interceptors = interceptors; this.dataConverter = dataConverter; this.identity = identity; this.externalStorage = externalStorage; + this.plugins = plugins; } /** Get the namespace this client will operate on. */ @@ -88,6 +92,11 @@ public String getIdentity() { return identity; } + /** Get the plugins of this client. */ + public NexusClientPlugin[] getPlugins() { + return plugins; + } + /** * Converts this {@link NexusClientOptions} instance into a {@link NexusClientResolvedOptions} * instance, which contains the fully resolved runtime settings used by the internal Nexus client. @@ -130,11 +139,14 @@ public static NexusClientOptions getDefaultInstance() { /** Builder for {@link NexusClientOptions}. */ public static class Builder { + private static final NexusClientPlugin[] EMPTY_PLUGINS = new NexusClientPlugin[0]; + private String namespace; private List interceptors = Collections.emptyList(); private DataConverter dataConverter = GlobalDataConverter.get(); private String identity; private ExternalStorage externalStorage; + private NexusClientPlugin[] plugins; private Builder() {} @@ -147,6 +159,7 @@ private Builder(NexusClientOptions options) { dataConverter = options.dataConverter; identity = options.identity; externalStorage = options.externalStorage; + plugins = options.plugins; } /** Set the namespace this client will operate on. */ @@ -195,6 +208,16 @@ public NexusClientOptions.Builder setExternalStorage( return this; } + /** + * Set the plugins for this client. + * + * @param plugins specifies the plugins to use with the client. + */ + public NexusClientOptions.Builder setPlugins(NexusClientPlugin... plugins) { + this.plugins = Objects.requireNonNull(plugins); + return this; + } + public NexusClientOptions build() { String resolvedIdentity = identity == null ? ManagementFactory.getRuntimeMXBean().getName() : identity; @@ -203,7 +226,8 @@ public NexusClientOptions build() { interceptors, dataConverter, resolvedIdentity, - externalStorage); + externalStorage, + plugins == null ? EMPTY_PLUGINS : plugins); } } } diff --git a/temporal-sdk/src/main/java/io/temporal/client/NexusClientPlugin.java b/temporal-sdk/src/main/java/io/temporal/client/NexusClientPlugin.java new file mode 100644 index 0000000000..b73ae83545 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/client/NexusClientPlugin.java @@ -0,0 +1,34 @@ +package io.temporal.client; + +import io.temporal.common.Experimental; +import javax.annotation.Nonnull; + +/** + * Plugin interface for customizing Temporal Nexus client configuration. + * + *

Plugins that implement both {@link io.temporal.serviceclient.WorkflowServiceStubsPlugin} and + * {@code NexusClientPlugin} are automatically propagated from the service stubs to the Nexus + * client. + * + * @see io.temporal.serviceclient.WorkflowServiceStubsPlugin + */ +@Experimental +public interface NexusClientPlugin { + + /** + * Returns a unique name for this plugin. Used for logging and duplicate detection. Recommended + * format: "organization.plugin-name" (e.g., "io.temporal.tracing") + * + * @return fully qualified plugin name + */ + @Nonnull + String getName(); + + /** + * Allows the plugin to modify Nexus client options before the client is created. Called during + * configuration phase in forward (registration) order. + * + * @param builder the options builder to modify + */ + void configureNexusClient(@Nonnull NexusClientOptions.Builder builder); +} diff --git a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java index a3b92aa219..1df272164f 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java @@ -30,7 +30,6 @@ import io.temporal.payload.storage.ExternalStorage; import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; -import io.temporal.serviceclient.WorkflowServiceStubsPlugin; import io.temporal.worker.WorkerFactory; import io.temporal.workflow.*; import java.lang.annotation.Annotation; @@ -82,7 +81,10 @@ public static WorkflowClient newInstance( WorkflowServiceStubs workflowServiceStubs, WorkflowClientOptions options) { // Extract WorkflowClientPlugins from service stubs plugins (propagation) WorkflowClientPlugin[] propagatedPlugins = - extractClientPlugins(workflowServiceStubs.getOptions().getPlugins()); + PluginUtils.extractPlugins( + workflowServiceStubs.getOptions().getPlugins(), + WorkflowClientPlugin.class, + WorkflowClientPlugin[]::new); // Merge propagated plugins with client-specified plugins WorkflowClientPlugin[] mergedPlugins = @@ -854,23 +856,4 @@ public NexusStartWorkflowResponse startNexus( WorkflowInvocationHandler.closeAsyncInvocation(); } } - - /** - * Extracts WorkflowClientPlugins from service stubs plugins. Only plugins that also implement - * {@link WorkflowClientPlugin} are included. This enables plugin propagation from service stubs - * to workflow client. - */ - private static WorkflowClientPlugin[] extractClientPlugins( - WorkflowServiceStubsPlugin[] stubsPlugins) { - if (stubsPlugins == null || stubsPlugins.length == 0) { - return new WorkflowClientPlugin[0]; - } - List clientPlugins = new ArrayList<>(); - for (WorkflowServiceStubsPlugin plugin : stubsPlugins) { - if (plugin instanceof WorkflowClientPlugin) { - clientPlugins.add((WorkflowClientPlugin) plugin); - } - } - return clientPlugins.toArray(new WorkflowClientPlugin[0]); - } } diff --git a/temporal-sdk/src/main/java/io/temporal/client/schedules/ScheduleClientImpl.java b/temporal-sdk/src/main/java/io/temporal/client/schedules/ScheduleClientImpl.java index 9f96afe652..8076bd5ca7 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/schedules/ScheduleClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/schedules/ScheduleClientImpl.java @@ -13,8 +13,6 @@ import io.temporal.internal.common.PluginUtils; import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; -import io.temporal.serviceclient.WorkflowServiceStubsPlugin; -import java.util.ArrayList; import java.util.List; import java.util.stream.Stream; import javax.annotation.Nullable; @@ -50,7 +48,10 @@ public static ScheduleClient newInstance( ScheduleClientImpl(WorkflowServiceStubs workflowServiceStubs, ScheduleClientOptions options) { // Extract ScheduleClientPlugins from service stubs plugins (propagation) ScheduleClientPlugin[] propagatedPlugins = - extractScheduleClientPlugins(workflowServiceStubs.getOptions().getPlugins()); + PluginUtils.extractPlugins( + workflowServiceStubs.getOptions().getPlugins(), + ScheduleClientPlugin.class, + ScheduleClientPlugin[]::new); // Merge propagated plugins with schedule client-specified plugins ScheduleClientPlugin[] mergedPlugins = @@ -86,20 +87,6 @@ public static ScheduleClient newInstance( this.scheduleClientCallsInvoker = initializeClientInvoker(); } - private static ScheduleClientPlugin[] extractScheduleClientPlugins( - WorkflowServiceStubsPlugin[] stubsPlugins) { - if (stubsPlugins == null || stubsPlugins.length == 0) { - return new ScheduleClientPlugin[0]; - } - List schedulePlugins = new ArrayList<>(); - for (WorkflowServiceStubsPlugin plugin : stubsPlugins) { - if (plugin instanceof ScheduleClientPlugin) { - schedulePlugins.add((ScheduleClientPlugin) plugin); - } - } - return schedulePlugins.toArray(new ScheduleClientPlugin[0]); - } - private ScheduleClientCallsInterceptor initializeClientInvoker() { ScheduleClientCallsInterceptor scheduleClientInvoker = new RootScheduleClientInvoker(genericClient, options); diff --git a/temporal-sdk/src/main/java/io/temporal/common/SimplePlugin.java b/temporal-sdk/src/main/java/io/temporal/common/SimplePlugin.java index 0216a957db..f56697cd17 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/SimplePlugin.java +++ b/temporal-sdk/src/main/java/io/temporal/common/SimplePlugin.java @@ -20,12 +20,19 @@ package io.temporal.common; +import io.temporal.client.ActivityClientOptions; +import io.temporal.client.ActivityClientPlugin; +import io.temporal.client.NexusClientOptions; +import io.temporal.client.NexusClientPlugin; import io.temporal.client.WorkflowClientOptions; import io.temporal.client.WorkflowClientPlugin; import io.temporal.client.schedules.ScheduleClientOptions; import io.temporal.client.schedules.ScheduleClientPlugin; import io.temporal.common.context.ContextPropagator; import io.temporal.common.converter.DataConverter; +import io.temporal.common.interceptors.ActivityClientInterceptor; +import io.temporal.common.interceptors.NexusClientInterceptor; +import io.temporal.common.interceptors.ScheduleClientInterceptor; import io.temporal.common.interceptors.WorkerInterceptor; import io.temporal.common.interceptors.WorkflowClientInterceptor; import io.temporal.serviceclient.WorkflowServiceStubs; @@ -49,7 +56,8 @@ /** * A plugin that implements {@link WorkflowServiceStubsPlugin}, {@link WorkflowClientPlugin}, {@link - * ScheduleClientPlugin}, and {@link WorkerPlugin}. This class can be used in two ways: + * ScheduleClientPlugin}, {@link ActivityClientPlugin}, {@link NexusClientPlugin}, and {@link + * WorkerPlugin}. This class can be used in two ways: * *

    *
  1. Builder pattern: Use {@link #newBuilder(String)} to declaratively configure a plugin @@ -104,6 +112,8 @@ * @see WorkflowServiceStubsPlugin * @see WorkflowClientPlugin * @see ScheduleClientPlugin + * @see ActivityClientPlugin + * @see NexusClientPlugin * @see WorkerPlugin */ @Experimental @@ -111,6 +121,8 @@ public abstract class SimplePlugin implements WorkflowServiceStubsPlugin, WorkflowClientPlugin, ScheduleClientPlugin, + ActivityClientPlugin, + NexusClientPlugin, WorkerPlugin { private final String name; @@ -122,6 +134,9 @@ public abstract class SimplePlugin private final List> replayExecutionCallbacks; private final List workerInterceptors; private final List clientInterceptors; + private final List scheduleClientInterceptors; + private final List activityClientInterceptors; + private final List nexusClientInterceptors; private final List contextPropagators; private final UnaryOperator dataConverterCustomizer; private final List> workflowImplementationTypes; @@ -146,6 +161,9 @@ protected SimplePlugin(@Nonnull String name) { this.replayExecutionCallbacks = Collections.emptyList(); this.workerInterceptors = Collections.emptyList(); this.clientInterceptors = Collections.emptyList(); + this.scheduleClientInterceptors = Collections.emptyList(); + this.activityClientInterceptors = Collections.emptyList(); + this.nexusClientInterceptors = Collections.emptyList(); this.contextPropagators = Collections.emptyList(); this.dataConverterCustomizer = null; this.workflowImplementationTypes = Collections.emptyList(); @@ -171,6 +189,9 @@ protected SimplePlugin(@Nonnull Builder builder) { this.replayExecutionCallbacks = new ArrayList<>(builder.replayExecutionCallbacks); this.workerInterceptors = new ArrayList<>(builder.workerInterceptors); this.clientInterceptors = new ArrayList<>(builder.clientInterceptors); + this.scheduleClientInterceptors = new ArrayList<>(builder.scheduleClientInterceptors); + this.activityClientInterceptors = new ArrayList<>(builder.activityClientInterceptors); + this.nexusClientInterceptors = new ArrayList<>(builder.nexusClientInterceptors); this.contextPropagators = new ArrayList<>(builder.contextPropagators); this.dataConverterCustomizer = builder.dataConverterCustomizer; this.workflowImplementationTypes = new ArrayList<>(builder.workflowImplementationTypes); @@ -229,7 +250,59 @@ public void configureWorkflowClient(@Nonnull WorkflowClientOptions.Builder build @Override public void configureScheduleClient(@Nonnull ScheduleClientOptions.Builder builder) { - // Subclasses can override this method for custom configuration + // Apply data converter customizer. + if (dataConverterCustomizer != null) { + DataConverter existing = builder.build().getDataConverter(); + builder.setDataConverter(dataConverterCustomizer.apply(existing)); + } + + // Add context propagators. + if (!contextPropagators.isEmpty()) { + List combined = new ArrayList<>(builder.build().getContextPropagators()); + combined.addAll(contextPropagators); + builder.setContextPropagators(combined); + } + + // Add client interceptors. + if (!scheduleClientInterceptors.isEmpty()) { + List combined = new ArrayList<>(builder.build().getInterceptors()); + combined.addAll(scheduleClientInterceptors); + builder.setInterceptors(combined); + } + } + + @Override + public void configureActivityClient(@Nonnull ActivityClientOptions.Builder builder) { + // Add context propagators. + if (!contextPropagators.isEmpty()) { + List existing = builder.build().getContextPropagators(); + List combined = new ArrayList<>(existing); + combined.addAll(contextPropagators); + builder.setContextPropagators(combined); + } + + // Add client interceptors. + if (!activityClientInterceptors.isEmpty()) { + List combined = new ArrayList<>(builder.build().getInterceptors()); + combined.addAll(activityClientInterceptors); + builder.setInterceptors(combined); + } + } + + @Override + public void configureNexusClient(@Nonnull NexusClientOptions.Builder builder) { + // Apply data converter customizer. + if (dataConverterCustomizer != null) { + DataConverter existing = builder.build().getDataConverter(); + builder.setDataConverter(dataConverterCustomizer.apply(existing)); + } + + // Add client interceptors. + if (!nexusClientInterceptors.isEmpty()) { + List combined = new ArrayList<>(builder.build().getInterceptors()); + combined.addAll(nexusClientInterceptors); + builder.setInterceptors(combined); + } } @Override @@ -343,6 +416,9 @@ public static final class Builder { new ArrayList<>(); private final List workerInterceptors = new ArrayList<>(); private final List clientInterceptors = new ArrayList<>(); + private final List scheduleClientInterceptors = new ArrayList<>(); + private final List activityClientInterceptors = new ArrayList<>(); + private final List nexusClientInterceptors = new ArrayList<>(); private final List contextPropagators = new ArrayList<>(); private UnaryOperator dataConverterCustomizer; private final List> workflowImplementationTypes = new ArrayList<>(); @@ -525,6 +601,42 @@ public Builder addClientInterceptors(WorkflowClientInterceptor... interceptors) return this; } + /** + * Adds schedule client interceptors. Interceptors are appended to any existing interceptors in + * the configuration. + * + * @param interceptors the interceptors to add + * @return this builder for chaining + */ + public Builder addScheduleClientInterceptors(ScheduleClientInterceptor... interceptors) { + scheduleClientInterceptors.addAll(Arrays.asList(interceptors)); + return this; + } + + /** + * Adds activity client interceptors. Interceptors are appended to any existing interceptors in + * the configuration. + * + * @param interceptors the interceptors to add + * @return this builder for chaining + */ + public Builder addActivityClientInterceptors(ActivityClientInterceptor... interceptors) { + activityClientInterceptors.addAll(Arrays.asList(interceptors)); + return this; + } + + /** + * Adds Nexus client interceptors. Interceptors are appended to any existing interceptors in the + * configuration. + * + * @param interceptors the interceptors to add + * @return this builder for chaining + */ + public Builder addNexusClientInterceptors(NexusClientInterceptor... interceptors) { + nexusClientInterceptors.addAll(Arrays.asList(interceptors)); + return this; + } + /** * Adds context propagators. Propagators are appended to any existing propagators in the * configuration. diff --git a/temporal-sdk/src/main/java/io/temporal/common/interceptors/NexusClientCallsInterceptor.java b/temporal-sdk/src/main/java/io/temporal/common/interceptors/NexusClientCallsInterceptor.java index 5f0bd59980..dad2c69cef 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/interceptors/NexusClientCallsInterceptor.java +++ b/temporal-sdk/src/main/java/io/temporal/common/interceptors/NexusClientCallsInterceptor.java @@ -12,9 +12,9 @@ import io.temporal.common.Experimental; import io.temporal.payload.context.NexusSerializationContext; import java.lang.reflect.Type; -import java.util.Collections; import java.util.Map; import java.util.Optional; +import java.util.TreeMap; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeoutException; import java.util.stream.Stream; @@ -159,7 +159,12 @@ public StartNexusOperationExecutionInput( this.operation = operation; this.input = input; this.options = options; - this.headers = headers == null ? Collections.emptyMap() : headers; + // Interceptors add propagation headers in place, so use a mutable, case-insensitive copy + // without modifying the caller's map. + this.headers = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + if (headers != null) { + this.headers.putAll(headers); + } } public String getEndpoint() { diff --git a/temporal-sdk/src/main/java/io/temporal/internal/common/PluginUtils.java b/temporal-sdk/src/main/java/io/temporal/internal/common/PluginUtils.java index 3b8e92770a..6a2b3b616e 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/common/PluginUtils.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/common/PluginUtils.java @@ -21,10 +21,13 @@ package io.temporal.internal.common; import java.lang.reflect.Array; +import java.util.ArrayList; import java.util.Arrays; import java.util.HashSet; +import java.util.List; import java.util.Set; import java.util.function.Function; +import java.util.function.IntFunction; import javax.annotation.Nullable; import org.slf4j.Logger; @@ -33,6 +36,28 @@ public final class PluginUtils { private PluginUtils() {} + /** + * Returns the entries in {@code plugins} that implement {@code pluginType}, preserving order. + * + * @param plugins plugins to filter (may be null or empty) + * @param pluginType type of plugin to retain + * @param arrayFactory creates an array of the retained plugin type + * @param the retained plugin type + * @return the matching plugins, never null + */ + public static T[] extractPlugins( + @Nullable Object[] plugins, Class pluginType, IntFunction arrayFactory) { + List extracted = new ArrayList<>(); + if (plugins != null) { + for (Object plugin : plugins) { + if (pluginType.isInstance(plugin)) { + extracted.add(pluginType.cast(plugin)); + } + } + } + return extracted.toArray(arrayFactory.apply(extracted.size())); + } + /** * Merges propagated plugins with explicitly specified plugins. Propagated plugins come first, * followed by explicit plugins. Warns about duplicate plugin instances. diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowRunTaskHandler.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowRunTaskHandler.java index 3b0894d5d9..439318bf93 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowRunTaskHandler.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowRunTaskHandler.java @@ -338,6 +338,12 @@ private Map executeQueries(Map streams = new HashMap<>(); long deriveSeed(@Nonnull String runId, @Nonnull String name) { - // The separators keep ("ab", "c") from colliding with ("a", "bc") + // The separators keep ("ab", "c") from colliding with ("a", "bc"). String seed = String.join("\0", SEED_VERSION, runId, name); HashCode hash = Hashing.sha256().hashString(seed, StandardCharsets.UTF_8); return ByteBuffer.wrap(hash.asBytes()).getLong(); diff --git a/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java b/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java index e7b83034d5..f10dbad741 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java @@ -97,7 +97,9 @@ private WorkerFactory(WorkflowClient workflowClient, WorkerFactoryOptions factor String namespace = workflowClientOptions.getNamespace(); // Extract worker plugins from client (auto-propagation) - WorkerPlugin[] propagatedPlugins = extractWorkerPlugins(workflowClientOptions.getPlugins()); + WorkerPlugin[] propagatedPlugins = + PluginUtils.extractPlugins( + workflowClientOptions.getPlugins(), WorkerPlugin.class, WorkerPlugin[]::new); // Get plugins explicitly set on factory options WorkerPlugin[] explicitPlugins = factoryOptions != null ? factoryOptions.getPlugins() : null; @@ -571,24 +573,6 @@ public String toString() { return String.format("WorkerFactory{identity=%s}", workflowClient.getOptions().getIdentity()); } - /** - * Extracts worker plugins from the workflow client plugins array. Only plugins that also - * implement {@link WorkerPlugin} are included. - */ - private static WorkerPlugin[] extractWorkerPlugins( - io.temporal.client.WorkflowClientPlugin[] clientPlugins) { - if (clientPlugins == null || clientPlugins.length == 0) { - return new WorkerPlugin[0]; - } - List workerPlugins = new ArrayList<>(); - for (io.temporal.client.WorkflowClientPlugin plugin : clientPlugins) { - if (plugin instanceof WorkerPlugin) { - workerPlugins.add((WorkerPlugin) plugin); - } - } - return workerPlugins.toArray(new WorkerPlugin[0]); - } - /** * Applies plugin configuration to worker options. Plugins are called in forward (registration) * order. diff --git a/temporal-sdk/src/test/java/io/temporal/client/ActivityClientOptionsTest.java b/temporal-sdk/src/test/java/io/temporal/client/ActivityClientOptionsTest.java index 7239973d74..ac7f8aa86f 100644 --- a/temporal-sdk/src/test/java/io/temporal/client/ActivityClientOptionsTest.java +++ b/temporal-sdk/src/test/java/io/temporal/client/ActivityClientOptionsTest.java @@ -21,6 +21,7 @@ public void testDefaultIdentityIsNotNull() { @Test public void testToBuilderCopiesAllFields() { ActivityClientInterceptor interceptor = mock(ActivityClientInterceptor.class); + ActivityClientPlugin plugin = mock(ActivityClientPlugin.class); ContextPropagator propagator = mock(ContextPropagator.class); DataConverter dc = mock(DataConverter.class); @@ -31,6 +32,7 @@ public void testToBuilderCopiesAllFields() { .setDataConverter(dc) .setInterceptors(Collections.singletonList(interceptor)) .setContextPropagators(Collections.singletonList(propagator)) + .setPlugins(plugin) .build(); ActivityClientOptions copy = original.toBuilder().build(); @@ -40,6 +42,7 @@ public void testToBuilderCopiesAllFields() { assertSame(original.getDataConverter(), copy.getDataConverter()); assertEquals(original.getInterceptors(), copy.getInterceptors()); assertEquals(original.getContextPropagators(), copy.getContextPropagators()); + assertArrayEquals(new ActivityClientPlugin[] {plugin}, copy.getPlugins()); } @Test diff --git a/temporal-sdk/src/test/java/io/temporal/client/NexusClientOptionsTest.java b/temporal-sdk/src/test/java/io/temporal/client/NexusClientOptionsTest.java index 27f1ee5be8..78da01d22b 100644 --- a/temporal-sdk/src/test/java/io/temporal/client/NexusClientOptionsTest.java +++ b/temporal-sdk/src/test/java/io/temporal/client/NexusClientOptionsTest.java @@ -35,6 +35,7 @@ public void testDefaultIdentityIsNotNull() { @Test public void testNewBuilderFromOptionsCopiesAllFields() { NexusClientInterceptor interceptor = mock(NexusClientInterceptor.class); + NexusClientPlugin plugin = mock(NexusClientPlugin.class); DataConverter dc = mock(DataConverter.class); NexusClientOptions original = @@ -44,6 +45,7 @@ public void testNewBuilderFromOptionsCopiesAllFields() { .setDataConverter(dc) .setExternalStorage(storage()) .setInterceptors(Collections.singletonList(interceptor)) + .setPlugins(plugin) .build(); NexusClientOptions copy = NexusClientOptions.newBuilder(original).build(); @@ -53,6 +55,7 @@ public void testNewBuilderFromOptionsCopiesAllFields() { assertSame(original.getDataConverter(), copy.getDataConverter()); assertEquals(original.getInterceptors(), copy.getInterceptors()); assertSame(original.getExternalStorage(), copy.getExternalStorage()); + assertArrayEquals(new NexusClientPlugin[] {plugin}, copy.getPlugins()); } @Test diff --git a/temporal-sdk/src/test/java/io/temporal/common/PluginPropagationTest.java b/temporal-sdk/src/test/java/io/temporal/common/PluginPropagationTest.java index 5b3c2a7118..f21c90248d 100644 --- a/temporal-sdk/src/test/java/io/temporal/common/PluginPropagationTest.java +++ b/temporal-sdk/src/test/java/io/temporal/common/PluginPropagationTest.java @@ -22,7 +22,12 @@ import static org.junit.Assert.*; +import io.temporal.client.ActivityClient; +import io.temporal.client.NexusClient; import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.schedules.ScheduleClient; +import io.temporal.client.schedules.ScheduleClientOptions; +import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.serviceclient.WorkflowServiceStubsOptions; import io.temporal.testing.CloudTestExclusion.RequiresLocalServer; import io.temporal.testing.CloudTestExclusionNote; @@ -334,4 +339,47 @@ public void testMergedPluginsAtWorkerFactoryLevel() { env.close(); } } + + @Test + public void testPluginPropagatesFromServiceStubsToStandaloneClients() { + List installed = new ArrayList<>(); + SimplePlugin plugin = + SimplePlugin.newBuilder("client-tracking") + .addScheduleClientInterceptors( + next -> { + installed.add("schedule"); + return next; + }) + .addActivityClientInterceptors( + next -> { + installed.add("activity"); + return next; + }) + .addNexusClientInterceptors( + next -> { + installed.add("nexus"); + return next; + }) + .build(); + + WorkflowServiceStubsOptions stubsOptions = + WorkflowServiceStubsOptions.newBuilder() + .setPlugins((io.temporal.serviceclient.WorkflowServiceStubsPlugin) plugin) + .build(); + TestWorkflowEnvironment env = + TestWorkflowEnvironment.newInstance( + TestEnvironmentOptions.newBuilder() + .setWorkflowServiceStubsOptions(stubsOptions) + .build()); + try { + WorkflowServiceStubs stubs = env.getWorkflowServiceStubs(); + ScheduleClient.newInstance(stubs, ScheduleClientOptions.newBuilder().build()); + ActivityClient.newInstance(stubs); + NexusClient.newInstance(stubs); + + assertEquals(Arrays.asList("schedule", "activity", "nexus"), installed); + } finally { + env.close(); + } + } } diff --git a/temporal-sdk/src/test/java/io/temporal/common/SimplePluginBuilderTest.java b/temporal-sdk/src/test/java/io/temporal/common/SimplePluginBuilderTest.java index 380f60daba..1d7be89ba8 100644 --- a/temporal-sdk/src/test/java/io/temporal/common/SimplePluginBuilderTest.java +++ b/temporal-sdk/src/test/java/io/temporal/common/SimplePluginBuilderTest.java @@ -23,8 +23,21 @@ import static org.junit.Assert.*; import static org.mockito.Mockito.*; +import io.temporal.client.ActivityClientOptions; +import io.temporal.client.ActivityClientPlugin; +import io.temporal.client.NexusClientOptions; +import io.temporal.client.NexusClientPlugin; import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.schedules.ScheduleClientOptions; +import io.temporal.client.schedules.ScheduleClientPlugin; +import io.temporal.common.context.ContextPropagator; import io.temporal.common.converter.DataConverter; +import io.temporal.common.interceptors.ActivityClientInterceptor; +import io.temporal.common.interceptors.ActivityClientInterceptorBase; +import io.temporal.common.interceptors.NexusClientInterceptor; +import io.temporal.common.interceptors.NexusClientInterceptorBase; +import io.temporal.common.interceptors.ScheduleClientInterceptor; +import io.temporal.common.interceptors.ScheduleClientInterceptorBase; import io.temporal.common.interceptors.WorkerInterceptor; import io.temporal.common.interceptors.WorkerInterceptorBase; import io.temporal.common.interceptors.WorkflowClientInterceptor; @@ -32,6 +45,7 @@ import io.temporal.worker.Worker; import io.temporal.worker.WorkerFactoryOptions; import io.temporal.worker.WorkerPlugin; +import java.util.Collections; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -55,6 +69,9 @@ public void testSimplePluginImplementsAllInterfaces() { "Should implement WorkflowClientPlugin", plugin instanceof io.temporal.client.WorkflowClientPlugin); assertTrue("Should implement WorkerPlugin", plugin instanceof io.temporal.worker.WorkerPlugin); + assertTrue("Should implement ScheduleClientPlugin", plugin instanceof ScheduleClientPlugin); + assertTrue("Should implement ActivityClientPlugin", plugin instanceof ActivityClientPlugin); + assertTrue("Should implement NexusClientPlugin", plugin instanceof NexusClientPlugin); } @Test @@ -72,6 +89,30 @@ public void testAddWorkerInterceptors() { assertSame(interceptor, interceptors[0]); } + @Test + public void testAddContextPropagatorsToActivityClient() { + ContextPropagator propagator = mock(ContextPropagator.class); + + SimplePlugin plugin = SimplePlugin.newBuilder("test").addContextPropagators(propagator).build(); + + ActivityClientOptions.Builder builder = ActivityClientOptions.newBuilder(); + ((ActivityClientPlugin) plugin).configureActivityClient(builder); + + assertEquals(Collections.singletonList(propagator), builder.build().getContextPropagators()); + } + + @Test + public void testAddContextPropagatorsToScheduleClient() { + ContextPropagator propagator = mock(ContextPropagator.class); + + SimplePlugin plugin = SimplePlugin.newBuilder("test").addContextPropagators(propagator).build(); + + ScheduleClientOptions.Builder builder = ScheduleClientOptions.newBuilder(); + ((ScheduleClientPlugin) plugin).configureScheduleClient(builder); + + assertEquals(Collections.singletonList(propagator), builder.build().getContextPropagators()); + } + @Test public void testAddClientInterceptors() { WorkflowClientInterceptor interceptor = new WorkflowClientInterceptorBase() {}; @@ -87,6 +128,45 @@ public void testAddClientInterceptors() { assertSame(interceptor, interceptors[0]); } + @Test + public void testAddScheduleClientInterceptors() { + ScheduleClientInterceptor interceptor = new ScheduleClientInterceptorBase() {}; + + SimplePlugin plugin = + SimplePlugin.newBuilder("test").addScheduleClientInterceptors(interceptor).build(); + + ScheduleClientOptions.Builder builder = ScheduleClientOptions.newBuilder(); + ((ScheduleClientPlugin) plugin).configureScheduleClient(builder); + + assertEquals(Collections.singletonList(interceptor), builder.build().getInterceptors()); + } + + @Test + public void testAddActivityClientInterceptors() { + ActivityClientInterceptor interceptor = new ActivityClientInterceptorBase() {}; + + SimplePlugin plugin = + SimplePlugin.newBuilder("test").addActivityClientInterceptors(interceptor).build(); + + ActivityClientOptions.Builder builder = ActivityClientOptions.newBuilder(); + ((ActivityClientPlugin) plugin).configureActivityClient(builder); + + assertEquals(Collections.singletonList(interceptor), builder.build().getInterceptors()); + } + + @Test + public void testAddNexusClientInterceptors() { + NexusClientInterceptor interceptor = new NexusClientInterceptorBase() {}; + + SimplePlugin plugin = + SimplePlugin.newBuilder("test").addNexusClientInterceptors(interceptor).build(); + + NexusClientOptions.Builder builder = NexusClientOptions.newBuilder(); + ((NexusClientPlugin) plugin).configureNexusClient(builder); + + assertEquals(Collections.singletonList(interceptor), builder.build().getInterceptors()); + } + @Test public void testInterceptorsAppendToExisting() { WorkerInterceptor existingInterceptor = new WorkerInterceptorBase() {}; @@ -350,6 +430,52 @@ public void testCustomizeDataConverterReceivesExisting() { assertSame(newConverter, builder.build().getDataConverter()); } + @Test + public void testCustomizeDataConverterAppliesToScheduleClient() { + DataConverter existingConverter = mock(DataConverter.class); + DataConverter newConverter = mock(DataConverter.class); + AtomicReference capturedExisting = new AtomicReference<>(); + + SimplePlugin plugin = + SimplePlugin.newBuilder("test") + .customizeDataConverter( + existing -> { + capturedExisting.set(existing); + return newConverter; + }) + .build(); + + ScheduleClientOptions.Builder builder = + ScheduleClientOptions.newBuilder().setDataConverter(existingConverter); + ((ScheduleClientPlugin) plugin).configureScheduleClient(builder); + + assertSame(existingConverter, capturedExisting.get()); + assertSame(newConverter, builder.build().getDataConverter()); + } + + @Test + public void testCustomizeDataConverterAppliesToNexusClient() { + DataConverter existingConverter = mock(DataConverter.class); + DataConverter newConverter = mock(DataConverter.class); + AtomicReference capturedExisting = new AtomicReference<>(); + + SimplePlugin plugin = + SimplePlugin.newBuilder("test") + .customizeDataConverter( + existing -> { + capturedExisting.set(existing); + return newConverter; + }) + .build(); + + NexusClientOptions.Builder builder = + NexusClientOptions.newBuilder().setDataConverter(existingConverter); + ((NexusClientPlugin) plugin).configureNexusClient(builder); + + assertSame(existingConverter, capturedExisting.get()); + assertSame(newConverter, builder.build().getDataConverter()); + } + @Test(expected = NullPointerException.class) public void testNullDataConverterCustomizer() { SimplePlugin.newBuilder("test").customizeDataConverter(null); diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/EvictionIsReplayingTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/EvictionIsReplayingTest.java new file mode 100644 index 0000000000..3101cdcca3 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/EvictionIsReplayingTest.java @@ -0,0 +1,108 @@ +package io.temporal.workflow; + +import static org.junit.Assert.assertEquals; + +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowStub; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.unsafe.WorkflowUnsafe; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import org.junit.Rule; +import org.junit.Test; + +public class EvictionIsReplayingTest { + private static final List replayingInFinally = new CopyOnWriteArrayList<>(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(TestWorkflowImpl.class, BlockedHandlerWorkflowImpl.class) + .build(); + + @Test + public void evictionUnwindsWorkflowCodeAsReplay() { + replayingInFinally.clear(); + TestWorkflow workflow = testWorkflowRule.newWorkflowStub(TestWorkflow.class); + WorkflowExecution execution = WorkflowClient.start(workflow::execute); + testWorkflowRule.waitForTheEndOfWFT(execution.getWorkflowId()); + testWorkflowRule.invalidateWorkflowCache(); + + workflow.finish(); + WorkflowStub.fromTyped(workflow).getResult(Void.class); + + // The eviction unwinds the finally block as replay; the completion runs it live. + assertEquals(Arrays.asList(true, false), replayingInFinally); + } + + @Test + public void completionUnwindsBlockedHandlersLive() { + replayingInFinally.clear(); + BlockedHandlerWorkflow workflow = + testWorkflowRule.newWorkflowStub(BlockedHandlerWorkflow.class); + WorkflowClient.start(workflow::execute); + + workflow.block(); + WorkflowStub.fromTyped(workflow).getResult(Void.class); + + assertEquals(Collections.singletonList(false), replayingInFinally); + } + + @WorkflowInterface + public interface TestWorkflow { + @WorkflowMethod + void execute(); + + @SignalMethod + void finish(); + } + + public static class TestWorkflowImpl implements TestWorkflow { + private boolean finished; + + @Override + public void execute() { + try { + Workflow.await(() -> finished); + } finally { + replayingInFinally.add(WorkflowUnsafe.isReplaying()); + } + } + + @Override + public void finish() { + finished = true; + } + } + + @WorkflowInterface + public interface BlockedHandlerWorkflow { + @WorkflowMethod + void execute(); + + @SignalMethod + void block(); + } + + public static class BlockedHandlerWorkflowImpl implements BlockedHandlerWorkflow { + private boolean blocked; + + @Override + public void execute() { + Workflow.await(() -> blocked); + } + + @Override + public void block() { + blocked = true; + try { + Workflow.await(() -> false); + } finally { + replayingInFinally.add(WorkflowUnsafe.isReplaying()); + } + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowRandomStreamTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowRandomStreamTest.java index 549d6e242c..51ba7a69c5 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowRandomStreamTest.java +++ b/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowRandomStreamTest.java @@ -105,7 +105,7 @@ private void assertResetValues(Class workflowType) { assertNotEquals(original[0], original[1]); // The reset targets the second Workflow Task (id=10), so the first draw is replayed and the - // second draw is redrawn + // second draw is redrawn. ResetWorkflowExecutionResponse response = client .getWorkflowServiceStubs()