diff --git a/CHANGELOG.md b/CHANGELOG.md index a390056d46..416b847a6e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,11 @@ to docs, or any other relevant information. `WorkflowOutboundCallsInterceptorBase` need to add them, and Base subclasses override them to see calls with options. ### Added +- Tracing options and `ContextPropagator.encodeHeaders()` can opt into header processing using + the configured client's payload codec. Only opted-in producers' output is encoded, while + application-supplied wire-ready headers are unchanged. Receiving consumers use private decoded + views. Existing propagators default to false; custom converter wrappers can expose their codec + through `PayloadCodecProvider`. - Applications can use the experimental `@TransferTypeConvertible` annotation and `TransferTypeConverter` interface to convert top-level model values to transfer types. The SDK applies this conversion to workflow, activity, Nexus, and schedule payloads, including failure @@ -78,6 +83,7 @@ to docs, or any other relevant information. specific to the Temporal CLI dev server implementation and may no longer be valid if that implementation changes. ### Fixed +- Schedule action headers are now preserved when describing and updating a schedule. - `ActivityClient` now sends the `temporal-namespace` gRPC header like `WorkflowClient` and `ScheduleClient` do, so standalone activity calls are no longer rejected with `PERMISSION_DENIED` by namespace-scoped auth (e.g. Temporal Cloud API keys). diff --git a/README.md b/README.md index 27b6f63288..18108d43f2 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,38 @@ If you cannot use protobuf-java 3.25 >=, you can try `temporal-shaded` which inc We'd love your help in improving the Temporal Java SDK. Please review our [contribution guidelines](CONTRIBUTING.md). +## Automatic payload-header codecs + +User-supplied header payloads are wire-ready, including any codec encoding. +Tracing and context propagators can independently opt into processing their own +headers with the codec from the configured client's data converter: + +```java +DataConverter payloadConverter = DefaultDataConverter.newDefaultInstance(); +DataConverter converter = new CodecDataConverter(payloadConverter, codecs); +WorkflowClientOptions options = + WorkflowClientOptions.newBuilder() + .setDataConverter(converter) + .build(); +``` + +Set `OpenTracingOptions.Builder.setEncodeHeaders(true)` to encode only that integration's +tracing headers, retaining standard tracing serialization. The OpenTelemetry plugin and worker +configuration builders expose the same option. Context propagators opt in by overriding +`ContextPropagator.encodeHeaders()` to return true; the default is false. + +Opted-in propagators must produce serialized **but not already codec-encoded** payloads. +The SDK applies the configured client's codec to that producer's output, not to other headers. +Receiving integrations and propagators must opt in to read encoded headers and use a compatible codec. +They receive private decoded views of the original wire headers; codecs must pass through +encodings they do not recognize. Ordinary interceptors continue to receive wire-ready payloads. +Retained schedule headers remain wire-ready and are not re-encoded when an action is reused. + +`CodecDataConverter` exposes its codec through `PayloadCodecProvider`; custom converter wrappers +must forward that capability. Without a configured codec, processing is an identity operation. +No value conversion is applied to header payloads. Nexus string headers and gRPC metadata are +unaffected. Codec failures are propagated, never retried as plaintext. + ## Development ### Development Environment diff --git a/contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryPlugin.java b/contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryPlugin.java index 3b1e2f1604..555fea03b9 100644 --- a/contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryPlugin.java +++ b/contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryPlugin.java @@ -37,6 +37,7 @@ public final class OpenTelemetryPlugin extends SimplePlugin { private final Runnable flushHook; private final IdGenerator idGenerator; private final boolean flushOnWorkerFactoryShutdown; + private final boolean encodeHeaders; private final List metricsScopes = new ArrayList<>(); private OpenTelemetry openTelemetry; @@ -53,6 +54,7 @@ private OpenTelemetryPlugin(Builder builder) { this.flushHook = builder.flushHook; this.idGenerator = builder.idGenerator; this.flushOnWorkerFactoryShutdown = builder.flushOnWorkerFactoryShutdown; + this.encodeHeaders = builder.encodeHeaders; } public static Builder newBuilder() { @@ -94,7 +96,7 @@ public void configureServiceStubs(@Nonnull WorkflowServiceStubsOptions.Builder b @Override public void configureWorkflowClient(@Nonnull WorkflowClientOptions.Builder builder) { OpenTracingOptions tracingOptions = - OpenTelemetryWorker.createOpenTracingOptions(getOpenTelemetry()); + OpenTelemetryWorker.createOpenTracingOptions(getOpenTelemetry(), encodeHeaders); OpenTelemetryWorker.appendClientInterceptor( builder, new io.temporal.opentracing.OpenTracingClientInterceptor(tracingOptions)); } @@ -102,7 +104,7 @@ public void configureWorkflowClient(@Nonnull WorkflowClientOptions.Builder build @Override public void configureWorkerFactory(@Nonnull WorkerFactoryOptions.Builder builder) { OpenTracingOptions tracingOptions = - OpenTelemetryWorker.createOpenTracingOptions(getOpenTelemetry()); + OpenTelemetryWorker.createOpenTracingOptions(getOpenTelemetry(), encodeHeaders); OpenTelemetryWorker.appendWorkerInterceptor( builder, new io.temporal.opentracing.OpenTracingWorkerInterceptor(tracingOptions)); } @@ -155,6 +157,17 @@ private List drainMetricsScopes() { /** Builder for {@link OpenTelemetryPlugin}. */ public static final class Builder { + private boolean encodeHeaders; + + /** + * Uses the configured client's payload codec for tracing headers. Defaults to false; + * application-supplied headers and standard tracing serialization are unchanged. + */ + public Builder setEncodeHeaders(boolean encodeHeaders) { + this.encodeHeaders = encodeHeaders; + return this; + } + private final Map env; private OpenTelemetry openTelemetry; private String endpoint; diff --git a/contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryWorker.java b/contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryWorker.java index 8d1eabfd15..d2e89d273d 100644 --- a/contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryWorker.java +++ b/contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryWorker.java @@ -141,10 +141,23 @@ public static void configureTracing( @Nonnull WorkflowClientOptions.Builder clientOptions, @Nonnull WorkerFactoryOptions.Builder workerFactoryOptions, @Nonnull OpenTelemetry openTelemetry) { + configureTracing(clientOptions, workerFactoryOptions, openTelemetry, false); + } + + /** + * Configures tracing with optional codec processing using the configured client's payload codec. + * Only tracing headers are encoded; application-supplied headers are unchanged. + */ + public static void configureTracing( + @Nonnull WorkflowClientOptions.Builder clientOptions, + @Nonnull WorkerFactoryOptions.Builder workerFactoryOptions, + @Nonnull OpenTelemetry openTelemetry, + boolean encodeHeaders) { Objects.requireNonNull(clientOptions, "clientOptions"); Objects.requireNonNull(workerFactoryOptions, "workerFactoryOptions"); OpenTracingOptions tracingOptions = - createOpenTracingOptions(Objects.requireNonNull(openTelemetry, "openTelemetry")); + createOpenTracingOptions( + Objects.requireNonNull(openTelemetry, "openTelemetry"), encodeHeaders); appendClientInterceptor(clientOptions, new OpenTracingClientInterceptor(tracingOptions)); appendWorkerInterceptor(workerFactoryOptions, new OpenTracingWorkerInterceptor(tracingOptions)); } @@ -205,6 +218,17 @@ public static String resolveServiceName( } public static final class Builder { + private boolean encodeHeaders; + + /** + * Uses the configured client's payload codec for tracing headers. Defaults to false; + * application-supplied headers and standard tracing serialization are unchanged. + */ + public Builder setEncodeHeaders(boolean encodeHeaders) { + this.encodeHeaders = encodeHeaders; + return this; + } + private final Map env; private OpenTelemetry openTelemetry; private String endpoint; @@ -294,7 +318,7 @@ public void apply( resolvedOpenTelemetry, getServiceName(), metricsReportInterval); - configureTracing(clientOptions, workerFactoryOptions, resolvedOpenTelemetry); + configureTracing(clientOptions, workerFactoryOptions, resolvedOpenTelemetry, encodeHeaders); if (flushHook == null) { configureFlushHook(addShutdownHook, resolvedOpenTelemetry, flushTimeout); } else { @@ -377,8 +401,10 @@ static Scope createMetricsScope( requirePositive(reportInterval, "reportInterval").toMillis())); } - static OpenTracingOptions createOpenTracingOptions(@Nonnull OpenTelemetry openTelemetry) { + static OpenTracingOptions createOpenTracingOptions( + @Nonnull OpenTelemetry openTelemetry, boolean encodeHeaders) { return OpenTracingOptions.newBuilder() + .setEncodeHeaders(encodeHeaders) .setTracer( OpenTracingShim.createTracerShim( Objects.requireNonNull(openTelemetry, "openTelemetry"))) diff --git a/contrib/temporal-opentracing/src/main/java/io/temporal/opentracing/OpenTracingOptions.java b/contrib/temporal-opentracing/src/main/java/io/temporal/opentracing/OpenTracingOptions.java index e1a189c9f0..92f40f60e6 100644 --- a/contrib/temporal-opentracing/src/main/java/io/temporal/opentracing/OpenTracingOptions.java +++ b/contrib/temporal-opentracing/src/main/java/io/temporal/opentracing/OpenTracingOptions.java @@ -17,6 +17,7 @@ public class OpenTracingOptions { private final SpanBuilderProvider spanBuilderProvider; private final OpenTracingSpanContextCodec spanContextCodec; private final Predicate isErrorPredicate; + private final boolean encodeHeaders; public static OpenTracingOptions getDefaultInstance() { return DEFAULT_INSTANCE; @@ -26,12 +27,14 @@ private OpenTracingOptions( Tracer tracer, SpanBuilderProvider spanBuilderProvider, OpenTracingSpanContextCodec spanContextCodec, - Predicate isErrorPredicate) { + Predicate isErrorPredicate, + boolean encodeHeaders) { if (tracer == null) throw new IllegalArgumentException("tracer shouldn't be null"); this.tracer = tracer; this.spanBuilderProvider = spanBuilderProvider; this.spanContextCodec = spanContextCodec; this.isErrorPredicate = isErrorPredicate; + this.encodeHeaders = encodeHeaders; } @Nonnull @@ -54,6 +57,10 @@ public Predicate getIsErrorPredicate() { return isErrorPredicate; } + public boolean isEncodeHeaders() { + return encodeHeaders; + } + public static Builder newBuilder() { return new Builder(); } @@ -64,9 +71,21 @@ public static final class Builder { private OpenTracingSpanContextCodec spanContextCodec = OpenTracingSpanContextCodec.TEXT_MAP_INJECT_EXTRACT_CODEC; private Predicate isErrorPredicate = t -> true; + private boolean encodeHeaders; private Builder() {} + /** + * Uses the configured client's payload codec for this integration's tracing headers. Defaults + * to false. Standard tracing serialization and other producers' headers are unchanged. + * Configure the same setting on receiving workers. Codecs must pass through unrecognized + * encodings. + */ + public Builder setEncodeHeaders(boolean encodeHeaders) { + this.encodeHeaders = encodeHeaders; + return this; + } + public Builder setTracer(@Nullable Tracer tracer) { this.tracer = tracer; return this; @@ -116,7 +135,8 @@ public OpenTracingOptions build() { MoreObjects.firstNonNull(tracer, GlobalTracer.get()), spanBuilderProvider, spanContextCodec, - isErrorPredicate); + isErrorPredicate, + encodeHeaders); } } } diff --git a/contrib/temporal-opentracing/src/main/java/io/temporal/opentracing/internal/ContextAccessor.java b/contrib/temporal-opentracing/src/main/java/io/temporal/opentracing/internal/ContextAccessor.java index cd847bdf4e..2241231561 100644 --- a/contrib/temporal-opentracing/src/main/java/io/temporal/opentracing/internal/ContextAccessor.java +++ b/contrib/temporal-opentracing/src/main/java/io/temporal/opentracing/internal/ContextAccessor.java @@ -13,6 +13,7 @@ import io.temporal.opentracing.OpenTracingOptions; import io.temporal.opentracing.OpenTracingSpanContextCodec; import java.lang.reflect.Type; +import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.Optional; @@ -24,9 +25,11 @@ public class ContextAccessor { new TypeToken>() {}.getType(); private final OpenTracingSpanContextCodec codec; + private final boolean encodeHeaders; public ContextAccessor(OpenTracingOptions options) { this.codec = options.getSpanContextCodec(); + this.encodeHeaders = options.isEncodeHeaders(); } public Span writeSpanContextToHeader( @@ -40,11 +43,13 @@ public void writeSpanContextToHeader(SpanContext spanContext, Header header, Tra Map serializedSpanContext = codec.encode(spanContext, tracer); Optional payload = DefaultDataConverter.STANDARD_INSTANCE.toPayload(serializedSpanContext); - header.getValues().put(TRACER_HEADER_KEY, payload.get()); + header.putAll(Collections.singletonMap(TRACER_HEADER_KEY, payload.get()), encodeHeaders); } public SpanContext readSpanContextFromHeader(Header header, Tracer tracer) { - Payload payload = header.getValues().get(TRACER_HEADER_KEY); + Payload payload = + (encodeHeaders ? header.getCodecDecodedValues() : header.getValues()) + .get(TRACER_HEADER_KEY); if (payload == null) { return null; } diff --git a/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/HeaderCodecTest.java b/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/HeaderCodecTest.java new file mode 100644 index 0000000000..2df391c676 --- /dev/null +++ b/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/HeaderCodecTest.java @@ -0,0 +1,88 @@ +package io.temporal.opentracing; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; + +import com.google.protobuf.ByteString; +import io.opentracing.Span; +import io.opentracing.mock.MockTracer; +import io.temporal.api.common.v1.Payload; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.interceptors.Header; +import io.temporal.internal.common.HeaderUtils; +import io.temporal.opentracing.internal.ContextAccessor; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.codec.PayloadCodecException; +import java.util.Collections; +import java.util.List; +import javax.annotation.Nonnull; +import org.junit.Test; + +public class HeaderCodecTest { + @Test + public void tracingKeepsStandardSerializationAndUsesHeaderCodecExactlyOnce() { + for (boolean enabled : new boolean[] {false, true}) { + MockTracer tracer = new MockTracer(); + ContextAccessor accessor = + new ContextAccessor( + OpenTracingOptions.newBuilder().setTracer(tracer).setEncodeHeaders(enabled).build()); + Span span = tracer.buildSpan("parent").start(); + CountingCodec codec = new CountingCodec(); + DataConverter valuesConverter = mock(DataConverter.class); + CodecDataConverter converter = + new CodecDataConverter(valuesConverter, Collections.singletonList(codec)); + PayloadCodec headerCodec = HeaderUtils.resolveHeaderCodec(converter); + Header header = Header.empty(); + accessor.writeSpanContextToHeader(span.context(), header, tracer); + Payload plaintext = DefaultDataConverter.STANDARD_INSTANCE.toPayload("other producer").get(); + header.getValues().put("other", plaintext); + io.temporal.api.common.v1.Header wire = HeaderUtils.toHeaderGrpc(header, null, headerCodec); + Header received = new Header(wire, headerCodec); + assertEquals(plaintext, received.getValues().get("other")); + + assertEquals( + span.context().toTraceId(), + accessor.readSpanContextFromHeader(received, tracer).toTraceId()); + assertEquals(enabled ? 1 : 0, codec.encoded); + assertEquals(enabled ? 1 : 0, codec.decoded); + assertEquals(wire.getFieldsMap(), received.getValues()); + verifyNoInteractions(valuesConverter); + span.finish(); + tracer.close(); + } + } + + static class CountingCodec implements PayloadCodec { + private int encoded; + private int decoded; + + @Override + @Nonnull + public List encode(@Nonnull List payloads) { + encoded++; + return Collections.singletonList( + Payload.newBuilder() + .putMetadata("encoding", ByteString.copyFromUtf8("binary/test-tracing")) + .setData(payloads.get(0).toByteString()) + .build()); + } + + @Override + @Nonnull + public List decode(@Nonnull List payloads) { + if (!ByteString.copyFromUtf8("binary/test-tracing") + .equals(payloads.get(0).getMetadataOrDefault("encoding", ByteString.EMPTY))) { + return payloads; + } + decoded++; + try { + return Collections.singletonList(Payload.parseFrom(payloads.get(0).getData())); + } catch (Exception e) { + throw new PayloadCodecException(e); + } + } + } +} diff --git a/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/SignalWithStartTest.java b/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/SignalWithStartTest.java index 4879a4ce39..3cf92a9db5 100644 --- a/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/SignalWithStartTest.java +++ b/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/SignalWithStartTest.java @@ -8,11 +8,14 @@ import io.opentracing.mock.MockTracer; import io.opentracing.util.ThreadLocalScopeManager; import io.temporal.client.*; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.DefaultDataConverter; import io.temporal.testing.internal.SDKTestWorkflowRule; import io.temporal.worker.WorkerFactoryOptions; import io.temporal.workflow.SignalMethod; import io.temporal.workflow.WorkflowInterface; import io.temporal.workflow.WorkflowMethod; +import java.util.Collections; import java.util.List; import org.junit.After; import org.junit.Rule; @@ -24,13 +27,17 @@ public class SignalWithStartTest { new MockTracer(new ThreadLocalScopeManager(), MockTracer.Propagator.TEXT_MAP); private final OpenTracingOptions OT_OPTIONS = - OpenTracingOptions.newBuilder().setTracer(mockTracer).build(); + OpenTracingOptions.newBuilder().setTracer(mockTracer).setEncodeHeaders(true).build(); @Rule public SDKTestWorkflowRule testWorkflowRule = SDKTestWorkflowRule.newBuilder() .setWorkflowClientOptions( WorkflowClientOptions.newBuilder() + .setDataConverter( + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, + Collections.singletonList(new HeaderCodecTest.CountingCodec()))) .setInterceptors(new OpenTracingClientInterceptor(OT_OPTIONS)) .validateAndBuildWithDefaults()) .setWorkerFactoryOptions( diff --git a/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/SpanContextPropagationTest.java b/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/SpanContextPropagationTest.java index e6496e5cba..907ce83041 100644 --- a/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/SpanContextPropagationTest.java +++ b/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/SpanContextPropagationTest.java @@ -13,12 +13,16 @@ import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowClientOptions; import io.temporal.client.WorkflowOptions; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.DefaultDataConverter; import io.temporal.testing.internal.SDKTestWorkflowRule; import io.temporal.worker.WorkerFactoryOptions; +import io.temporal.workflow.QueryMethod; import io.temporal.workflow.Workflow; import io.temporal.workflow.WorkflowInterface; import io.temporal.workflow.WorkflowMethod; import java.time.Duration; +import java.util.Collections; import org.junit.After; import org.junit.Rule; import org.junit.Test; @@ -31,13 +35,17 @@ public class SpanContextPropagationTest { new MockTracer(new ThreadLocalScopeManager(), MockTracer.Propagator.TEXT_MAP); private final OpenTracingOptions OT_OPTIONS = - OpenTracingOptions.newBuilder().setTracer(mockTracer).build(); + OpenTracingOptions.newBuilder().setTracer(mockTracer).setEncodeHeaders(true).build(); @Rule public SDKTestWorkflowRule testWorkflowRule = SDKTestWorkflowRule.newBuilder() .setWorkflowClientOptions( WorkflowClientOptions.newBuilder() + .setDataConverter( + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, + Collections.singletonList(new HeaderCodecTest.CountingCodec()))) .setInterceptors(new OpenTracingClientInterceptor(OT_OPTIONS)) .validateAndBuildWithDefaults()) .setWorkerFactoryOptions( @@ -63,6 +71,9 @@ public interface TestActivity { public interface TestWorkflow { @WorkflowMethod String workflow1(String input); + + @QueryMethod + String query(); } public static class OpenTracingAwareActivityImpl implements TestActivity { @@ -96,6 +107,11 @@ public String workflow1(String input) { return activity.activity1(input); } + + @Override + public String query() { + return mockTracer.activeSpan().getBaggageItem(BAGGAGE_ITEM_KEY); + } } /** @@ -121,6 +137,7 @@ public void testBaggageItemPropagationToActivity() { .setTaskQueue(testWorkflowRule.getTaskQueue()) .validateBuildWithDefaults()); assertEquals(BAGGAGE_ITEM_VALUE, workflow.workflow1("input")); + assertEquals(BAGGAGE_ITEM_VALUE, workflow.query()); } finally { span.finish(); } diff --git a/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/UpdateWithStartTest.java b/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/UpdateWithStartTest.java index 12a92aad40..302a342b39 100644 --- a/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/UpdateWithStartTest.java +++ b/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/UpdateWithStartTest.java @@ -9,6 +9,8 @@ import io.opentracing.util.ThreadLocalScopeManager; import io.temporal.api.enums.v1.WorkflowIdConflictPolicy; import io.temporal.client.*; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.DefaultDataConverter; import io.temporal.testing.internal.SDKTestWorkflowRule; import io.temporal.worker.WorkerFactoryOptions; import io.temporal.workflow.CompletablePromise; @@ -17,6 +19,7 @@ import io.temporal.workflow.WorkflowInterface; import io.temporal.workflow.WorkflowMethod; import java.util.Arrays; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; @@ -31,13 +34,17 @@ public class UpdateWithStartTest { new MockTracer(new ThreadLocalScopeManager(), MockTracer.Propagator.TEXT_MAP); private final OpenTracingOptions OT_OPTIONS = - OpenTracingOptions.newBuilder().setTracer(mockTracer).build(); + OpenTracingOptions.newBuilder().setTracer(mockTracer).setEncodeHeaders(true).build(); @Rule public SDKTestWorkflowRule testWorkflowRule = SDKTestWorkflowRule.newBuilder() .setWorkflowClientOptions( WorkflowClientOptions.newBuilder() + .setDataConverter( + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, + Collections.singletonList(new HeaderCodecTest.CountingCodec()))) .setInterceptors(new OpenTracingClientInterceptor(OT_OPTIONS)) .validateAndBuildWithDefaults()) .setWorkerFactoryOptions( 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 06130f3804..e3d5d71adf 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java @@ -1,12 +1,9 @@ package io.temporal.client; import com.uber.m3.tally.Scope; -import io.temporal.api.common.v1.Payload; -import io.temporal.common.context.ContextPropagator; import io.temporal.common.converter.DataConverter; import io.temporal.common.interceptors.ActivityClientCallsInterceptor; import io.temporal.common.interceptors.ActivityClientInterceptor; -import io.temporal.common.interceptors.Header; import io.temporal.internal.client.ActivityClientInternal; import io.temporal.internal.client.ActivityHandleImpl; import io.temporal.internal.client.ClientDataConverterFactory; @@ -14,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.HeaderUtils; import io.temporal.internal.util.MethodExtractor; import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; @@ -21,9 +19,6 @@ import java.lang.reflect.Method; import java.lang.reflect.Type; import java.util.Arrays; -import java.util.HashMap; -import java.util.List; -import java.util.Map; import java.util.stream.Stream; import javax.annotation.Nullable; @@ -311,7 +306,7 @@ public UntypedActivityHandle start( activityType, Arrays.asList(args != null ? args : new Object[0]), options, - propagatedHeader())); + HeaderUtils.fromContextPropagators(this.options.getContextPropagators()))); return new ActivityHandleImpl(output.getActivityId(), output.getActivityRunId(), invoker); } @@ -358,18 +353,6 @@ public ActivityHandle getHandle( return ActivityHandle.fromUntyped(untyped, resultClass, resultType); } - private Header propagatedHeader() { - List propagators = options.getContextPropagators(); - if (propagators.isEmpty()) { - return Header.empty(); - } - Map result = new HashMap<>(); - for (ContextPropagator propagator : propagators) { - result.putAll(propagator.serializeContext(propagator.getCurrentContext())); - } - return new Header(result); - } - // ---- List / count ---- @Override diff --git a/temporal-sdk/src/main/java/io/temporal/common/context/ContextPropagator.java b/temporal-sdk/src/main/java/io/temporal/common/context/ContextPropagator.java index 16e068e363..a37ebb0e37 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/context/ContextPropagator.java +++ b/temporal-sdk/src/main/java/io/temporal/common/context/ContextPropagator.java @@ -97,6 +97,19 @@ */ public interface ContextPropagator { + /** + * Whether to apply the configured client's payload codec to this propagator's headers. + * + *

