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..9e97985c1a 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 ActivityClientPlugins from service stubs plugins (propagation) + 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..acdd8fedf9 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 NexusClientPlugins from service stubs plugins (propagation) + 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..11b357e3bf 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,38 @@ public void configureWorkflowClient(@Nonnull WorkflowClientOptions.Builder build @Override public void configureScheduleClient(@Nonnull ScheduleClientOptions.Builder builder) { - // Subclasses can override this method for custom configuration + 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) { + if (!nexusClientInterceptors.isEmpty()) { + List combined = new ArrayList<>(builder.build().getInterceptors()); + combined.addAll(nexusClientInterceptors); + builder.setInterceptors(combined); + } } @Override @@ -242,6 +294,15 @@ public void configureWorkerFactory(@Nonnull WorkerFactoryOptions.Builder builder combined.addAll(workerInterceptors); builder.setWorkerInterceptors(combined.toArray(new WorkerInterceptor[0])); } + + // Add context propagators + if (!contextPropagators.isEmpty()) { + List existing = builder.build().getContextPropagators(); + List combined = + new ArrayList<>(existing != null ? existing : new ArrayList<>()); + combined.addAll(contextPropagators); + builder.setContextPropagators(combined); + } } @Override @@ -343,6 +404,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 +589,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/sync/CompletablePromiseImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/CompletablePromiseImpl.java index d4ea14dddf..04f03e0634 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/CompletablePromiseImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/CompletablePromiseImpl.java @@ -1,6 +1,7 @@ package io.temporal.internal.sync; import io.temporal.failure.TemporalFailure; +import io.temporal.internal.context.ContextThreadLocal; import io.temporal.workflow.CancellationScope; import io.temporal.workflow.CompletablePromise; import io.temporal.workflow.Functions; @@ -9,6 +10,7 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -246,7 +248,19 @@ private Promise then(Functions.Proc1> proc) { proc.apply(resultPromise); unregisterWithRunner(); } else { - handlers.add(() -> proc.apply(resultPromise)); + // Handlers run in a callback thread created by the runner, which inherits its propagated + // contexts from the workflow header rather than from the thread registering the handler. + Map contexts = ContextThreadLocal.getCurrentContextForPropagation(); + handlers.add( + () -> { + Map previous = ContextThreadLocal.getCurrentContextForPropagation(); + ContextThreadLocal.propagateContextToCurrentThread(contexts); + try { + proc.apply(resultPromise); + } finally { + ContextThreadLocal.propagateContextToCurrentThread(previous); + } + }); } return resultPromise; } 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 dfb9b51326..efddefa082 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java @@ -10,6 +10,7 @@ import io.temporal.api.workflowservice.v1.DescribeNamespaceResponse; import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.context.ContextPropagator; import io.temporal.common.converter.DataConverter; import io.temporal.internal.client.WorkflowClientInternal; import io.temporal.internal.common.PluginUtils; @@ -27,6 +28,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -96,7 +98,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; @@ -190,6 +194,15 @@ public synchronized Worker newWorker(String taskQueue, WorkerOptions options) { // Apply plugin configuration to worker options (forward order) options = applyWorkerPluginConfiguration(taskQueue, options, this.plugins); + Map contextPropagatorsByName = new LinkedHashMap<>(); + for (ContextPropagator propagator : workflowClient.getOptions().getContextPropagators()) { + contextPropagatorsByName.putIfAbsent(propagator.getName(), propagator); + } + for (ContextPropagator propagator : factoryOptions.getContextPropagators()) { + contextPropagatorsByName.putIfAbsent(propagator.getName(), propagator); + } + List contextPropagators = new ArrayList<>(contextPropagatorsByName.values()); + // Only one worker can exist for a task queue Worker existingWorker = workers.get(taskQueue); if (existingWorker == null) { @@ -204,7 +217,7 @@ public synchronized Worker newWorker(String taskQueue, WorkerOptions options) { cache, true, workflowThreadExecutor, - workflowClient.getOptions().getContextPropagators(), + contextPropagators, plugins, ((WorkflowClientInternal) workflowClient.getInternal()).getWorkerGroupingKey(), namespaceCapabilities); @@ -569,24 +582,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/main/java/io/temporal/worker/WorkerFactoryOptions.java b/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactoryOptions.java index dae83334de..ec1734346e 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactoryOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/WorkerFactoryOptions.java @@ -3,8 +3,11 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import io.temporal.common.Experimental; +import io.temporal.common.context.ContextPropagator; import io.temporal.common.interceptors.WorkerInterceptor; import java.time.Duration; +import java.util.Collections; +import java.util.List; import java.util.concurrent.ExecutorService; import javax.annotation.Nullable; @@ -38,6 +41,7 @@ public static class Builder { private int workflowCacheSize; private int maxWorkflowThreadCount; private WorkerInterceptor[] workerInterceptors; + private List contextPropagators; private WorkerPlugin[] plugins; private boolean enableLoggingInReplay; private boolean usingVirtualWorkflowThreads; @@ -55,6 +59,7 @@ private Builder(WorkerFactoryOptions options) { this.workflowCacheSize = options.workflowCacheSize; this.maxWorkflowThreadCount = options.maxWorkflowThreadCount; this.workerInterceptors = options.workerInterceptors; + this.contextPropagators = options.contextPropagators; this.plugins = options.plugins; this.enableLoggingInReplay = options.enableLoggingInReplay; this.usingVirtualWorkflowThreads = options.usingVirtualWorkflowThreads; @@ -106,6 +111,15 @@ public Builder setWorkerInterceptors(WorkerInterceptor... workerInterceptors) { return this; } + /** + * Sets the context propagators to use with workers created by this factory. These are appended + * to the propagators configured on the workflow client. + */ + public Builder setContextPropagators(List contextPropagators) { + this.contextPropagators = contextPropagators; + return this; + } + /** * Sets the worker plugins to use with workers created by this factory. Plugins can modify * worker configuration and wrap worker lifecycle. @@ -179,6 +193,7 @@ public WorkerFactoryOptions build() { maxWorkflowThreadCount, workflowHostLocalTaskQueueScheduleToStartTimeout, workerInterceptors, + contextPropagators == null ? Collections.emptyList() : contextPropagators, plugins, enableLoggingInReplay, usingVirtualWorkflowThreads, @@ -204,6 +219,7 @@ public WorkerFactoryOptions validateAndBuildWithDefaults() { maxWorkflowThreadCount, workflowHostLocalTaskQueueScheduleToStartTimeout, workerInterceptors == null ? new WorkerInterceptor[0] : workerInterceptors, + contextPropagators == null ? Collections.emptyList() : contextPropagators, plugins == null ? new WorkerPlugin[0] : plugins, enableLoggingInReplay, usingVirtualWorkflowThreads, @@ -217,6 +233,7 @@ public WorkerFactoryOptions validateAndBuildWithDefaults() { private final int maxWorkflowThreadCount; private final @Nullable Duration workflowHostLocalTaskQueueScheduleToStartTimeout; private final WorkerInterceptor[] workerInterceptors; + private final List contextPropagators; private final WorkerPlugin[] plugins; private final boolean enableLoggingInReplay; private final boolean usingVirtualWorkflowThreads; @@ -228,6 +245,7 @@ private WorkerFactoryOptions( int maxWorkflowThreadCount, @Nullable Duration workflowHostLocalTaskQueueScheduleToStartTimeout, WorkerInterceptor[] workerInterceptors, + List contextPropagators, WorkerPlugin[] plugins, boolean enableLoggingInReplay, boolean usingVirtualWorkflowThreads, @@ -268,6 +286,7 @@ private WorkerFactoryOptions( this.workflowHostLocalTaskQueueScheduleToStartTimeout = workflowHostLocalTaskQueueScheduleToStartTimeout; this.workerInterceptors = workerInterceptors; + this.contextPropagators = contextPropagators; this.plugins = plugins; this.enableLoggingInReplay = enableLoggingInReplay; this.usingVirtualWorkflowThreads = usingVirtualWorkflowThreads; @@ -292,6 +311,15 @@ public WorkerInterceptor[] getWorkerInterceptors() { return workerInterceptors; } + /** + * Returns the context propagators used with workers created by this factory. + * + * @return the list of context propagators, never null + */ + public List getContextPropagators() { + return contextPropagators; + } + /** * Returns the worker plugins configured for this factory. * 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 d91922884a..2ce2d67df6 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.TestEnvironmentOptions; import io.temporal.testing.TestWorkflowEnvironment; @@ -329,4 +334,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..cc3bb6f67b 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 testAddContextPropagatorsToWorkerFactory() { + ContextPropagator propagator = mock(ContextPropagator.class); + + SimplePlugin plugin = SimplePlugin.newBuilder("test").addContextPropagators(propagator).build(); + + WorkerFactoryOptions.Builder builder = WorkerFactoryOptions.newBuilder(); + ((io.temporal.worker.WorkerPlugin) plugin).configureWorkerFactory(builder); + + assertEquals(Collections.singletonList(propagator), builder.build().getContextPropagators()); + } + + @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 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() {}; diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerFactoryOptionsTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerFactoryOptionsTest.java index c14a53af49..c6f45a5d68 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerFactoryOptionsTest.java +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerFactoryOptionsTest.java @@ -3,6 +3,7 @@ import static org.junit.Assert.assertEquals; import java.time.Duration; +import java.util.Collections; import org.junit.Test; public class WorkerFactoryOptionsTest { @@ -13,6 +14,12 @@ public void shutdownCheckIntervalDefaultIs250ms() { assertEquals(Duration.ofMillis(250), options.getShutdownCheckInterval()); } + @Test + public void contextPropagatorsDefaultToEmptyList() { + assertEquals( + Collections.emptyList(), WorkerFactoryOptions.newBuilder().build().getContextPropagators()); + } + @Test public void shutdownCheckIntervalCanBeSet() { Duration interval = Duration.ofMillis(5); diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/ContextPropagationTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/ContextPropagationTest.java index 4de7876bc9..61868333ef 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/ContextPropagationTest.java +++ b/temporal-sdk/src/test/java/io/temporal/workflow/ContextPropagationTest.java @@ -99,6 +99,20 @@ public void testThreadContextPropagation() { assertEquals("asynctesting123", result); } + @Test + public void testPromiseCallbackContextPropagation() { + Worker worker = testEnvironment.newWorker(TASK_QUEUE); + worker.registerWorkflowImplementationTypes(ContextPropagationCallbackWorkflowImpl.class); + testEnvironment.start(); + MDC.put("test", "testing123"); + WorkflowClient client = testEnvironment.getWorkflowClient(); + WorkflowOptions options = WorkflowOptions.newBuilder().setTaskQueue(TASK_QUEUE).build(); + TestWorkflows.TestWorkflow1 workflow = + client.newWorkflowStub(TestWorkflows.TestWorkflow1.class, options); + String result = workflow.execute("input1"); + assertEquals("asynccallback123", result); + } + @Test public void testActivityContextPropagation() { Worker worker = testEnvironment.newWorker(TASK_QUEUE); @@ -334,6 +348,25 @@ private String async() { } } + public static class ContextPropagationCallbackWorkflowImpl + implements TestWorkflows.TestWorkflow1 { + + @Override + public String execute(String input) { + Promise asyncPromise = Async.function(this::async); + // The callback must see the context of the thread that registered it, not the context the + // thread completing the promise happens to carry. + MDC.put("test", "callback123"); + Promise callback = asyncPromise.thenApply(result -> result + MDC.get("test")); + MDC.put("test", "testing123"); + return callback.get(); + } + + private String async() { + return "async"; + } + } + public static class ContextActivityImpl implements TestActivities.TestActivity1 { @Override public String execute(String input) { diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/WorkerFactoryContextPropagationTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/WorkerFactoryContextPropagationTest.java new file mode 100644 index 0000000000..a93e159b20 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/WorkerFactoryContextPropagationTest.java @@ -0,0 +1,87 @@ +package io.temporal.workflow; + +import static org.junit.Assert.assertEquals; + +import io.temporal.client.WorkflowClientOptions; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.worker.WorkerFactoryOptions; +import io.temporal.workflow.shared.TestWorkflows; +import java.util.Collections; +import org.junit.Rule; +import org.junit.Test; +import org.junit.experimental.runners.Enclosed; +import org.junit.runner.RunWith; +import org.slf4j.MDC; + +@RunWith(Enclosed.class) +public class WorkerFactoryContextPropagationTest { + + public static class FactoryOnly { + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkerFactoryOptions( + WorkerFactoryOptions.newBuilder() + .setContextPropagators( + Collections.singletonList( + new ContextPropagationTest.TestContextPropagator())) + .build()) + .setWorkflowTypes(FactoryContextPropagationThreadWorkflowImpl.class) + .build(); + + @Test + public void testThreadContextPropagationFromWorkerFactoryOptions() { + TestWorkflows.TestWorkflow1 workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class); + assertEquals("testing123", workflow.execute("testing123")); + } + + public static class FactoryContextPropagationThreadWorkflowImpl + implements TestWorkflows.TestWorkflow1 { + + @Override + public String execute(String input) { + MDC.put("test", input); + return Async.function(() -> MDC.get("test")).get(); + } + } + } + + public static class DuplicateNames { + private final ContextPropagationTest.TestContextPropagator clientPropagator = + new ContextPropagationTest.TestContextPropagator(); + private final ContextPropagationTest.TestContextPropagator factoryPropagator = + new ContextPropagationTest.TestContextPropagator() { + @Override + public String getName() { + return clientPropagator.getName(); + } + + @Override + public Object getCurrentContext() { + throw new AssertionError("Duplicate factory context propagator was used"); + } + }; + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder() + .setContextPropagators(Collections.singletonList(clientPropagator)) + .build()) + .setWorkerFactoryOptions( + WorkerFactoryOptions.newBuilder() + .setContextPropagators(Collections.singletonList(factoryPropagator)) + .build()) + .setWorkflowTypes(FactoryOnly.FactoryContextPropagationThreadWorkflowImpl.class) + .build(); + + @Test + public void testDuplicateContextPropagatorsAreIgnored() { + TestWorkflows.TestWorkflow1 workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class); + assertEquals("testing123", workflow.execute("testing123")); + } + } +}