Defaults to false. When enabled, {@link #serializeContext} must return serialized, not + * codec-encoded, payloads. The SDK encodes only this propagator's output and supplies a privately + * decoded header map to {@link #deserializeContext}. Other producers' headers are unchanged. + * Configure the same setting on receiving workers. Codecs must pass through encodings they do not + * recognize. + */ + default boolean encodeHeaders() { + return false; + } + /** * Returns the unique name of the context propagator. This name is used to store and address * context objects to pass them between threads. This name is not used for serialization, @@ -113,6 +126,9 @@ public interface ContextPropagator { * interceptors, because they all share the same single keyspace of a single instance of {@code * temporal.api.common.v1.Header} * + *

When {@link #encodeHeaders()} returns true, return serialized but not codec-encoded + * payloads. + * * @return serialized representation to be sent in request {@code temporal.api.common.v1.Header} */ Map serializeContext(Object context); diff --git a/temporal-sdk/src/main/java/io/temporal/common/converter/CodecDataConverter.java b/temporal-sdk/src/main/java/io/temporal/common/converter/CodecDataConverter.java index a82172348b..af6aaffed4 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/converter/CodecDataConverter.java +++ b/temporal-sdk/src/main/java/io/temporal/common/converter/CodecDataConverter.java @@ -11,6 +11,7 @@ import io.temporal.api.failure.v1.TimeoutFailureInfo; import io.temporal.payload.codec.ChainCodec; import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.codec.PayloadCodecProvider; import io.temporal.payload.context.SerializationContext; import java.lang.reflect.Type; import java.util.Collection; @@ -27,8 +28,11 @@ *

The underlying {@link DataConverter} is expected to be responsible for conversion between user * objects and bytes represented as {@link Payloads}, while the underlying chain of codecs is * responsible for a subsequent byte <-> byte manipulation such as encryption or compression + * + *

This converter exposes its codec stage for tracing and context propagators that opt into + * header codec processing. Ordinary user-supplied header payloads remain wire-ready. */ -public class CodecDataConverter implements DataConverter, PayloadCodec { +public class CodecDataConverter implements DataConverter, PayloadCodecProvider, PayloadCodec { private static final String ENCODED_FAILURE_MESSAGE = "Encoded failure"; private final DataConverter dataConverter; @@ -190,6 +194,12 @@ public CodecDataConverter withContext(@Nonnull SerializationContext context) { return new CodecDataConverter(dataConverter, chainCodec, encodeFailureAttributes, context); } + @Nullable + @Override + public PayloadCodec getPayloadCodec() { + return ConverterUtils.withContext(chainCodec, serializationContext); + } + @Nonnull @Override public List encode(@Nonnull List payloads) { diff --git a/temporal-sdk/src/main/java/io/temporal/common/interceptors/Header.java b/temporal-sdk/src/main/java/io/temporal/common/interceptors/Header.java index bbf15d579b..314467fe46 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/interceptors/Header.java +++ b/temporal-sdk/src/main/java/io/temporal/common/interceptors/Header.java @@ -1,15 +1,34 @@ package io.temporal.common.interceptors; import io.temporal.api.common.v1.Payload; +import io.temporal.internal.common.HeaderUtils; +import io.temporal.payload.codec.PayloadCodec; import java.util.HashMap; import java.util.Map; import javax.annotation.Nonnull; +import javax.annotation.Nullable; +/** + * Already serialized payload headers shared by interceptors and context propagators. + * + *

Entries supplied through {@link #getValues()} should be wire-ready, including any codec + * encoding. Tracing and context propagation may separately opt into SDK codec processing of their + * own output. Encoding opt-in belongs to a key and payload instance: removing and reinserting that + * same instance retains its opt-in. Use {@link #putAll(Map, boolean)} with {@code false} to + * explicitly clear it. + */ public class Header { private final Map values; + private final Map codecValues = new HashMap<>(); + private final @Nullable PayloadCodec codec; public Header(@Nonnull io.temporal.api.common.v1.Header header) { - values = header.getFieldsMap(); + this(header, null); + } + + public Header(@Nonnull io.temporal.api.common.v1.Header header, @Nullable PayloadCodec codec) { + this.values = header.getFieldsMap(); + this.codec = codec; } public static Header empty() { @@ -18,9 +37,32 @@ public static Header empty() { public Header(Map values) { this.values = values; + this.codec = null; } public Map getValues() { return values; } + + /** Used by opted-in header producers; ordinary interceptor entries remain wire-ready. */ + public void putAll(Map payloads, boolean encode) { + values.putAll(payloads); + payloads.keySet().forEach(codecValues::remove); + if (encode) { + codecValues.putAll(payloads); + } + } + + /** Returns whether this producer requested codec encoding for an entry. */ + public boolean shouldEncode(String key) { + return codecValues.containsKey(key) && codecValues.get(key) == values.get(key); + } + + /** Returns a private decoded view for an opted-in tracing or propagation consumer. */ + public Map getCodecDecodedValues() { + return new HashMap<>( + HeaderUtils.decode( + io.temporal.api.common.v1.Header.newBuilder().putAllFields(values).build(), codec) + .getFieldsMap()); + } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java index 50ade355f9..158e262c37 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java @@ -17,7 +17,9 @@ import io.temporal.common.interceptors.Header; import io.temporal.common.interceptors.WorkerInterceptor; import io.temporal.internal.common.FailureUtils; +import io.temporal.internal.common.HeaderUtils; import io.temporal.internal.worker.ActivityTaskHandler; +import io.temporal.payload.codec.PayloadCodec; import io.temporal.payload.context.ActivitySerializationContext; import io.temporal.payload.context.NexusSerializationContext; import io.temporal.serviceclient.CheckedExceptionWrapper; @@ -40,6 +42,7 @@ interface ActivityTaskExecutor { abstract static class BaseActivityTaskExecutor implements ActivityTaskExecutor { private final DataConverter dataConverter; + private final @Nullable PayloadCodec headerCodec; private final List contextPropagators; private final WorkerInterceptor[] interceptors; private final ActivityExecutionContextFactory executionContextFactory; @@ -50,6 +53,7 @@ public BaseActivityTaskExecutor( WorkerInterceptor[] interceptors, ActivityExecutionContextFactory executionContextFactory) { this.dataConverter = dataConverter; + this.headerCodec = HeaderUtils.resolveHeaderCodec(dataConverter); this.contextPropagators = contextPropagators; this.interceptors = interceptors; this.executionContextFactory = executionContextFactory; @@ -80,8 +84,13 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri : dataConverterWithActivityContext; try { - info.getHeader() - .ifPresent(value -> deserializeAndPopulateContext(value, contextPropagators)); + Header header = + new Header( + info.getHeader().orElse(io.temporal.api.common.v1.Header.getDefaultInstance()), + HeaderUtils.withContext(headerCodec, serializationContext)); + if (info.getHeader().isPresent()) { + deserializeAndPopulateContext(header, contextPropagators); + } ActivityInboundCallsInterceptor inboundCallsInterceptor = createRootInboundInterceptor(); for (WorkerInterceptor interceptor : interceptors) { @@ -90,9 +99,6 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri inboundCallsInterceptor.init(context); Object[] args = provideArgs(info.getInput(), dataConverterWithActivityContext); - Header header = - new Header( - info.getHeader().orElse(io.temporal.api.common.v1.Header.getDefaultInstance())); ActivityOutput result = inboundCallsInterceptor.execute( new ActivityInboundCallsInterceptor.ActivityInput(header, args)); @@ -176,14 +182,16 @@ ActivityTaskHandler.Result constructResultValue( } static void deserializeAndPopulateContext( - @Nonnull io.temporal.api.common.v1.Header header, - @Nullable List contextPropagatorList) { + @Nonnull Header header, @Nullable List contextPropagatorList) { if (contextPropagatorList == null || contextPropagatorList.isEmpty()) { return; } - Map headerData = new HashMap<>(header.getFieldsMap()); for (ContextPropagator propagator : contextPropagatorList) { + Map headerData = + propagator.encodeHeaders() + ? header.getCodecDecodedValues() + : new HashMap<>(header.getValues()); propagator.setCurrentContext(propagator.deserializeContext(headerData)); } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java index 6c8f801cf6..1b0f56eaa4 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java @@ -34,6 +34,7 @@ import io.temporal.internal.nexus.InternalNexusOperationContext; import io.temporal.internal.nexus.NexusOperationMetadata; import io.temporal.internal.nexus.PropagatedNexusSerializationContext; +import io.temporal.payload.codec.PayloadCodec; import io.temporal.payload.context.ActivitySerializationContext; import io.temporal.payload.context.SerializationContext; import io.temporal.serviceclient.StatusUtils; @@ -55,6 +56,7 @@ public class RootActivityClientInvoker implements ActivityClientCallsInterceptor private final String namespace; private final String identity; private final DataConverter dataConverter; + private final @Nullable PayloadCodec headerCodec; public RootActivityClientInvoker( GenericWorkflowClient genericClient, @@ -65,6 +67,7 @@ public RootActivityClientInvoker( this.namespace = namespace; this.identity = identity; this.dataConverter = dataConverter; + this.headerCodec = HeaderUtils.resolveHeaderCodec(dataConverter); } @Override @@ -80,10 +83,10 @@ public StartActivityOutput startActivity(StartActivityInput input) { @Nullable PropagatedSerializationContext propagatedNexusSerializationContext = PropagatedNexusSerializationContext.current(); - DataConverter dc = - dataConverter.withContext( - new ActivitySerializationContext( - namespace, null, null, input.getActivityType(), options.getTaskQueue(), false)); + ActivitySerializationContext serializationContext = + new ActivitySerializationContext( + namespace, null, null, input.getActivityType(), options.getTaskQueue(), false); + DataConverter dc = dataConverter.withContext(serializationContext); InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.isNexusContext() ? CurrentNexusOperationContext.get() : null; @@ -149,7 +152,9 @@ public StartActivityOutput startActivity(StartActivityInput input) { request.setStartDelay(ProtobufTimeUtils.toProtoDuration(options.getStartDelay())); } - io.temporal.api.common.v1.Header grpcHeader = HeaderUtils.toHeaderGrpc(input.getHeader(), null); + io.temporal.api.common.v1.Header grpcHeader = + HeaderUtils.toHeaderGrpc( + input.getHeader(), null, HeaderUtils.withContext(headerCodec, serializationContext)); request.setHeader(grpcHeader); List protoLinks = Collections.emptyList(); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java index 147a7bb7a9..e636678434 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java @@ -39,6 +39,8 @@ import io.temporal.internal.nexus.PropagatedNexusSerializationContext; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.worker.WorkerVersioningProtoUtils; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.context.WorkflowSerializationContext; import io.temporal.payload.storage.StorageDriverWorkflowInfo; import io.temporal.serviceclient.StatusUtils; import io.temporal.worker.WorkflowTaskDispatchHandle; @@ -64,6 +66,7 @@ public class RootWorkflowClientInvoker implements WorkflowClientCallsInterceptor private final WorkflowClientRequestFactory requestsHelper; private final ClientDataConverterFactory converterFactory; private final @Nullable ExternalStorageRunner externalStorage; + private final @Nullable PayloadCodec headerCodec; public RootWorkflowClientInvoker( GenericWorkflowClient genericClient, @@ -96,18 +99,29 @@ public RootWorkflowClientInvoker( this.converterFactory = new ClientDataConverterFactory(namespace, dataConverter, externalStorage); this.externalStorage = externalStorage; + this.headerCodec = HeaderUtils.resolveHeaderCodec(dataConverter); this.genericClient = genericClient; this.namespace = namespace; this.identity = identity; this.queryRejectCondition = queryRejectCondition; this.eagerWorkflowTaskDispatcher = new EagerWorkflowTaskDispatcher(workerFactoryRegistry); - this.requestsHelper = new WorkflowClientRequestFactory(namespace, identity, contextPropagators); + this.requestsHelper = + new WorkflowClientRequestFactory(namespace, identity, contextPropagators, headerCodec); } private DataConverter workflowConverter(WorkflowExecution execution) { return workflowConverter(execution, null); } + private Header workflowHeader( + io.temporal.common.interceptors.Header header, WorkflowExecution execution) { + return HeaderUtils.toHeaderGrpc( + header, + null, + HeaderUtils.withContext( + headerCodec, new WorkflowSerializationContext(namespace, execution.getWorkflowId()))); + } + private DataConverter workflowConverter( WorkflowExecution execution, @Nullable String workflowType) { return workflowConverter(execution.getWorkflowId(), execution.getRunId(), workflowType); @@ -223,7 +237,7 @@ public WorkflowSignalOutput signal(WorkflowSignalInput input) { .setIdentity(identity) .setNamespace(namespace) .setRequestId(UUID.randomUUID().toString()) - .setHeader(HeaderUtils.toHeaderGrpc(input.getHeader(), null)); + .setHeader(workflowHeader(input.getHeader(), input.getWorkflowExecution())); // If this signal is being issued from inside a Nexus operation handler, forward the inbound // Nexus task links so the SignalWorkflowExecution history event links back to the caller. @@ -405,13 +419,7 @@ public WorkflowUpdateWithStartOutput updateWithStart( WorkflowExecution execution = WorkflowExecution.newBuilder() .setRunId(startResponse.getRunId()) - .setWorkflowId( - toStartRequest( - dataConverterWithWorkflowContext, - startInput, - propagatedNexusSerializationContext) - .build() - .getWorkflowId()) + .setWorkflowId(startInput.getWorkflowId()) .build(); return new WorkflowUpdateWithStartOutput<>(new WorkflowStartOutput(execution), updateHandle); } @@ -516,7 +524,7 @@ public QueryOutput query(QueryInput input) { WorkflowQuery.Builder query = WorkflowQuery.newBuilder() .setQueryType(input.getQueryType()) - .setHeader(HeaderUtils.toHeaderGrpc(input.getHeader(), null)); + .setHeader(workflowHeader(input.getHeader(), input.getWorkflowExecution())); DataConverter dataConverterWithWorkflowContext = workflowConverter(input.getWorkflowExecution()); @@ -620,7 +628,7 @@ private UpdateWorkflowExecutionRequest toUpdateWorkflowExecutionRequest( dataConverterWithWorkflowContext.toPayloads(input.getArguments()); Input.Builder updateInput = Input.newBuilder() - .setHeader(HeaderUtils.toHeaderGrpc(input.getHeader(), null)) + .setHeader(workflowHeader(input.getHeader(), input.getWorkflowExecution())) .setName(input.getUpdateName()); inputArgs.ifPresent(updateInput::setArgs); storeHeader( diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/ScheduleProtoUtil.java b/temporal-sdk/src/main/java/io/temporal/internal/client/ScheduleProtoUtil.java index 82065127ae..d9ccbdf5ac 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/ScheduleProtoUtil.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/ScheduleProtoUtil.java @@ -24,10 +24,12 @@ import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.EncodedValues; import io.temporal.internal.client.external.GenericWorkflowClient; +import io.temporal.internal.common.HeaderUtils; import io.temporal.internal.common.ProtoConverters; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.common.RetryOptionsUtils; import io.temporal.internal.common.SearchAttributesUtil; +import io.temporal.payload.codec.PayloadCodec; import io.temporal.payload.context.WorkflowSerializationContext; import java.time.Instant; import java.util.*; @@ -41,6 +43,7 @@ public class ScheduleProtoUtil { private final String namespace; private final List contextPropagators; private final DataConverter dataConverter; + private final @Nullable PayloadCodec headerCodec; public ScheduleProtoUtil( GenericWorkflowClient genericClient, @@ -51,23 +54,7 @@ public ScheduleProtoUtil( this.namespace = namespace; this.contextPropagators = contextPropagators; this.dataConverter = dataConverter; - } - - private io.temporal.common.interceptors.Header extractContextsAndConvertToBytes( - List scheduleOptionsContextPropagators) { - if ((contextPropagators.isEmpty() && scheduleOptionsContextPropagators == null) - || (scheduleOptionsContextPropagators != null - && scheduleOptionsContextPropagators.isEmpty())) { - return null; - } - - List listToUse = - MoreObjects.firstNonNull(scheduleOptionsContextPropagators, contextPropagators); - Map result = new HashMap<>(); - for (ContextPropagator propagator : listToUse) { - result.putAll(propagator.serializeContext(propagator.getCurrentContext())); - } - return new io.temporal.common.interceptors.Header(result); + this.headerCodec = HeaderUtils.resolveHeaderCodec(dataConverter); } public ScheduleAction actionToProto(io.temporal.client.schedules.ScheduleAction action) { @@ -157,7 +144,11 @@ public ScheduleAction actionToProto(io.temporal.client.schedules.ScheduleAction Header grpcHeader = toHeaderGrpc( startWorkflowAction.getHeader(), - extractContextsAndConvertToBytes(wfOptions.getContextPropagators())); + HeaderUtils.fromContextPropagators( + MoreObjects.firstNonNull(wfOptions.getContextPropagators(), contextPropagators)), + HeaderUtils.withContext( + headerCodec, + new WorkflowSerializationContext(namespace, wfOptions.getWorkflowId()))); workflowRequest.setHeader(grpcHeader); if (wfOptions.getPriority() != null) { @@ -417,6 +408,7 @@ public io.temporal.client.schedules.ScheduleAction protoToAction(@Nonnull Schedu ScheduleActionStartWorkflow.Builder builder = ScheduleActionStartWorkflow.newBuilder(); builder.setWorkflowType(startWfAction.getWorkflowType().getName()); + builder.setHeader(new io.temporal.common.interceptors.Header(startWfAction.getHeader())); builder.setRawArguments( new EncodedValues( diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientRequestFactory.java b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientRequestFactory.java index 61f9579f5a..c81b13db87 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientRequestFactory.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientRequestFactory.java @@ -18,9 +18,12 @@ import io.temporal.client.WorkflowOptions; import io.temporal.common.RetryOptions; import io.temporal.common.context.ContextPropagator; +import io.temporal.internal.common.HeaderUtils; import io.temporal.internal.common.ProtoConverters; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.common.SearchAttributesUtil; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.context.WorkflowSerializationContext; import java.util.*; import javax.annotation.Nonnull; import javax.annotation.Nullable; @@ -29,12 +32,17 @@ final class WorkflowClientRequestFactory { private final String namespace; private final String identity; private final List contextPropagators; + private final @Nullable PayloadCodec headerCodec; public WorkflowClientRequestFactory( - String namespace, String identity, List contextPropagators) { + String namespace, + String identity, + List contextPropagators, + @Nullable PayloadCodec headerCodec) { this.namespace = namespace; this.identity = identity; this.contextPropagators = contextPropagators; + this.headerCodec = headerCodec; } // If you add anything new here, keep newSignalWithStartWorkflowExecutionRequest in sync @@ -143,7 +151,12 @@ StartWorkflowExecutionRequest.Builder newStartWorkflowExecutionRequest( } Header grpcHeader = - toHeaderGrpc(header, extractContextsAndConvertToBytes(options.getContextPropagators())); + toHeaderGrpc( + header, + HeaderUtils.fromContextPropagators( + MoreObjects.firstNonNull(options.getContextPropagators(), contextPropagators)), + HeaderUtils.withContext( + headerCodec, new WorkflowSerializationContext(namespace, workflowId))); request.setHeader(grpcHeader); return request; @@ -232,23 +245,6 @@ GetWorkflowExecutionHistoryRequest newHistoryLongPollRequest( .build(); } - private io.temporal.common.interceptors.Header extractContextsAndConvertToBytes( - List workflowOptionsContextPropagators) { - if ((contextPropagators.isEmpty() && workflowOptionsContextPropagators == null) - || (workflowOptionsContextPropagators != null - && workflowOptionsContextPropagators.isEmpty())) { - return null; - } - - List listToUse = - MoreObjects.firstNonNull(workflowOptionsContextPropagators, contextPropagators); - Map result = new HashMap<>(); - for (ContextPropagator propagator : listToUse) { - result.putAll(propagator.serializeContext(propagator.getCurrentContext())); - } - return new io.temporal.common.interceptors.Header(result); - } - private static String generateUniqueId() { return UUID.randomUUID().toString(); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/common/HeaderUtils.java b/temporal-sdk/src/main/java/io/temporal/internal/common/HeaderUtils.java index 73de3e429b..7ca53d4623 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/common/HeaderUtils.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/common/HeaderUtils.java @@ -1,11 +1,20 @@ package io.temporal.internal.common; +import com.google.common.base.Suppliers; import io.temporal.api.common.v1.Header; import io.temporal.api.common.v1.Payload; +import io.temporal.common.context.ContextPropagator; import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.DataConverterException; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.codec.PayloadCodecException; +import io.temporal.payload.codec.PayloadCodecProvider; +import io.temporal.payload.context.SerializationContext; +import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.function.Supplier; import javax.annotation.Nullable; public class HeaderUtils { @@ -13,13 +22,104 @@ public class HeaderUtils { public static Header toHeaderGrpc( io.temporal.common.interceptors.Header header, @Nullable io.temporal.common.interceptors.Header overrides) { - Header.Builder builder = Header.newBuilder().putAllFields(header.getValues()); - if (overrides != null) { - for (Map.Entry item : overrides.getValues().entrySet()) { - builder.putFields(item.getKey(), item.getValue()); + return toHeaderGrpc(header, overrides, null); + } + + public static Header toHeaderGrpc( + io.temporal.common.interceptors.Header header, + @Nullable io.temporal.common.interceptors.Header overrides, + @Nullable PayloadCodec codec) { + Header.Builder result = Header.newBuilder(); + Header.Builder selected = Header.newBuilder(); + for (io.temporal.common.interceptors.Header source : + new io.temporal.common.interceptors.Header[] {header, overrides}) { + if (source == null) { + continue; + } + for (Map.Entry entry : source.getValues().entrySet()) { + result.putFields(entry.getKey(), entry.getValue()); + selected.removeFields(entry.getKey()); + if (source.shouldEncode(entry.getKey())) { + selected.putFields(entry.getKey(), entry.getValue()); + } + } + } + result.putAllFields(encode(selected.build(), codec).getFieldsMap()); + return result.build(); + } + + public static io.temporal.common.interceptors.Header fromContextPropagators( + @Nullable List propagators) { + io.temporal.common.interceptors.Header result = io.temporal.common.interceptors.Header.empty(); + if (propagators != null) { + for (ContextPropagator propagator : propagators) { + result.putAll( + propagator.serializeContext(propagator.getCurrentContext()), + propagator.encodeHeaders()); + } + } + return result; + } + + public static Header encode(Header header, @Nullable PayloadCodec codec) { + return transform(header, codec, true); + } + + public static Header decode(Header header, @Nullable PayloadCodec codec) { + return transform(header, codec, false); + } + + @Nullable + public static PayloadCodec resolveHeaderCodec(DataConverter converter) { + return converter instanceof PayloadCodecProvider + ? ((PayloadCodecProvider) converter).getPayloadCodec() + : null; + } + + @Nullable + public static PayloadCodec withContext( + @Nullable PayloadCodec codec, SerializationContext context) { + if (codec == null) { + return null; + } + // Binding must not inspect or reject codec stages for producers that did not opt in. + Supplier bound = + Suppliers.memoize( + () -> { + PayloadCodec contextual = codec.withContext(context); + if (contextual == null) { + throw new IllegalArgumentException("Header codec binding must not return null"); + } + return contextual; + }); + return new PayloadCodec() { + @Override + public List encode(List payloads) { + return bound.get().encode(payloads); + } + + @Override + public List decode(List payloads) { + return bound.get().decode(payloads); + } + }; + } + + private static Header transform(Header header, @Nullable PayloadCodec codec, boolean encode) { + if (codec == null || header.getFieldsCount() == 0) { + return header; + } + Header.Builder result = Header.newBuilder(); + for (Map.Entry entry : header.getFieldsMap().entrySet()) { + List input = Collections.singletonList(entry.getValue()); + List output = encode ? codec.encode(input) : codec.decode(input); + if (output.size() != 1) { + throw new PayloadCodecException( + "Header codec must return one payload for key " + entry.getKey()); } + result.putFields(entry.getKey(), output.get(0)); } - return builder.build(); + return result.build(); } /** diff --git a/temporal-sdk/src/main/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverter.java b/temporal-sdk/src/main/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverter.java index a5d32b36d1..aeb05e94b9 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverter.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverter.java @@ -10,6 +10,9 @@ import io.temporal.common.converter.RawValue; import io.temporal.common.converter.TransferTypeConverter; import io.temporal.common.converter.TransferTypeConvertible; +import io.temporal.internal.common.HeaderUtils; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.codec.PayloadCodecProvider; import io.temporal.payload.context.SerializationContext; import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; @@ -18,9 +21,11 @@ import java.util.Arrays; import java.util.Optional; import javax.annotation.Nonnull; +import javax.annotation.Nullable; /** Applies type-owned transfer conversion around an SDK-managed data converter. */ -public final class TemporalTransferTypeDataConverter implements DataConverter { +public final class TemporalTransferTypeDataConverter + implements DataConverter, PayloadCodecProvider { private static final ClassValue CONVERTER_DESCRIPTORS = new ClassValue() { @Override @@ -37,6 +42,12 @@ private TemporalTransferTypeDataConverter(DataConverter delegate) { this.delegate = delegate; } + @Nullable + @Override + public PayloadCodec getPayloadCodec() { + return HeaderUtils.resolveHeaderCodec(delegate); + } + /** Wraps a converter once. */ public static DataConverter wrap(DataConverter converter) { if (converter instanceof TemporalTransferTypeDataConverter diff --git a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageDataConverter.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageDataConverter.java index 1f8c1198c7..c229caa260 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageDataConverter.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageDataConverter.java @@ -6,6 +6,9 @@ import io.temporal.common.CancellationToken; import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.DataConverterException; +import io.temporal.internal.common.HeaderUtils; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.codec.PayloadCodecProvider; import io.temporal.payload.context.SerializationContext; import io.temporal.payload.storage.StorageDriverTargetInfo; import java.lang.reflect.Type; @@ -19,7 +22,7 @@ *

This is an internal class that is not exposed to users or workflow code. The intent is to use * this data converter to consolidate extstore usage within the SDK. */ -public final class ExternalStorageDataConverter implements DataConverter { +public final class ExternalStorageDataConverter implements DataConverter, PayloadCodecProvider { private final DataConverter delegate; private final @Nullable ExternalStorageRunner externalStorage; @@ -44,6 +47,12 @@ public ExternalStorageDataConverter withStorageTarget( return new ExternalStorageDataConverter(delegate, externalStorage, storageTarget); } + @Nullable + @Override + public PayloadCodec getPayloadCodec() { + return HeaderUtils.resolveHeaderCodec(delegate); + } + @Override public Optional toPayload(T value) throws DataConverterException { Optional converted = delegate.toPayload(value); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java index eb99bdc7a0..a87272658e 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java @@ -10,6 +10,7 @@ import com.google.common.base.MoreObjects; import com.google.common.base.Preconditions; +import com.google.common.base.Strings; import com.uber.m3.tally.Scope; import io.temporal.activity.ActivityOptions; import io.temporal.activity.LocalActivityOptions; @@ -45,6 +46,7 @@ import io.temporal.internal.replay.ReplayWorkflowContext; import io.temporal.internal.replay.WorkflowContext; import io.temporal.internal.statemachines.*; +import io.temporal.payload.codec.PayloadCodec; import io.temporal.payload.context.ActivitySerializationContext; import io.temporal.payload.context.NexusSerializationContext; import io.temporal.payload.context.WorkflowSerializationContext; @@ -81,6 +83,7 @@ final class SyncWorkflowContext implements WorkflowContext, WorkflowOutboundCall private final SyncWorkflowDefinition workflowDefinition; private final WorkflowImplementationOptions workflowImplementationOptions; private final DataConverter dataConverter; + private final @Nullable PayloadCodec headerCodec; // to be used in this class, should not be passed down. Pass the original #dataConverter instead private final DataConverter dataConverterWithCurrentWorkflowContext; private final List contextPropagators; @@ -123,6 +126,7 @@ public SyncWorkflowContext( this.workflowExecution = workflowExecution; this.workflowDefinition = workflowDefinition; this.dataConverter = dataConverter; + this.headerCodec = HeaderUtils.resolveHeaderCodec(dataConverter); this.dataConverterWithCurrentWorkflowContext = dataConverter.withContext( new WorkflowSerializationContext(namespace, workflowExecution.getWorkflowId())); @@ -271,19 +275,13 @@ public void applyLocalActivityOptions(Map activity @Override public ActivityOutput executeActivity(ActivityInput input) { - ActivitySerializationContext serializationContext = - new ActivitySerializationContext( - replayContext.getNamespace(), - replayContext.getWorkflowId(), - replayContext.getWorkflowType().getName(), - input.getActivityName(), - // input.getOptions().getTaskQueue() may be not specified, workflow task queue is used - // by the Server in this case - MoreObjects.firstNonNull( - input.getOptions().getTaskQueue(), replayContext.getTaskQueue()), - false); DataConverter dataConverterWithActivityContext = - dataConverter.withContext(serializationContext); + dataConverter.withContext( + activityContext( + input.getActivityName(), + MoreObjects.firstNonNull( + input.getOptions().getTaskQueue(), replayContext.getTaskQueue()), + false)); Optional args = dataConverterWithActivityContext.toPayloads(input.getArgs()); ActivityOutput> output = @@ -450,16 +448,9 @@ public void apply(Optional successOutput, LocalActivityFailedException @Override public LocalActivityOutput executeLocalActivity(LocalActivityInput input) { - ActivitySerializationContext serializationContext = - new ActivitySerializationContext( - replayContext.getNamespace(), - replayContext.getWorkflowId(), - replayContext.getWorkflowType().getName(), - input.getActivityName(), - replayContext.getTaskQueue(), - true); DataConverter dataConverterWithActivityContext = - dataConverter.withContext(serializationContext); + dataConverter.withContext( + activityContext(input.getActivityName(), replayContext.getTaskQueue(), true)); Optional payloads = dataConverterWithActivityContext.toPayloads(input.getArgs()); long originalScheduledTime = System.currentTimeMillis(); @@ -593,6 +584,17 @@ private CompletablePromise> executeLocalActivityLocally( return callback.result; } + private ActivitySerializationContext activityContext( + String name, String taskQueue, boolean local) { + return new ActivitySerializationContext( + replayContext.getNamespace(), + replayContext.getWorkflowId(), + replayContext.getWorkflowType().getName(), + name, + taskQueue, + local); + } + @SuppressWarnings("deprecation") private ExecuteActivityParameters constructExecuteActivityParameters( String name, @@ -636,7 +638,10 @@ private ExecuteActivityParameters constructExecuteActivityParameters( propagators = this.contextPropagators; } io.temporal.api.common.v1.Header grpcHeader = - toHeaderGrpc(header, extractContextsAndConvertToBytes(propagators)); + toHeaderGrpc( + header, + HeaderUtils.fromContextPropagators(propagators), + HeaderUtils.withContext(headerCodec, activityContext(name, taskQueue, false))); attributes.setHeader(grpcHeader); if (options.getVersioningIntent() != null) { @@ -697,7 +702,11 @@ private ExecuteLocalActivityParameters constructExecuteLocalActivityParameters( } io.temporal.api.common.v1.Header grpcHeader = - toHeaderGrpc(header, extractContextsAndConvertToBytes(contextPropagators)); + toHeaderGrpc( + header, + HeaderUtils.fromContextPropagators(contextPropagators), + HeaderUtils.withContext( + headerCodec, activityContext(name, replayContext.getTaskQueue(), true))); activityTask.setHeader(grpcHeader); input.ifPresent(activityTask::setInput); RetryOptions retryOptions = options.getRetryOptions(); @@ -978,7 +987,16 @@ private StartChildWorkflowExecutionParameters createChildWorkflowParameters( propagators = this.contextPropagators; } io.temporal.api.common.v1.Header grpcHeader = - toHeaderGrpc(header, extractContextsAndConvertToBytes(propagators)); + toHeaderGrpc( + header, + HeaderUtils.fromContextPropagators(propagators), + HeaderUtils.withContext( + headerCodec, + new WorkflowSerializationContext( + Strings.isNullOrEmpty(options.getNamespace()) + ? replayContext.getNamespace() + : options.getNamespace(), + workflowId))); attributes.setHeader(grpcHeader); ParentClosePolicy parentClosePolicy = options.getParentClosePolicy(); @@ -1003,18 +1021,6 @@ private StartChildWorkflowExecutionParameters createChildWorkflowParameters( attributes, options.getCancellationType(), metadata); } - private static Header extractContextsAndConvertToBytes( - List contextPropagators) { - if (contextPropagators == null) { - return null; - } - Map result = new HashMap<>(); - for (ContextPropagator propagator : contextPropagators) { - result.putAll(propagator.serializeContext(propagator.getCurrentContext())); - } - return new Header(result); - } - private static RuntimeException mapChildWorkflowException( Exception failure, DataConverter dataConverterWithChildWorkflowContext) { if (failure == null) { @@ -1372,7 +1378,14 @@ public SignalExternalOutput signalExternalWorkflow(SignalExternalInput input) { SignalExternalWorkflowExecutionCommandAttributes.newBuilder(); attributes.setSignalName(input.getSignalName()); attributes.setExecution(childExecution); - attributes.setHeader(HeaderUtils.toHeaderGrpc(input.getHeader(), null)); + attributes.setHeader( + HeaderUtils.toHeaderGrpc( + input.getHeader(), + null, + HeaderUtils.withContext( + headerCodec, + new WorkflowSerializationContext( + replayContext.getNamespace(), childExecution.getWorkflowId())))); Optional payloads = dataConverterWithChildWorkflowContext.toPayloads(input.getArgs()); payloads.ifPresent(attributes::setInput); CompletablePromise result = Workflow.newPromise(); @@ -1549,7 +1562,12 @@ public void continueAsNew(ContinueAsNewInput input) { ? options.getContextPropagators() : this.contextPropagators; io.temporal.api.common.v1.Header grpcHeader = - toHeaderGrpc(input.getHeader(), extractContextsAndConvertToBytes(propagators)); + toHeaderGrpc( + input.getHeader(), + HeaderUtils.fromContextPropagators(propagators), + HeaderUtils.withContext( + headerCodec, + new WorkflowSerializationContext(namespace, workflowExecution.getWorkflowId()))); attributes.setHeader(grpcHeader); Optional payloads = @@ -1688,15 +1706,31 @@ public List getContextPropagators() { return contextPropagators; } + Header headerWithCodec(io.temporal.api.common.v1.Header header) { + return new Header( + header, + HeaderUtils.withContext( + headerCodec, + new WorkflowSerializationContext(namespace, workflowExecution.getWorkflowId()))); + } + @Override public Map getPropagatedContexts() { if (contextPropagators == null || contextPropagators.isEmpty()) { return new HashMap<>(); } - Map headerData = new HashMap<>(replayContext.getHeader()); Map contextData = new HashMap<>(); + Header header = + headerWithCodec( + io.temporal.api.common.v1.Header.newBuilder() + .putAllFields(replayContext.getHeader()) + .build()); for (ContextPropagator propagator : contextPropagators) { + Map headerData = + propagator.encodeHeaders() + ? header.getCodecDecodedValues() + : new HashMap<>(header.getValues()); contextData.put(propagator.getName(), propagator.deserializeContext(headerData)); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowExecutionHandler.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowExecutionHandler.java index ddebca88a4..c79051c658 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowExecutionHandler.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowExecutionHandler.java @@ -6,7 +6,6 @@ import io.temporal.api.common.v1.Payloads; import io.temporal.api.history.v1.WorkflowExecutionStartedEventAttributes; import io.temporal.common.converter.DataConverter; -import io.temporal.common.interceptors.Header; import io.temporal.failure.CanceledFailure; import io.temporal.failure.TemporalFailure; import io.temporal.internal.replay.ReplayWorkflowContext; @@ -49,7 +48,7 @@ public void runWorkflowMethod() { try { Optional input = attributes.hasInput() ? Optional.of(attributes.getInput()) : Optional.empty(); - output = workflow.execute(new Header(attributes.getHeader()), input); + output = workflow.execute(context.headerWithCodec(attributes.getHeader()), input); } catch (Throwable e) { applyWorkflowFailurePolicyAndRethrow(e); } finally { @@ -86,7 +85,7 @@ public void handleSignal( long eventId, io.temporal.api.common.v1.Header header) { try { - context.handleSignal(signalName, input, eventId, new Header(header)); + context.handleSignal(signalName, input, eventId, context.headerWithCodec(header)); } catch (Throwable e) { applyWorkflowFailurePolicyAndRethrow(e); } @@ -94,7 +93,7 @@ public void handleSignal( public Optional handleQuery( String type, io.temporal.api.common.v1.Header header, Optional args) { - return context.handleQuery(type, new Header(header), args); + return context.handleQuery(type, context.headerWithCodec(header), args); } public void handleValidateUpdate( @@ -105,7 +104,8 @@ public void handleValidateUpdate( io.temporal.api.common.v1.Header header, DataConverter outcomeDataConverter) { try { - context.handleValidateUpdate(updateName, updateId, input, eventId, new Header(header)); + context.handleValidateUpdate( + updateName, updateId, input, eventId, context.headerWithCodec(header)); } catch (Throwable e) { applyWorkflowFailurePolicyAndRethrow(e, outcomeDataConverter); } @@ -120,7 +120,12 @@ public Optional handleExecuteUpdate( DataConverter outcomeDataConverter) { try { return context.handleExecuteUpdate( - updateName, updateId, input, eventId, new Header(header), outcomeDataConverter); + updateName, + updateId, + input, + eventId, + context.headerWithCodec(header), + outcomeDataConverter); } catch (UnsupportedContinueAsNewRequest e) { // Re-throw to fail the workflow task throw e; diff --git a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java index d962d38770..7b57aa5197 100644 --- a/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java @@ -7,7 +7,6 @@ import io.nexusrpc.handler.HandlerException.RetryBehavior; import io.nexusrpc.handler.OperationContext; import io.nexusrpc.handler.OperationStartDetails; -import io.temporal.api.common.v1.Payload; import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.client.ActivityClient; import io.temporal.client.ActivityClientOptions; @@ -21,12 +20,11 @@ import io.temporal.client.WorkflowUpdateHandle; import io.temporal.client.WorkflowUpdateStage; import io.temporal.common.Experimental; -import io.temporal.common.context.ContextPropagator; import io.temporal.common.interceptors.ActivityClientCallsInterceptor; -import io.temporal.common.interceptors.Header; import io.temporal.internal.client.ActivityClientInternal; import io.temporal.internal.client.NexusStartActivityResponse; import io.temporal.internal.client.NexusStartWorkflowResponse; +import io.temporal.internal.common.HeaderUtils; import io.temporal.internal.nexus.CurrentNexusOperationContext; import io.temporal.internal.nexus.InternalNexusOperationContext; import io.temporal.internal.nexus.NexusOperationMetadata; @@ -40,9 +38,7 @@ import java.lang.reflect.Type; import java.util.Arrays; import java.util.Collections; -import java.util.HashMap; import java.util.List; -import java.util.Map; import java.util.Objects; import java.util.concurrent.atomic.AtomicBoolean; @@ -1106,7 +1102,7 @@ private TemporalOperationResult startActivityImpl( activityType, args, options, - propagatedHeader(), + HeaderUtils.fromContextPropagators(client.getOptions().getContextPropagators()), request -> { ActivityClientCallsInterceptor.StartActivityInput input = new ActivityClientCallsInterceptor.StartActivityInput( @@ -1167,16 +1163,4 @@ private TemporalOperationResult startActivityImpl( nexusContext.setNexusOperationMetadata(null); } } - - private Header propagatedHeader() { - List propagators = client.getOptions().getContextPropagators(); - if (propagators.isEmpty()) { - return Header.empty(); - } - Map result = new HashMap<>(); - for (ContextPropagator propagator : propagators) { - result.putAll(propagator.serializeContext(propagator.getCurrentContext())); - } - return new Header(result); - } } diff --git a/temporal-sdk/src/main/java/io/temporal/payload/codec/PayloadCodecProvider.java b/temporal-sdk/src/main/java/io/temporal/payload/codec/PayloadCodecProvider.java new file mode 100644 index 0000000000..7111b61f9e --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/payload/codec/PayloadCodecProvider.java @@ -0,0 +1,16 @@ +package io.temporal.payload.codec; + +import javax.annotation.Nullable; + +/** + * Exposes the codec-only stage of value conversion without enabling automatic header processing. + * + *

The stage handles already serialized payloads, not application values, and must preserve one + * payload per header entry when used by opted-in tracing or context propagators. It must support + * serialization contexts. + */ +public interface PayloadCodecProvider { + /** Returns the codec-only stage, or {@code null} if no codec is available. */ + @Nullable + PayloadCodec getPayloadCodec(); +} diff --git a/temporal-sdk/src/test/java/io/temporal/functional/serialization/WorkflowIdSignedPayloadsTest.java b/temporal-sdk/src/test/java/io/temporal/functional/serialization/WorkflowIdSignedPayloadsTest.java index c5068d8f87..903e3df764 100644 --- a/temporal-sdk/src/test/java/io/temporal/functional/serialization/WorkflowIdSignedPayloadsTest.java +++ b/temporal-sdk/src/test/java/io/temporal/functional/serialization/WorkflowIdSignedPayloadsTest.java @@ -11,6 +11,7 @@ import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.api.enums.v1.EventType; import io.temporal.client.*; +import io.temporal.common.context.ContextPropagator; import io.temporal.common.converter.*; import io.temporal.failure.CanceledFailure; import io.temporal.payload.codec.PayloadCodec; @@ -67,7 +68,10 @@ public class WorkflowIdSignedPayloadsTest { DynamicWorkflowImpl.class, WorkflowWithQuery.class) .setWorkflowClientOptions( - WorkflowClientOptions.newBuilder().setDataConverter(codecDataConverter).build()) + WorkflowClientOptions.newBuilder() + .setDataConverter(codecDataConverter) + .setContextPropagators(Collections.singletonList(new HeaderContext())) + .build()) .setActivityImplementations( heartbeatingActivity, manualCompletionActivity, new DynamicActivityImpl()) .build(); @@ -79,6 +83,18 @@ public void testSimpleWorkflowWithAnActivity() { TestWorkflows.TestWorkflow1 workflowStub = testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class); assertEquals("result", workflowStub.execute("input")); + assertEquals( + "signed", + testWorkflowRule + .getHistoryEvents( + WorkflowStub.fromTyped(workflowStub).getExecution().getWorkflowId(), + EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED) + .get(0) + .getWorkflowExecutionStartedEventAttributes() + .getHeader() + .getFieldsOrThrow("header-context") + .getMetadataOrThrow(EncodingKeys.METADATA_ENCODING_KEY) + .toStringUtf8()); } @Test(timeout = 30_000) @@ -307,6 +323,45 @@ public Object execute(EncodedValues args) { } } + private static class HeaderContext implements ContextPropagator { + private static final Payload VALUE = + DefaultDataConverter.STANDARD_INSTANCE.toPayload("context").get(); + + @Override + public boolean encodeHeaders() { + return true; + } + + @Override + public String getName() { + return "header-context"; + } + + @Override + public Map serializeContext(Object context) { + return Collections.singletonMap("header-context", VALUE); + } + + @Override + public Object deserializeContext(Map header) { + // Cron runs started by the server need not carry the original headers. + if (header.containsKey("header-context")) { + assertEquals(VALUE, header.get("header-context")); + } + return "context"; + } + + @Override + public Object getCurrentContext() { + return "context"; + } + + @Override + public void setCurrentContext(Object context) { + assertEquals("context", context); + } + } + private static class PayloadEncoderWithWorkflowIdSignature implements PayloadCodec { private final ByteString METADATA_ENCODING = ByteString.copyFromUtf8("signed"); diff --git a/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityTaskExecutorsTest.java b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityTaskExecutorsTest.java new file mode 100644 index 0000000000..f3319f36dc --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityTaskExecutorsTest.java @@ -0,0 +1,59 @@ +package io.temporal.internal.activity; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.uber.m3.tally.NoopScope; +import io.temporal.activity.DynamicActivity; +import io.temporal.api.common.v1.Header; +import io.temporal.common.context.ContextPropagator; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.CodecDataConverterTest.PrefixPayloadCodec; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.interceptors.WorkerInterceptor; +import java.util.Collections; +import java.util.Optional; +import org.junit.Test; + +public class ActivityTaskExecutorsTest { + @Test + public void contextPropagationPreservesHeaderPresenceWithAndWithoutCodecs() { + for (boolean enabled : new boolean[] {false, true}) { + for (boolean present : new boolean[] {false, true}) { + ActivityInfoInternal info = mock(ActivityInfoInternal.class); + when(info.getNamespace()).thenReturn("namespace"); + when(info.getInput()).thenReturn(Optional.empty()); + when(info.getHeader()) + .thenReturn(present ? Optional.of(Header.getDefaultInstance()) : Optional.empty()); + InternalActivityExecutionContext context = mock(InternalActivityExecutionContext.class); + when(context.getInfo()).thenReturn(info); + ActivityExecutionContextFactory factory = mock(ActivityExecutionContextFactory.class); + when(factory.createContext(eq(info), any(), any())).thenReturn(context); + ContextPropagator propagator = mock(ContextPropagator.class); + when(propagator.encodeHeaders()).thenReturn(enabled); + when(propagator.deserializeContext(any())).thenReturn("context"); + DynamicActivity activity = mock(DynamicActivity.class); + CodecDataConverter converter = + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, + Collections.singletonList(new PrefixPayloadCodec())); + + new ActivityTaskExecutors.DynamicActivityImplementation( + activity, + converter, + Collections.singletonList(propagator), + new WorkerInterceptor[0], + factory) + .execute(info, new NoopScope()); + + verify(activity).execute(any()); + verify(propagator, times(present ? 1 : 0)).deserializeContext(Collections.emptyMap()); + verify(propagator, times(present ? 1 : 0)).setCurrentContext("context"); + } + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/HeaderCodecClientTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/HeaderCodecClientTest.java new file mode 100644 index 0000000000..25e9cd3bf4 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/HeaderCodecClientTest.java @@ -0,0 +1,129 @@ +package io.temporal.internal.client; + +import static org.junit.Assert.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.enums.v1.UpdateWorkflowExecutionLifecycleStage; +import io.temporal.api.update.v1.UpdateRef; +import io.temporal.api.update.v1.WaitPolicy; +import io.temporal.api.workflowservice.v1.ExecuteMultiOperationRequest; +import io.temporal.api.workflowservice.v1.ExecuteMultiOperationResponse; +import io.temporal.api.workflowservice.v1.StartWorkflowExecutionResponse; +import io.temporal.api.workflowservice.v1.UpdateWorkflowExecutionResponse; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.WorkflowOptions; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.CodecDataConverterTest.PrefixPayloadCodec; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.interceptors.Header; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.StartUpdateInput; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowStartInput; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowUpdateWithStartInput; +import io.temporal.internal.client.external.GenericWorkflowClient; +import io.temporal.payload.codec.PayloadCodec; +import java.util.Collections; +import java.util.Optional; +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +public class HeaderCodecClientTest { + @Test + public void multioperationEncodesReusedHeaderOncePerOperation() { + for (boolean enabled : new boolean[] {false, true}) { + PayloadCodec codec = mock(PayloadCodec.class); + PrefixPayloadCodec encoding = new PrefixPayloadCodec(); + when(codec.encode(any())).thenAnswer(call -> encoding.encode(call.getArgument(0))); + when(codec.withContext(any())).thenReturn(codec); + CodecDataConverter converter = + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, Collections.singletonList(codec)); + WorkflowClientOptions options = + WorkflowClientOptions.newBuilder() + .setDataConverter(converter) + .validateAndBuildWithDefaults(); + GenericWorkflowClient rpc = mock(GenericWorkflowClient.class); + WorkflowExecution execution = WorkflowExecution.newBuilder().setWorkflowId("id").build(); + UpdateWorkflowExecutionLifecycleStage accepted = + UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED; + when(rpc.executeMultiOperation(any(), any())) + .thenReturn( + ExecuteMultiOperationResponse.newBuilder() + .addResponses( + ExecuteMultiOperationResponse.Response.newBuilder() + .setStartWorkflow( + StartWorkflowExecutionResponse.newBuilder().setRunId("run"))) + .addResponses( + ExecuteMultiOperationResponse.Response.newBuilder() + .setUpdateWorkflow( + UpdateWorkflowExecutionResponse.newBuilder() + .setStage(accepted) + .setUpdateRef( + UpdateRef.newBuilder() + .setWorkflowExecution(execution) + .setUpdateId("u")))) + .build()); + RootWorkflowClientInvoker invoker = + new RootWorkflowClientInvoker( + rpc, + options.getNamespace(), + options.getIdentity(), + options.getQueryRejectCondition(), + options.getContextPropagators(), + options.getDataConverter(), + new WorkerFactoryRegistry(), + null); + Payload payload = DefaultDataConverter.STANDARD_INSTANCE.toPayload("trace").get(); + Header header = Header.empty(); + header.putAll(Collections.singletonMap("_tracer-data", payload), enabled); + invoker.updateWithStart( + new WorkflowUpdateWithStartInput<>( + new WorkflowStartInput( + "id", + "type", + header, + new Object[0], + WorkflowOptions.newBuilder().setTaskQueue("q").build()), + new StartUpdateInput<>( + execution, + Optional.of("type"), + "update", + header, + "u", + new Object[0], + String.class, + String.class, + "", + WaitPolicy.newBuilder().setLifecycleStage(accepted).build()))); + + ArgumentCaptor sent = + ArgumentCaptor.forClass(ExecuteMultiOperationRequest.class); + verify(rpc).executeMultiOperation(sent.capture(), any()); + Payload expected = + enabled ? encoding.encode(Collections.singletonList(payload)).get(0) : payload; + assertEquals( + expected, + sent.getValue() + .getOperations(0) + .getStartWorkflow() + .getHeader() + .getFieldsOrThrow("_tracer-data")); + assertEquals( + expected, + sent.getValue() + .getOperations(1) + .getUpdateWorkflow() + .getRequest() + .getInput() + .getHeader() + .getFieldsOrThrow("_tracer-data")); + assertEquals(payload, header.getValues().get("_tracer-data")); + verify(codec, times(enabled ? 2 : 0)).encode(any()); + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerStorageTargetTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerStorageTargetTest.java index 5b07d020f4..92a9beac39 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerStorageTargetTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerStorageTargetTest.java @@ -20,6 +20,10 @@ import io.temporal.api.workflowservice.v1.StartWorkflowExecutionResponse; import io.temporal.client.WorkflowClientOptions; import io.temporal.client.WorkflowOptions; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.CodecDataConverterTest.PrefixPayloadCodec; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; import io.temporal.common.interceptors.Header; import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.QueryInput; import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalInput; @@ -207,6 +211,29 @@ public void signalWithStartOffloadsHeaderPayloads() { sent.getValue().getHeader().getFieldsOrThrow("trace")); } + @Test + public void headerCodecRunsBeforeExternalStorage() { + CapturingDriver driver = new CapturingDriver(); + GenericWorkflowClient rpc = mock(GenericWorkflowClient.class); + PrefixPayloadCodec codec = new PrefixPayloadCodec(); + DataConverter converter = + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, Collections.singletonList(codec), false); + Header header = Header.empty(); + header.putAll(Collections.singletonMap("trace", tracePayload("context")), true); + invoker(rpc, driver, converter) + .signal( + new WorkflowSignalInput( + WorkflowExecution.newBuilder().setWorkflowId("wf").build(), + "signal", + header, + new Object[0])); + assertEquals( + Collections.singletonList( + codec.encode(Collections.singletonList(header.getValues().get("trace"))).get(0)), + driver.stored); + } + private static Header headerWith(String key, String value) { return new Header(Collections.singletonMap(key, tracePayload(value))); } @@ -221,8 +248,16 @@ private static Payload tracePayload(String value) { private static RootWorkflowClientInvoker invoker( GenericWorkflowClient rpc, StorageDriver driver) { + return invoker(rpc, driver, DefaultDataConverter.STANDARD_INSTANCE); + } + + private static RootWorkflowClientInvoker invoker( + GenericWorkflowClient rpc, StorageDriver driver, DataConverter converter) { WorkflowClientOptions options = - WorkflowClientOptions.newBuilder().setNamespace(NAMESPACE).validateAndBuildWithDefaults(); + WorkflowClientOptions.newBuilder() + .setNamespace(NAMESPACE) + .setDataConverter(converter) + .validateAndBuildWithDefaults(); return new RootWorkflowClientInvoker( rpc, options.getNamespace(), diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/ScheduleProtoUtilTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/ScheduleProtoUtilTest.java index 3325a8a2fe..fb6ea9170e 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/ScheduleProtoUtilTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/ScheduleProtoUtilTest.java @@ -5,9 +5,12 @@ import io.temporal.client.schedules.ScheduleActionStartWorkflow; import io.temporal.client.schedules.ScheduleClientOptions; import io.temporal.client.schedules.SchedulePolicy; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.CodecDataConverterTest.PrefixPayloadCodec; import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.DefaultDataConverter; import io.temporal.common.converter.TransferTypeTestModel; +import io.temporal.common.interceptors.Header; import java.time.Duration; import java.util.Collections; import org.junit.Assert; @@ -32,6 +35,48 @@ public void policyToProtoIncludesExplicitCatchupWindow() { .getSeconds()); } + @Test + public void headerCodecRoundTripPreservesScheduleUpdates() { + for (boolean enabled : new boolean[] {false, true}) { + CodecDataConverter codecConverter = + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, + Collections.singletonList(new PrefixPayloadCodec()), + false); + DataConverter converter = codecConverter; + ScheduleProtoUtil util = + new ScheduleProtoUtil(null, "namespace", Collections.emptyList(), converter); + Header header = Header.empty(); + header.putAll( + Collections.singletonMap( + "_tracer-data", DefaultDataConverter.STANDARD_INSTANCE.toPayload("trace").get()), + enabled); + ScheduleActionStartWorkflow action = + ScheduleActionStartWorkflow.newBuilder() + .setWorkflowType("workflow") + .setOptions( + WorkflowOptions.newBuilder().setWorkflowId("id").setTaskQueue("q").build()) + .setHeader(header) + .build(); + ScheduleAction proto = util.actionToProto(action); + Assert.assertEquals( + enabled, + proto + .getStartWorkflow() + .getHeader() + .getFieldsOrThrow("_tracer-data") + .getData() + .startsWith(PrefixPayloadCodec.PREFIX)); + ScheduleActionStartWorkflow restored = + (ScheduleActionStartWorkflow) util.protoToAction(proto); + Assert.assertEquals( + proto.getStartWorkflow().getHeader().getFieldsMap(), restored.getHeader().getValues()); + Assert.assertEquals( + proto.getStartWorkflow().getHeader(), + util.actionToProto(restored).getStartWorkflow().getHeader()); + } + } + @Test public void actionUsesInternalConverterWhileOptionsKeepConfiguredConverter() { DataConverter configured = DefaultDataConverter.newDefaultInstance(); diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/WorkflowClientRequestFactoryTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/WorkflowClientRequestFactoryTest.java index 7ab084cb63..8b497fbdc2 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/WorkflowClientRequestFactoryTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/WorkflowClientRequestFactoryTest.java @@ -25,7 +25,7 @@ public class WorkflowClientRequestFactoryTest { public void requestsUseExplicitNamespaceAndIdentity() { WorkflowClientRequestFactory factory = new WorkflowClientRequestFactory( - "test-namespace", "test-identity", Collections.emptyList()); + "test-namespace", "test-identity", Collections.emptyList(), null); StartWorkflowExecutionRequest start = start(factory, null); assertEquals("test-namespace", start.getNamespace()); assertEquals("test-identity", start.getIdentity()); @@ -45,7 +45,7 @@ public void requestsUseExplicitNamespaceAndIdentity() { public void clientContextPropagatorListRetainsReferenceSemantics() { List propagators = new ArrayList<>(); WorkflowClientRequestFactory factory = - new WorkflowClientRequestFactory("test-namespace", "test-identity", propagators); + new WorkflowClientRequestFactory("test-namespace", "test-identity", propagators, null); assertFalse(start(factory, null).getHeader().containsFields("trace")); propagators.add(propagator("client")); @@ -57,7 +57,10 @@ public void clientContextPropagatorListRetainsReferenceSemantics() { public void workflowContextPropagatorsOverrideClientDefaultsIncludingEmptyList() { WorkflowClientRequestFactory factory = new WorkflowClientRequestFactory( - "test-namespace", "test-identity", Collections.singletonList(propagator("client"))); + "test-namespace", + "test-identity", + Collections.singletonList(propagator("client")), + null); assertEquals( payload("workflow"), start(factory, Collections.singletonList(propagator("workflow"))) diff --git a/temporal-sdk/src/test/java/io/temporal/internal/common/HeaderUtilsTest.java b/temporal-sdk/src/test/java/io/temporal/internal/common/HeaderUtilsTest.java new file mode 100644 index 0000000000..92d03690c0 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/common/HeaderUtilsTest.java @@ -0,0 +1,181 @@ +package io.temporal.internal.common; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import com.google.protobuf.ByteString; +import io.temporal.api.common.v1.Payload; +import io.temporal.common.context.ContextPropagator; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.CodecDataConverterTest.PrefixPayloadCodec; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.interceptors.Header; +import io.temporal.internal.client.ClientDataConverterFactory; +import io.temporal.internal.payload.storage.ExternalStorageDataConverter; +import io.temporal.payload.codec.ChainCodec; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.codec.PayloadCodecException; +import io.temporal.payload.codec.ZlibPayloadCodec; +import io.temporal.payload.context.WorkflowSerializationContext; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import javax.annotation.Nonnull; +import org.junit.Test; + +public class HeaderUtilsTest { + @Test + public void producerOptInPreservesWireReadyHeadersAndPrivateViews() { + Payload serialized = + DefaultDataConverter.STANDARD_INSTANCE + .toPayload(String.join("", Collections.nCopies(100, "context"))) + .get(); + PayloadCodec codec = new ZlibPayloadCodec(); + Payload encoded = codec.encode(Collections.singletonList(serialized)).get(0); + assertEquals(ByteString.copyFromUtf8("binary/zlib"), encoded.getMetadataOrThrow("encoding")); + ContextPropagator optedIn = mock(ContextPropagator.class); + when(optedIn.encodeHeaders()).thenReturn(true); + when(optedIn.serializeContext(any())) + .thenReturn(Collections.singletonMap("context", serialized)); + ContextPropagator legacy = mock(ContextPropagator.class); + when(legacy.serializeContext(any())).thenReturn(Collections.singletonMap("legacy", encoded)); + Header header = HeaderUtils.fromContextPropagators(Arrays.asList(optedIn, legacy)); + header.getValues().put("raw", encoded); + header.getValues().put("plaintext", serialized); + io.temporal.api.common.v1.Header wire = HeaderUtils.toHeaderGrpc(header, null, codec); + assertEquals(encoded, wire.getFieldsOrThrow("context")); + assertEquals(encoded, wire.getFieldsOrThrow("legacy")); + assertEquals(encoded, wire.getFieldsOrThrow("raw")); + assertEquals(serialized, wire.getFieldsOrThrow("plaintext")); + Header received = new Header(wire, codec); + Map view = received.getCodecDecodedValues(); + assertEquals(serialized, view.get("context")); + assertEquals(serialized, view.get("plaintext")); + view.clear(); + assertEquals(serialized, received.getCodecDecodedValues().get("context")); + assertEquals(wire.getFieldsMap(), received.getValues()); + assertEquals(serialized, header.getValues().get("context")); + + header.getValues().clear(); + header.getValues().put("context", serialized); + assertEquals( + encoded, HeaderUtils.toHeaderGrpc(header, null, codec).getFieldsOrThrow("context")); + header.putAll(Collections.singletonMap("context", serialized), false); + assertEquals( + serialized, HeaderUtils.toHeaderGrpc(header, null, codec).getFieldsOrThrow("context")); + header.putAll(Collections.singletonMap("context", serialized), true); + header.getValues().put("context", encoded); + assertEquals( + encoded, HeaderUtils.toHeaderGrpc(header, null, codec).getFieldsOrThrow("context")); + } + + @Test + public void codecsProcessSerializedHeadersOnceWithoutMutatingSources() { + Map values = new HashMap<>(); + Payload serialized = DefaultDataConverter.STANDARD_INSTANCE.toPayload("context").get(); + values.put("_tracer-data", serialized); + values.put("__temporal_custom", serialized); + Header header = Header.empty(); + header.putAll(values, true); + Header overrides = Header.empty(); + overrides.putAll(Collections.singletonMap("_tracer-data", serialized), true); + CountingCodec codec = new CountingCodec(); + DataConverter valuesConverter = mock(DataConverter.class); + CodecDataConverter configured = + new CodecDataConverter(valuesConverter, Collections.singletonList(codec)); + DataConverter enabled = + new ExternalStorageDataConverter(ClientDataConverterFactory.forClient(configured), null) + .withContext(new WorkflowSerializationContext("namespace", "workflow")); + PayloadCodec enabledCodec = HeaderUtils.resolveHeaderCodec(enabled); + + io.temporal.api.common.v1.Header plain = HeaderUtils.toHeaderGrpc(header, overrides); + assertEquals(plain, HeaderUtils.toHeaderGrpc(header, overrides, null)); + assertSame(plain, HeaderUtils.decode(plain, null)); + for (int i = 0; i < 2; i++) { + io.temporal.api.common.v1.Header encoded = + HeaderUtils.toHeaderGrpc(header, overrides, enabledCodec); + assertEquals(plain, HeaderUtils.decode(encoded, enabledCodec)); + } + assertEquals(4, codec.encoded); + assertEquals(4, codec.decoded); + assertEquals(serialized, header.getValues().get("_tracer-data")); + assertEquals(serialized, overrides.getValues().get("_tracer-data")); + verifyNoInteractions(valuesConverter); + } + + @Test(expected = PayloadCodecException.class) + public void codecFailuresAreNotRetriedAsPlaintext() { + PayloadCodec codec = mock(PayloadCodec.class); + when(codec.decode(any())).thenThrow(new PayloadCodecException("invalid encoded header")); + HeaderUtils.decode( + io.temporal.api.common.v1.Header.newBuilder() + .putFields("key", Payload.getDefaultInstance()) + .build(), + codec); + } + + @Test + public void contextualizationIsOptInAndRejectsNull() { + PayloadCodec codec = mock(PayloadCodec.class); + WorkflowSerializationContext context = + new WorkflowSerializationContext("namespace", "workflow"); + Header legacy = new Header(Collections.singletonMap("raw", Payload.getDefaultInstance())); + assertEquals( + HeaderUtils.toHeaderGrpc(legacy, null), + HeaderUtils.toHeaderGrpc(legacy, null, HeaderUtils.withContext(codec, context))); + verifyNoInteractions(codec); + assertThrows( + IllegalArgumentException.class, + () -> + HeaderUtils.encode( + HeaderUtils.toHeaderGrpc(legacy, null), HeaderUtils.withContext(codec, context))); + when(codec.withContext(any())).thenReturn(new ChainCodec(Collections.emptyList())); + assertEquals( + HeaderUtils.toHeaderGrpc(legacy, null), + HeaderUtils.encode( + HeaderUtils.toHeaderGrpc(legacy, null), HeaderUtils.withContext(codec, context))); + } + + @Test(expected = PayloadCodecException.class) + public void rejectsCodecsThatChangeHeaderCardinality() { + Header header = Header.empty(); + header.putAll(Collections.singletonMap("key", Payload.getDefaultInstance()), true); + HeaderUtils.toHeaderGrpc( + header, + null, + new CountingCodec() { + @Override + public List encode(@Nonnull List payloads) { + return Collections.emptyList(); + } + }); + } + + private static class CountingCodec implements PayloadCodec { + private final PayloadCodec delegate = new PrefixPayloadCodec(); + private int encoded; + private int decoded; + + @Override + @Nonnull + public List encode(@Nonnull List payloads) { + encoded++; + return delegate.encode(payloads); + } + + @Override + @Nonnull + public List decode(@Nonnull List payloads) { + decoded++; + return delegate.decode(payloads); + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/sync/SyncWorkflowContextTest.java b/temporal-sdk/src/test/java/io/temporal/internal/sync/SyncWorkflowContextTest.java index 10a7627c0e..ff85c71d14 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/sync/SyncWorkflowContextTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/sync/SyncWorkflowContextTest.java @@ -9,19 +9,28 @@ import static org.mockito.Mockito.when; import io.temporal.api.command.v1.ContinueAsNewWorkflowExecutionCommandAttributes; +import io.temporal.api.common.v1.Payload; import io.temporal.api.common.v1.SearchAttributes; +import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.api.common.v1.WorkflowType; import io.temporal.api.sdk.v1.UserMetadata; +import io.temporal.common.converter.CodecDataConverter; import io.temporal.common.converter.DefaultDataConverter; import io.temporal.common.interceptors.Header; +import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor.ChildWorkflowInput; import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor.ContinueAsNewInput; import io.temporal.internal.common.SdkFlag; import io.temporal.internal.common.SearchAttributesUtil; import io.temporal.internal.logging.PrefixedMdc; import io.temporal.internal.replay.ReplayWorkflowContext; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.context.SerializationContext; +import io.temporal.payload.context.WorkflowSerializationContext; +import io.temporal.workflow.ChildWorkflowOptions; import io.temporal.workflow.ContinueAsNewOptions; import io.temporal.workflow.TimerOptions; import java.time.Duration; +import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.concurrent.ExecutorService; @@ -42,6 +51,65 @@ public void setUp() { when(mockReplayWorkflowContext.getLoggerMdc()).thenReturn(new PrefixedMdc(null)); } + @Test + @SuppressWarnings("deprecation") + public void childHeaderCodecUsesEffectiveDestinationNamespace() { + when(mockReplayWorkflowContext.getNamespace()).thenReturn("parent"); + when(mockReplayWorkflowContext.getWorkflowType()) + .thenReturn(WorkflowType.newBuilder().setName("Parent").build()); + when(mockReplayWorkflowContext.getTaskQueue()).thenReturn("queue"); + for (String childNamespace : new String[] {null, "", "destination"}) { + PayloadCodec codec = mock(PayloadCodec.class); + when(codec.withContext(any())).thenReturn(codec); + when(codec.encode(any())).thenAnswer(call -> call.getArgument(0)); + SyncWorkflowContext context = + new SyncWorkflowContext( + "parent", + WorkflowExecution.newBuilder().setWorkflowId("parent-id").build(), + null, + null, + null, + null, + null, + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, Collections.singletonList(codec)), + Collections.emptyList()); + context.setReplayContext(mockReplayWorkflowContext); + Header header = Header.empty(); + header.putAll(Collections.singletonMap("key", Payload.getDefaultInstance()), true); + ExecutorService threadPool = Executors.newCachedThreadPool(); + DeterministicRunner runner = + DeterministicRunner.newRunner( + threadPool::submit, + context, + () -> + context.executeChildWorkflow( + new ChildWorkflowInput<>( + "child-id", + "Child", + Void.class, + Void.class, + new Object[0], + ChildWorkflowOptions.newBuilder().setNamespace(childNamespace).build(), + header))); + try { + runner.runUntilAllBlocked(DeterministicRunner.DEFAULT_DEADLOCK_DETECTION_TIMEOUT_MS); + ArgumentCaptor captured = + ArgumentCaptor.forClass(SerializationContext.class); + verify(codec).withContext(captured.capture()); + WorkflowSerializationContext serializationContext = + (WorkflowSerializationContext) captured.getValue(); + assertEquals( + childNamespace == null || childNamespace.isEmpty() ? "parent" : childNamespace, + serializationContext.getNamespace()); + assertEquals("child-id", serializationContext.getWorkflowId()); + } finally { + runner.close(); + threadPool.shutdown(); + } + } + } + @Test public void testUpsertSearchAttributes() { Map attr = new HashMap<>();