diff --git a/CHANGELOG.md b/CHANGELOG.md index d426a29297..cc110c46fe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,12 @@ to docs, or any other relevant information. `WorkflowOutboundCallsInterceptorBase` need to add them, and Base subclasses override them to see calls with options. ### Added +- Wrap a codec-capable data converter in `HeaderEncodingDataConverter` to opt into automatic + codec-only processing of payload headers, including tracing and context propagation, on workflow, + activity, and schedule clients. Workers inherit their workflow client's converter. Ordinary + `CodecDataConverter` retains legacy header behavior. Supply serialized but not already codec-encoded + headers, and ensure decoders support legacy plaintext before enabling this for existing histories + or schedules. Custom outer converter wrappers must forward the header capability. - 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 @@ -69,6 +75,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..38f0ca8f1b 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,39 @@ 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 + +Automatic header codec processing is opt-in: + +```java +DataConverter payloadConverter = DefaultDataConverter.newDefaultInstance(); +DataConverter converter = + new HeaderEncodingDataConverter(new CodecDataConverter(payloadConverter, codecs)); +WorkflowClientOptions options = + WorkflowClientOptions.newBuilder() + .setDataConverter(converter) + .build(); +``` + +The wrapper enables codec-only processing of already serialized payload headers. +Workers inherit it from their workflow client. Independently created activity and schedule clients +must also configure the wrapper and use compatible codecs. +Construct header payloads with `payloadConverter`, not `converter`: do not manually codec them or +serialize a `Payload` again as an application value. Context propagators and inbound interceptors +receive decoded payloads. Tracing retains its standard serialization, but its payload headers +participate in the codec stage. Nexus string transport headers and gRPC metadata are unaffected. + +Ordinary `CodecDataConverter` retains the previous header behavior. It implements +`DataConverterWithCodec`, combining value conversion and `PayloadCodecProvider` stage exposure without +opting into header processing. `HeaderEncodingDataConverter` requires this combined capability and +rejects a null or empty codec stage. Disabled converters' stages are not inspected. +Before enabling the wrapper for existing histories or schedules, ensure your decoder accepts their +legacy plaintext headers; codec failures do not fall back to plaintext. SDK-owned converter wrappers +forward the opt-in `HeaderCodecProvider` capability. Keep `HeaderEncodingDataConverter` outermost, +or make custom wrappers explicitly implement and forward `HeaderCodecProvider`, whose codec must +support serialization contexts. An opaque outer wrapper can hide the opt-in; the SDK does not +discover delegates through reflection or value conversion. + ## Development ### Development Environment 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..55e6950217 --- /dev/null +++ b/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/HeaderCodecTest.java @@ -0,0 +1,78 @@ +package io.temporal.opentracing; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; + +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.HeaderEncodingDataConverter; +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).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( + enabled ? new HeaderEncodingDataConverter(converter) : converter); + Header header = Header.empty(); + accessor.writeSpanContextToHeader(span.context(), header, tracer); + io.temporal.api.common.v1.Header wire = HeaderUtils.toHeaderGrpc(header, null, headerCodec); + Header received = new Header(HeaderUtils.decode(wire, headerCodec)); + + assertEquals( + span.context().toTraceId(), + accessor.readSpanContextFromHeader(received, tracer).toTraceId()); + assertEquals(enabled ? 1 : 0, codec.encoded); + assertEquals(enabled ? 1 : 0, codec.decoded); + assertEquals(header.getValues(), 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().setData(payloads.get(0).toByteString()).build()); + } + + @Override + @Nonnull + public List decode(@Nonnull List 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..0d1115560f 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,15 @@ 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.common.converter.HeaderEncodingDataConverter; 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; @@ -31,6 +35,11 @@ public class SignalWithStartTest { SDKTestWorkflowRule.newBuilder() .setWorkflowClientOptions( WorkflowClientOptions.newBuilder() + .setDataConverter( + new HeaderEncodingDataConverter( + 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..619aa6c5f5 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,17 @@ 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.common.converter.HeaderEncodingDataConverter; 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; @@ -38,6 +43,11 @@ public class SpanContextPropagationTest { SDKTestWorkflowRule.newBuilder() .setWorkflowClientOptions( WorkflowClientOptions.newBuilder() + .setDataConverter( + new HeaderEncodingDataConverter( + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, + Collections.singletonList(new HeaderCodecTest.CountingCodec())))) .setInterceptors(new OpenTracingClientInterceptor(OT_OPTIONS)) .validateAndBuildWithDefaults()) .setWorkerFactoryOptions( @@ -63,6 +73,9 @@ public interface TestActivity { public interface TestWorkflow { @WorkflowMethod String workflow1(String input); + + @QueryMethod + String query(); } public static class OpenTracingAwareActivityImpl implements TestActivity { @@ -96,6 +109,11 @@ public String workflow1(String input) { return activity.activity1(input); } + + @Override + public String query() { + return mockTracer.activeSpan().getBaggageItem(BAGGAGE_ITEM_KEY); + } } /** @@ -121,6 +139,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..dc9982db48 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,9 @@ 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.common.converter.HeaderEncodingDataConverter; import io.temporal.testing.internal.SDKTestWorkflowRule; import io.temporal.worker.WorkerFactoryOptions; import io.temporal.workflow.CompletablePromise; @@ -17,6 +20,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; @@ -38,6 +42,11 @@ public class UpdateWithStartTest { SDKTestWorkflowRule.newBuilder() .setWorkflowClientOptions( WorkflowClientOptions.newBuilder() + .setDataConverter( + new HeaderEncodingDataConverter( + 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/common/context/ContextPropagator.java b/temporal-sdk/src/main/java/io/temporal/common/context/ContextPropagator.java index 16e068e363..d2acef64b0 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 @@ -113,6 +113,9 @@ public interface ContextPropagator { * interceptors, because they all share the same single keyspace of a single instance of {@code * temporal.api.common.v1.Header} * + *

With automatic header codecs enabled, return serialized but not codec-encoded payloads. The + * SDK applies the codec after merging headers and decodes before {@link #deserializeContext}. + * * @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..f40009c495 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 @@ -27,8 +27,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 but does not enable automatic header processing. Wrap + * it in {@link HeaderEncodingDataConverter} to opt in. */ -public class CodecDataConverter implements DataConverter, PayloadCodec { +public class CodecDataConverter implements DataConverterWithCodec, PayloadCodec { private static final String ENCODED_FAILURE_MESSAGE = "Encoded failure"; private final DataConverter dataConverter; @@ -190,6 +193,14 @@ public CodecDataConverter withContext(@Nonnull SerializationContext context) { return new CodecDataConverter(dataConverter, chainCodec, encodeFailureAttributes, context); } + @Nullable + @Override + public PayloadCodec getPayloadCodec() { + return chainCodec.isEmpty() + ? null + : ConverterUtils.withContext(chainCodec, serializationContext); + } + @Nonnull @Override public List encode(@Nonnull List payloads) { diff --git a/temporal-sdk/src/main/java/io/temporal/common/converter/DataConverterWithCodec.java b/temporal-sdk/src/main/java/io/temporal/common/converter/DataConverterWithCodec.java new file mode 100644 index 0000000000..462dac5359 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/common/converter/DataConverterWithCodec.java @@ -0,0 +1,6 @@ +package io.temporal.common.converter; + +import io.temporal.payload.codec.PayloadCodecProvider; + +/** A data converter exposing its codec-only stage, without opting into header processing. */ +public interface DataConverterWithCodec extends DataConverter, PayloadCodecProvider {} diff --git a/temporal-sdk/src/main/java/io/temporal/common/converter/HeaderEncodingDataConverter.java b/temporal-sdk/src/main/java/io/temporal/common/converter/HeaderEncodingDataConverter.java new file mode 100644 index 0000000000..76703762c7 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/common/converter/HeaderEncodingDataConverter.java @@ -0,0 +1,99 @@ +package io.temporal.common.converter; + +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.failure.v1.Failure; +import io.temporal.payload.codec.ChainCodec; +import io.temporal.payload.codec.HeaderCodecProvider; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.context.SerializationContext; +import java.lang.reflect.Type; +import java.util.Objects; +import java.util.Optional; +import javax.annotation.Nonnull; + +/** + * Opts into automatic codec-only processing of payload headers, including tracing and context + * propagation. Configure this as the outermost data converter on workflow, activity, or schedule + * clients. Workers inherit their workflow client's converter. + * + *

Supply serialized, not already codec-encoded, header payloads. Inbound interceptors and + * propagators receive decoded payloads. Nexus string headers and gRPC metadata are unaffected. + * Before enabling this for existing histories or schedules, ensure the decoder accepts legacy + * plaintext headers. Codec failures never fall back to plaintext. + * + *

SDK-owned transfer and storage wrappers forward {@link HeaderCodecProvider}. Custom wrappers + * must forward that capability or keep this wrapper outermost. An opaque outer wrapper can hide the + * opt-in. Normal value and failure conversion are delegated unchanged. + */ +public final class HeaderEncodingDataConverter implements DataConverter, HeaderCodecProvider { + private final DataConverter delegate; + private final PayloadCodec headerCodec; + + /** + * @param delegate value converter with an exposed codec-only stage + * @throws IllegalArgumentException if the codec stage is missing or empty + */ + public HeaderEncodingDataConverter(DataConverterWithCodec delegate) { + this(Objects.requireNonNull(delegate, "delegate"), delegate.getPayloadCodec()); + } + + private HeaderEncodingDataConverter(DataConverter delegate, PayloadCodec codec) { + this.delegate = Objects.requireNonNull(delegate, "delegate"); + if (codec == null) { + throw new IllegalArgumentException("HeaderEncodingDataConverter requires a nonempty codec"); + } + this.headerCodec = codec instanceof ChainCodec ? ((ChainCodec) codec).forHeaders() : codec; + } + + @Override + public PayloadCodec getHeaderCodec() { + return headerCodec; + } + + @Override + public Optional toPayload(T value) { + return delegate.toPayload(value); + } + + @Override + public T fromPayload(Payload payload, Class valueClass, Type valueType) { + return delegate.fromPayload(payload, valueClass, valueType); + } + + @Override + public Optional toPayloads(Object... values) { + return delegate.toPayloads(values); + } + + @Override + public T fromPayloads( + int index, Optional content, Class valueType, Type valueGenericType) { + return delegate.fromPayloads(index, content, valueType, valueGenericType); + } + + @Override + public Object[] fromPayloads( + Optional content, Class[] parameterTypes, Type[] genericParameterTypes) { + return delegate.fromPayloads(content, parameterTypes, genericParameterTypes); + } + + @Nonnull + @Override + public Failure exceptionToFailure(@Nonnull Throwable throwable) { + return delegate.exceptionToFailure(throwable); + } + + @Nonnull + @Override + public RuntimeException failureToException(@Nonnull Failure failure) { + return delegate.failureToException(failure); + } + + @Nonnull + @Override + public HeaderEncodingDataConverter withContext(@Nonnull SerializationContext context) { + return new HeaderEncodingDataConverter( + delegate.withContext(context), headerCodec.withContext(context)); + } +} 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..143df7e872 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 @@ -5,6 +5,14 @@ import java.util.Map; import javax.annotation.Nonnull; +/** + * Already serialized payload headers shared by interceptors and context propagators. + * + *

When automatic header codecs are enabled through {@link + * io.temporal.common.converter.HeaderEncodingDataConverter}, outbound values must not already be + * codec-encoded. Inbound interceptors receive decoded payloads. Use the underlying value converter + * to construct entries, rather than serializing a {@link Payload} as an application value. + */ public class Header { private final Map values; 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 7789bd2716..6fab1beffd 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.serviceclient.CheckedExceptionWrapper; import java.lang.reflect.Method; @@ -39,6 +41,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; @@ -47,8 +50,10 @@ public BaseActivityTaskExecutor( DataConverter dataConverter, List contextPropagators, WorkerInterceptor[] interceptors, - ActivityExecutionContextFactory executionContextFactory) { + ActivityExecutionContextFactory executionContextFactory, + @Nullable PayloadCodec headerCodec) { this.dataConverter = dataConverter; + this.headerCodec = headerCodec; this.contextPropagators = contextPropagators; this.interceptors = interceptors; this.executionContextFactory = executionContextFactory; @@ -71,8 +76,13 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri dataConverter.withContext(serializationContext); try { - info.getHeader() - .ifPresent(value -> deserializeAndPopulateContext(value, contextPropagators)); + io.temporal.api.common.v1.Header decodedHeader = + HeaderUtils.decode( + info.getHeader().orElse(io.temporal.api.common.v1.Header.getDefaultInstance()), + HeaderUtils.withContext(headerCodec, serializationContext)); + if (info.getHeader().isPresent()) { + deserializeAndPopulateContext(decodedHeader, contextPropagators); + } ActivityInboundCallsInterceptor inboundCallsInterceptor = createRootInboundInterceptor(); for (WorkerInterceptor interceptor : interceptors) { @@ -81,9 +91,7 @@ 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())); + Header header = new Header(decodedHeader); ActivityOutput result = inboundCallsInterceptor.execute( new ActivityInboundCallsInterceptor.ActivityInput(header, args)); @@ -193,8 +201,9 @@ static class POJOActivityImplementation extends BaseActivityTaskExecutor { DataConverter dataConverter, List contextPropagators, WorkerInterceptor[] interceptors, - ActivityExecutionContextFactory executionContextFactory) { - super(dataConverter, contextPropagators, interceptors, executionContextFactory); + ActivityExecutionContextFactory executionContextFactory, + @Nullable PayloadCodec headerCodec) { + super(dataConverter, contextPropagators, interceptors, executionContextFactory, headerCodec); this.method = interfaceMethod; this.activity = activity; } @@ -237,8 +246,9 @@ static class DynamicActivityImplementation extends BaseActivityTaskExecutor { DataConverter dataConverter, List contextPropagators, WorkerInterceptor[] interceptors, - ActivityExecutionContextFactory executionContextFactory) { - super(dataConverter, contextPropagators, interceptors, executionContextFactory); + ActivityExecutionContextFactory executionContextFactory, + @Nullable PayloadCodec headerCodec) { + super(dataConverter, contextPropagators, interceptors, executionContextFactory, headerCodec); this.activity = activity; } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskHandlerImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskHandlerImpl.java index 5b2b2bf90a..0bffca3e46 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskHandlerImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskHandlerImpl.java @@ -20,6 +20,7 @@ import io.temporal.internal.common.env.ReflectionUtils; import io.temporal.internal.worker.ActivityTask; import io.temporal.internal.worker.ActivityTaskHandler; +import io.temporal.payload.codec.PayloadCodec; import io.temporal.serviceclient.MetricsTag; import io.temporal.worker.MetricsType; import io.temporal.worker.TypeAlreadyRegisteredException; @@ -29,6 +30,7 @@ import javax.annotation.Nullable; public final class ActivityTaskHandlerImpl implements ActivityTaskHandler { + private final @Nullable PayloadCodec headerCodec; public static final ImmutableSet ACTIVITY_HANDLER_STACKTRACE_CUTOFF = ImmutableSet.builder() // POJO @@ -64,13 +66,15 @@ public ActivityTaskHandlerImpl( @Nonnull DataConverter dataConverter, @Nonnull ActivityExecutionContextFactory executionContextFactory, @Nonnull WorkerInterceptor[] interceptors, - @Nullable List contextPropagators) { + @Nullable List contextPropagators, + @Nullable PayloadCodec headerCodec) { this.namespace = Objects.requireNonNull(namespace); this.taskQueue = Objects.requireNonNull(taskQueue); this.dataConverter = Objects.requireNonNull(dataConverter); this.executionContextFactory = Objects.requireNonNull(executionContextFactory); this.interceptors = Objects.requireNonNull(interceptors); this.contextPropagators = contextPropagators; + this.headerCodec = headerCodec; } @Override @@ -147,7 +151,8 @@ private void registerActivityImplementation(Object activity) { dataConverter, contextPropagators, interceptors, - executionContextFactory); + executionContextFactory, + headerCodec); } else { Class cls = activity.getClass(); POJOActivityImplMetadata activityImplMetadata = POJOActivityImplMetadata.newInstance(cls); @@ -166,7 +171,8 @@ private void registerActivityImplementation(Object activity) { dataConverter, contextPropagators, interceptors, - executionContextFactory); + executionContextFactory, + headerCodec); activities.put(typeName, implementation); } } 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 50a65dc07c..74a9e729b6 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 @@ -32,6 +32,7 @@ import io.temporal.internal.nexus.CurrentNexusOperationContext; import io.temporal.internal.nexus.InternalNexusOperationContext; import io.temporal.internal.nexus.NexusOperationMetadata; +import io.temporal.payload.codec.PayloadCodec; import io.temporal.payload.context.ActivitySerializationContext; import io.temporal.serviceclient.StatusUtils; import java.lang.reflect.Type; @@ -40,6 +41,7 @@ import java.util.concurrent.CompletionException; import java.util.concurrent.TimeoutException; import java.util.stream.StreamSupport; +import javax.annotation.Nullable; /** * Terminus of the activity interceptor chain. Implements all activity RPCs against the Temporal @@ -51,6 +53,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, @@ -61,6 +64,7 @@ public RootActivityClientInvoker( this.namespace = namespace; this.identity = identity; this.dataConverter = dataConverter; + this.headerCodec = HeaderUtils.resolveHeaderCodec(dataConverter); } @Override @@ -69,10 +73,10 @@ public StartActivityOutput startActivity(StartActivityInput input) { if (Strings.isNullOrEmpty(options.getTaskQueue())) { throw new IllegalArgumentException("taskQueue must not be null or empty"); } - 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; @@ -135,7 +139,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 ad56de7812..8692402476 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 @@ -35,6 +35,8 @@ import io.temporal.internal.nexus.OperationTokenUtil; 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; @@ -60,25 +62,7 @@ public class RootWorkflowClientInvoker implements WorkflowClientCallsInterceptor private final WorkflowClientRequestFactory requestsHelper; private final ClientDataConverterFactory converterFactory; private final @Nullable ExternalStorageRunner externalStorage; - - public RootWorkflowClientInvoker( - GenericWorkflowClient genericClient, - String namespace, - String identity, - QueryRejectCondition queryRejectCondition, - List contextPropagators, - DataConverter dataConverter, - WorkerFactoryRegistry workerFactoryRegistry) { - this( - genericClient, - namespace, - identity, - queryRejectCondition, - contextPropagators, - dataConverter, - workerFactoryRegistry, - null); - } + private final @Nullable PayloadCodec headerCodec; public RootWorkflowClientInvoker( GenericWorkflowClient genericClient, @@ -92,6 +76,7 @@ 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; @@ -119,6 +104,12 @@ private void storeHeader( String workflowId, @Nullable String runId, @Nullable String workflowType) { + Header encoded = + HeaderUtils.encode( + header.build(), + HeaderUtils.withContext( + headerCodec, new WorkflowSerializationContext(namespace, workflowId))); + header.clearFields().putAllFields(encoded.getFieldsMap()); if (externalStorage == null || header.getFieldsCount() == 0) { return; } @@ -372,10 +363,7 @@ public WorkflowUpdateWithStartOutput updateWithStart( WorkflowExecution execution = WorkflowExecution.newBuilder() .setRunId(startResponse.getRunId()) - .setWorkflowId( - toStartRequest(dataConverterWithWorkflowContext, startInput) - .build() - .getWorkflowId()) + .setWorkflowId(startInput.getWorkflowId()) .build(); return new WorkflowUpdateWithStartOutput<>(new WorkflowStartOutput(execution), updateHandle); } 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..9c6f0ecbd6 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,6 +54,7 @@ public ScheduleProtoUtil( this.namespace = namespace; this.contextPropagators = contextPropagators; this.dataConverter = dataConverter; + this.headerCodec = HeaderUtils.resolveHeaderCodec(dataConverter); } private io.temporal.common.interceptors.Header extractContextsAndConvertToBytes( @@ -157,7 +161,10 @@ public ScheduleAction actionToProto(io.temporal.client.schedules.ScheduleAction Header grpcHeader = toHeaderGrpc( startWorkflowAction.getHeader(), - extractContextsAndConvertToBytes(wfOptions.getContextPropagators())); + extractContextsAndConvertToBytes(wfOptions.getContextPropagators()), + HeaderUtils.withContext( + headerCodec, + new WorkflowSerializationContext(namespace, wfOptions.getWorkflowId()))); workflowRequest.setHeader(grpcHeader); if (wfOptions.getPriority() != null) { @@ -417,6 +424,14 @@ 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( + HeaderUtils.decode( + startWfAction.getHeader(), + HeaderUtils.withContext( + headerCodec, + new WorkflowSerializationContext( + namespace, startWfAction.getWorkflowId()))))); builder.setRawArguments( new EncodedValues( 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..58309ac642 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 @@ -4,7 +4,14 @@ import io.temporal.api.common.v1.Payload; import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.DataConverterException; +import io.temporal.payload.codec.ChainCodec; +import io.temporal.payload.codec.HeaderCodecProvider; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.codec.PayloadCodecException; +import io.temporal.payload.context.SerializationContext; +import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import javax.annotation.Nullable; @@ -22,6 +29,64 @@ public static Header toHeaderGrpc( return builder.build(); } + public static Header toHeaderGrpc( + io.temporal.common.interceptors.Header header, + @Nullable io.temporal.common.interceptors.Header overrides, + @Nullable PayloadCodec codec) { + return encode(toHeaderGrpc(header, overrides), codec); + } + + 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 getHeaderCodec(DataConverter converter) { + return converter instanceof HeaderCodecProvider + ? ((HeaderCodecProvider) converter).getHeaderCodec() + : null; + } + + @Nullable + public static PayloadCodec resolveHeaderCodec(DataConverter converter) { + PayloadCodec codec = getHeaderCodec(converter); + return codec == null ? null : requireHeaderCodec(codec); + } + + private static PayloadCodec requireHeaderCodec(@Nullable PayloadCodec codec) { + if (codec == null) { + throw new IllegalArgumentException("Header codec binding must not return null"); + } + return codec instanceof ChainCodec ? ((ChainCodec) codec).forHeaders() : codec; + } + + @Nullable + public static PayloadCodec withContext( + @Nullable PayloadCodec codec, SerializationContext context) { + return codec == null ? null : requireHeaderCodec(codec.withContext(context)); + } + + 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 result.build(); + } + /** * Converts a {@code Map} into a {@code Map} by applying * specified converter on each value. This util should be used for things like search attributes 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..8fbf7a085a 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.HeaderCodecProvider; +import io.temporal.payload.codec.PayloadCodec; import io.temporal.payload.context.SerializationContext; import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; @@ -18,9 +21,10 @@ 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, HeaderCodecProvider { private static final ClassValue CONVERTER_DESCRIPTORS = new ClassValue() { @Override @@ -37,6 +41,12 @@ private TemporalTransferTypeDataConverter(DataConverter delegate) { this.delegate = delegate; } + @Nullable + @Override + public PayloadCodec getHeaderCodec() { + return HeaderUtils.getHeaderCodec(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..c80df70aff 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.HeaderCodecProvider; +import io.temporal.payload.codec.PayloadCodec; 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, HeaderCodecProvider { 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 getHeaderCodec() { + return HeaderUtils.getHeaderCodec(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/POJOWorkflowImplementationFactory.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/POJOWorkflowImplementationFactory.java index fb20638296..9eec23d32f 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/POJOWorkflowImplementationFactory.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/POJOWorkflowImplementationFactory.java @@ -25,6 +25,7 @@ import io.temporal.internal.worker.SingleWorkerOptions; import io.temporal.internal.worker.WorkflowExecutionException; import io.temporal.internal.worker.WorkflowExecutorCache; +import io.temporal.payload.codec.PayloadCodec; import io.temporal.payload.context.WorkflowSerializationContext; import io.temporal.worker.TypeAlreadyRegisteredException; import io.temporal.worker.WorkflowImplementationOptions; @@ -62,6 +63,7 @@ public final class POJOWorkflowImplementationFactory implements ReplayWorkflowFa private final WorkerInterceptor[] workerInterceptors; private final DataConverter dataConverter; + private final @Nullable PayloadCodec headerCodec; private final List contextPropagators; private final long defaultDeadlockDetectionTimeout; @@ -97,6 +99,7 @@ public POJOWorkflowImplementationFactory( @Nonnull String namespace) { Objects.requireNonNull(singleWorkerOptions); this.dataConverter = singleWorkerOptions.getDataConverter(); + this.headerCodec = singleWorkerOptions.getHeaderCodec(); this.workflowThreadExecutor = Objects.requireNonNull(workflowThreadExecutor); this.workerInterceptors = Objects.requireNonNull(workerInterceptors); this.cache = cache; @@ -309,7 +312,8 @@ public ReplayWorkflow getWorkflow( workflowThreadExecutor, cache, contextPropagators, - defaultDeadlockDetectionTimeout); + defaultDeadlockDetectionTimeout, + headerCodec); } @Override diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java index 99ffbcbd0c..bd0bac5387 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java @@ -18,6 +18,7 @@ import io.temporal.internal.statemachines.UpdateProtocolCallback; import io.temporal.internal.worker.WorkflowExecutionException; import io.temporal.internal.worker.WorkflowExecutorCache; +import io.temporal.payload.codec.PayloadCodec; import io.temporal.payload.context.WorkflowSerializationContext; import io.temporal.worker.WorkflowImplementationOptions; import io.temporal.workflow.UpdateInfo; @@ -62,7 +63,8 @@ public SyncWorkflow( WorkflowThreadExecutor workflowThreadExecutor, WorkflowExecutorCache cache, List contextPropagators, - long defaultDeadlockDetectionTimeout) { + long defaultDeadlockDetectionTimeout, + @Nullable PayloadCodec headerCodec) { this.workflow = Objects.requireNonNull(workflow); this.workflowImplementationOptions = workflowImplementationOptions == null @@ -85,7 +87,8 @@ public SyncWorkflow( updateDispatcher, workflowImplementationOptions, dataConverter, - contextPropagators); + contextPropagators, + headerCodec); } @Override @@ -151,6 +154,8 @@ public void handleUpdate( "update " + updateName, () -> { try { + io.temporal.common.interceptors.Header decodedHeader = + workflowContext.decodeHeader(header); workflowContext.setCurrentUpdateInfo(updateInfo); PrefixedMdc mdc = workflowContext.getReplayContext().getLoggerMdc(); mdc.put(LoggerTag.UPDATE_ID, updateInfo.getUpdateId()); @@ -159,7 +164,8 @@ public void handleUpdate( if (!callbacks.isReplaying()) { try { workflowContext.setReadOnly(true); - workflowProc.handleValidateUpdate(updateName, updateId, input, eventId, header); + workflowProc.handleValidateUpdate( + updateName, updateId, input, eventId, decodedHeader); } catch (ReadOnlyException r) { // Rethrow instead on rejecting the update to fail the WFT throw r; @@ -179,7 +185,8 @@ public void handleUpdate( callbacks.accept(); try { Optional result = - workflowProc.handleExecuteUpdate(updateName, updateId, input, eventId, header); + workflowProc.handleExecuteUpdate( + updateName, updateId, input, eventId, decodedHeader); callbacks.complete(result, null); } catch (WorkflowExecutionException e) { callbacks.complete(Optional.empty(), e.getFailure()); 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 2d5cfcba33..71fe196231 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; @@ -104,6 +107,7 @@ final class SyncWorkflowContext implements WorkflowContext, WorkflowOutboundCall private boolean readOnly = false; private final WorkflowThreadLocal currentUpdateInfo = new WorkflowThreadLocal<>(); @Nullable private String currentDetails; + @Nullable private Header startHeader; public SyncWorkflowContext( @Nonnull String namespace, @@ -114,11 +118,13 @@ public SyncWorkflowContext( UpdateDispatcher updateDispatcher, @Nullable WorkflowImplementationOptions workflowImplementationOptions, DataConverter dataConverter, - List contextPropagators) { + List contextPropagators, + @Nullable PayloadCodec headerCodec) { this.namespace = namespace; this.workflowExecution = workflowExecution; this.workflowDefinition = workflowDefinition; this.dataConverter = dataConverter; + this.headerCodec = headerCodec; this.dataConverterWithCurrentWorkflowContext = dataConverter.withContext( new WorkflowSerializationContext(namespace, workflowExecution.getWorkflowId())); @@ -267,19 +273,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 = @@ -440,16 +440,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(); @@ -583,6 +576,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, @@ -626,7 +630,10 @@ private ExecuteActivityParameters constructExecuteActivityParameters( propagators = this.contextPropagators; } io.temporal.api.common.v1.Header grpcHeader = - toHeaderGrpc(header, extractContextsAndConvertToBytes(propagators)); + toHeaderGrpc( + header, + extractContextsAndConvertToBytes(propagators), + HeaderUtils.withContext(headerCodec, activityContext(name, taskQueue, false))); attributes.setHeader(grpcHeader); if (options.getVersioningIntent() != null) { @@ -687,7 +694,11 @@ private ExecuteLocalActivityParameters constructExecuteLocalActivityParameters( } io.temporal.api.common.v1.Header grpcHeader = - toHeaderGrpc(header, extractContextsAndConvertToBytes(contextPropagators)); + toHeaderGrpc( + header, + extractContextsAndConvertToBytes(contextPropagators), + HeaderUtils.withContext( + headerCodec, activityContext(name, replayContext.getTaskQueue(), true))); activityTask.setHeader(grpcHeader); input.ifPresent(activityTask::setInput); RetryOptions retryOptions = options.getRetryOptions(); @@ -968,7 +979,16 @@ private StartChildWorkflowExecutionParameters createChildWorkflowParameters( propagators = this.contextPropagators; } io.temporal.api.common.v1.Header grpcHeader = - toHeaderGrpc(header, extractContextsAndConvertToBytes(propagators)); + toHeaderGrpc( + header, + extractContextsAndConvertToBytes(propagators), + HeaderUtils.withContext( + headerCodec, + new WorkflowSerializationContext( + Strings.isNullOrEmpty(options.getNamespace()) + ? replayContext.getNamespace() + : options.getNamespace(), + workflowId))); attributes.setHeader(grpcHeader); ParentClosePolicy parentClosePolicy = options.getParentClosePolicy(); @@ -1338,7 +1358,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(); @@ -1515,7 +1542,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(), + extractContextsAndConvertToBytes(propagators), + HeaderUtils.withContext( + headerCodec, + new WorkflowSerializationContext(namespace, workflowExecution.getWorkflowId()))); attributes.setHeader(grpcHeader); Optional payloads = @@ -1640,13 +1672,34 @@ public List getContextPropagators() { return contextPropagators; } + Header decodeHeader(io.temporal.api.common.v1.Header header) { + return new Header( + HeaderUtils.decode( + header, + HeaderUtils.withContext( + headerCodec, + new WorkflowSerializationContext(namespace, workflowExecution.getWorkflowId())))); + } + + Header getStartHeader() { + if (startHeader == null) { + // Sharing prevents decoding the start payloads again for each new workflow thread. + startHeader = + decodeHeader( + io.temporal.api.common.v1.Header.newBuilder() + .putAllFields(replayContext.getHeader()) + .build()); + } + return startHeader; + } + @Override public Map getPropagatedContexts() { if (contextPropagators == null || contextPropagators.isEmpty()) { return new HashMap<>(); } - Map headerData = new HashMap<>(replayContext.getHeader()); + Map headerData = new HashMap<>(getStartHeader().getValues()); Map contextData = new HashMap<>(); for (ContextPropagator propagator : contextPropagators) { 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 afeb70b14b..a52278e1fc 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 @@ -48,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.getStartHeader(), input); } catch (Throwable e) { applyWorkflowFailurePolicyAndRethrow(e); } finally { @@ -85,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.decodeHeader(header)); } catch (Throwable e) { applyWorkflowFailurePolicyAndRethrow(e); } @@ -93,30 +93,22 @@ 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.decodeHeader(header), args); } public void handleValidateUpdate( - String updateName, - String updateId, - Optional input, - long eventId, - io.temporal.api.common.v1.Header header) { + String updateName, String updateId, Optional input, long eventId, Header header) { try { - context.handleValidateUpdate(updateName, updateId, input, eventId, new Header(header)); + context.handleValidateUpdate(updateName, updateId, input, eventId, header); } catch (Throwable e) { applyWorkflowFailurePolicyAndRethrow(e); } } public Optional handleExecuteUpdate( - String updateName, - String updateId, - Optional input, - long eventId, - io.temporal.api.common.v1.Header header) { + String updateName, String updateId, Optional input, long eventId, Header header) { try { - return context.handleExecuteUpdate(updateName, updateId, input, eventId, new Header(header)); + return context.handleExecuteUpdate(updateName, updateId, input, eventId, header); } catch (UnsupportedContinueAsNewRequest e) { // Re-throw to fail the workflow task throw e; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java index 3d6633dbbe..409f254c84 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java @@ -10,6 +10,7 @@ import io.temporal.common.interceptors.WorkerInterceptor; import io.temporal.internal.logging.PrefixedMdc; import io.temporal.internal.payload.storage.ExternalStorageRunner; +import io.temporal.payload.codec.PayloadCodec; import io.temporal.worker.PreferredVersionProvider; import io.temporal.worker.WorkerDeploymentOptions; import java.time.Duration; @@ -34,6 +35,7 @@ public static final class Builder { private String buildId; private boolean useBuildIdForVersioning; private DataConverter dataConverter; + private PayloadCodec headerCodec; private PollerOptions pollerOptions; private Scope metricsScope; private boolean enableLoggingInReplay; @@ -63,6 +65,7 @@ private Builder(SingleWorkerOptions options) { this.identity = options.getIdentity(); this.binaryChecksum = options.getBinaryChecksum(); this.dataConverter = options.getDataConverter(); + this.headerCodec = options.getHeaderCodec(); this.pollerOptions = options.getPollerOptions(); this.metricsScope = options.getMetricsScope(); this.enableLoggingInReplay = options.getEnableLoggingInReplay(); @@ -102,6 +105,11 @@ public Builder setDataConverter(DataConverter dataConverter) { return this; } + public Builder setHeaderCodec(@Nullable PayloadCodec headerCodec) { + this.headerCodec = headerCodec; + return this; + } + public Builder setPollerOptions(PollerOptions pollerOptions) { this.pollerOptions = pollerOptions; return this; @@ -245,6 +253,7 @@ public SingleWorkerOptions build() { this.buildId, this.useBuildIdForVersioning, dataConverter, + headerCodec, pollerOptions, metricsScope, this.enableLoggingInReplay, @@ -291,6 +300,7 @@ public SingleWorkerOptions build() { private final @Nullable ExternalStorageRunner externalStorageRunner; private final CancellationToken storageCancellation; private final PrefixedMdc loggerMdc; + private final @Nullable PayloadCodec headerCodec; private SingleWorkerOptions( String identity, @@ -298,6 +308,7 @@ private SingleWorkerOptions( String buildId, boolean useBuildIdForVersioning, DataConverter dataConverter, + @Nullable PayloadCodec headerCodec, PollerOptions pollerOptions, Scope metricsScope, boolean enableLoggingInReplay, @@ -322,6 +333,7 @@ private SingleWorkerOptions( this.buildId = buildId; this.useBuildIdForVersioning = useBuildIdForVersioning; this.dataConverter = dataConverter; + this.headerCodec = headerCodec; this.pollerOptions = pollerOptions; this.metricsScope = metricsScope; this.enableLoggingInReplay = enableLoggingInReplay; @@ -382,6 +394,11 @@ public DataConverter getDataConverter() { return dataConverter; } + @Nullable + public PayloadCodec getHeaderCodec() { + return headerCodec; + } + public PollerOptions getPollerOptions() { return pollerOptions; } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java index d5cf77f135..e5c388d65a 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java @@ -68,7 +68,8 @@ public SyncActivityWorker( options.getDataConverter(), activityExecutionContextFactory, options.getWorkerInterceptors(), - options.getContextPropagators()); + options.getContextPropagators(), + options.getHeaderCodec()); this.worker = new ActivityWorker( client.getWorkflowServiceStubs(), diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncWorkflowWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncWorkflowWorker.java index c86ff4ad77..a7ae002fd3 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncWorkflowWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncWorkflowWorker.java @@ -100,7 +100,8 @@ public SyncWorkflowWorker( localActivityOptions.getDataConverter(), laActivityExecutionContextFactory, localActivityOptions.getWorkerInterceptors(), - localActivityOptions.getContextPropagators()); + localActivityOptions.getContextPropagators(), + localActivityOptions.getHeaderCodec()); laWorker = new LocalActivityWorker( namespace, taskQueue, localActivityOptions, laTaskHandler, laSlotSupplier); diff --git a/temporal-sdk/src/main/java/io/temporal/payload/codec/ChainCodec.java b/temporal-sdk/src/main/java/io/temporal/payload/codec/ChainCodec.java index eb91ccff30..98d14ec9c3 100644 --- a/temporal-sdk/src/main/java/io/temporal/payload/codec/ChainCodec.java +++ b/temporal-sdk/src/main/java/io/temporal/payload/codec/ChainCodec.java @@ -57,4 +57,60 @@ public List decode(@Nonnull List payloads) { public PayloadCodec withContext(@Nonnull SerializationContext context) { return new ChainCodec(codecs, context); } + + /** Returns true when this chain contains no codecs other than empty chains. */ + public boolean isEmpty() { + return codecs.stream() + .allMatch(codec -> codec instanceof ChainCodec && ((ChainCodec) codec).isEmpty()); + } + + /** + * Returns a header-only view that rejects missing encoders after contextual binding. + * + *

Binding before processing headers prevents a component's contextual codec from silently + * turning an enabled header pipeline into an empty chain. Normal value codec behavior is + * unchanged. + */ + public PayloadCodec forHeaders() { + List bound = new ArrayList<>(); + collectHeaderCodecs(bound); + ChainCodec execution = new ChainCodec(bound); + return new PayloadCodec() { + @Nonnull + @Override + public List encode(@Nonnull List payloads) { + return execution.encode(payloads); + } + + @Nonnull + @Override + public List decode(@Nonnull List payloads) { + return execution.decode(payloads); + } + + @Nonnull + @Override + public PayloadCodec withContext(@Nonnull SerializationContext context) { + return new ChainCodec(codecs, context).forHeaders(); + } + }; + } + + private void collectHeaderCodecs(List bound) { + if (codecs.isEmpty()) { + throw new IllegalArgumentException("Header codec binding must not return an empty chain"); + } + for (PayloadCodec codec : codecs) { + PayloadCodec scoped = + codec == null ? null : context == null ? codec : codec.withContext(context); + if (scoped == null) { + throw new IllegalArgumentException("Header codec binding must not return null"); + } + if (scoped instanceof ChainCodec) { + ((ChainCodec) scoped).collectHeaderCodecs(bound); + } else { + bound.add(scoped); + } + } + } } diff --git a/temporal-sdk/src/main/java/io/temporal/payload/codec/HeaderCodecProvider.java b/temporal-sdk/src/main/java/io/temporal/payload/codec/HeaderCodecProvider.java new file mode 100644 index 0000000000..2cf456a0c1 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/payload/codec/HeaderCodecProvider.java @@ -0,0 +1,22 @@ +package io.temporal.payload.codec; + +import javax.annotation.Nullable; + +/** + * Optional capability opting a data converter into automatic payload-header codec processing. + * + *

The returned codec handles already serialized payloads, not application values. A custom + * data-converter wrapper must implement and forward this capability if it wants to preserve its + * delegate's capability. The SDK does not inspect opaque custom wrappers. Returning {@code null} + * disables header processing without inspecting any value-conversion codec stage. + * + *

The codec must preserve one payload per header entry. Return a codec using the same + * serialization context as the provider's data converter and support {@link + * PayloadCodec#withContext}. SDK-owned wrappers forward the capability without applying value + * conversion or external storage. + */ +public interface HeaderCodecProvider { + /** Returns the codec-only stage, or {@code null} if no codec is available. */ + @Nullable + PayloadCodec getHeaderCodec(); +} 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..397f230a3e --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/payload/codec/PayloadCodecProvider.java @@ -0,0 +1,17 @@ +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 {@link + * io.temporal.common.converter.HeaderEncodingDataConverter}. 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/main/java/io/temporal/worker/Worker.java b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java index e2a45a1361..e43c892dfd 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -24,6 +24,7 @@ import io.temporal.common.converter.EncodedValues; import io.temporal.failure.TemporalFailure; import io.temporal.internal.client.WorkflowClientInternal; +import io.temporal.internal.common.HeaderUtils; import io.temporal.internal.logging.PrefixedMdc; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.sync.WorkflowInternal; @@ -1186,6 +1187,7 @@ private static SingleWorkerOptions.Builder toSingleWorkerOptions( return SingleWorkerOptions.newBuilder() .setDataConverter(clientOptions.getDataConverter()) + .setHeaderCodec(HeaderUtils.resolveHeaderCodec(clientOptions.getDataConverter())) .setExternalStorageRunner(externalStorageRunner) .setIdentity(identity) .setBuildId(buildId) diff --git a/temporal-sdk/src/test/java/io/temporal/client/HeaderCodecOptionsTest.java b/temporal-sdk/src/test/java/io/temporal/client/HeaderCodecOptionsTest.java new file mode 100644 index 0000000000..4705c58b28 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/client/HeaderCodecOptionsTest.java @@ -0,0 +1,221 @@ +package io.temporal.client; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; +import static org.mockito.Mockito.withSettings; + +import com.uber.m3.tally.NoopScope; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.failure.v1.Failure; +import io.temporal.client.schedules.Schedule; +import io.temporal.client.schedules.ScheduleActionStartWorkflow; +import io.temporal.client.schedules.ScheduleClient; +import io.temporal.client.schedules.ScheduleClientOptions; +import io.temporal.client.schedules.ScheduleClientPlugin; +import io.temporal.client.schedules.ScheduleOptions; +import io.temporal.client.schedules.ScheduleSpec; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.CodecDataConverterTest.PrefixPayloadCodec; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DataConverterWithCodec; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.HeaderEncodingDataConverter; +import io.temporal.common.interceptors.Header; +import io.temporal.internal.common.HeaderUtils; +import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter; +import io.temporal.internal.payload.storage.ExternalStorageDataConverter; +import io.temporal.internal.worker.SingleWorkerOptions; +import io.temporal.payload.codec.ChainCodec; +import io.temporal.payload.codec.HeaderCodecProvider; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.codec.PayloadCodecException; +import io.temporal.payload.context.WorkflowSerializationContext; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; +import java.util.Collections; +import java.util.Optional; +import org.junit.Test; + +public class HeaderCodecOptionsTest { + @Test + public void disabledClientsDoNotInspectCustomConverterCapability() { + DataConverterWithCodec custom = mock(DataConverterWithCodec.class); + assertNull(HeaderUtils.resolveHeaderCodec(custom)); + assertNull( + HeaderUtils.resolveHeaderCodec( + new ExternalStorageDataConverter( + TemporalTransferTypeDataConverter.wrap(custom), null))); + verifyNoInteractions(custom); + } + + @Test + public void enabledOptionsAndWorkerCopiesPreservePolicyAndCodec() { + CodecDataConverter ordinary = + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, + Collections.singletonList(new PrefixPayloadCodec())); + assertNull(HeaderUtils.resolveHeaderCodec(ordinary)); + HeaderEncodingDataConverter converter = new HeaderEncodingDataConverter(ordinary); + SingleWorkerOptions worker = + SingleWorkerOptions.newBuilder() + .setDataConverter(converter) + .setHeaderCodec(HeaderUtils.resolveHeaderCodec(converter)) + .build(); + assertSame( + worker.getHeaderCodec(), SingleWorkerOptions.newBuilder(worker).build().getHeaderCodec()); + assertSame( + converter.getHeaderCodec(), + HeaderUtils.resolveHeaderCodec( + new ExternalStorageDataConverter( + TemporalTransferTypeDataConverter.wrap(converter), null))); + DataConverter forwarding = + mock(DataConverter.class, withSettings().extraInterfaces(HeaderCodecProvider.class)); + when(((HeaderCodecProvider) forwarding).getHeaderCodec()) + .thenReturn(converter.getHeaderCodec()); + assertSame(converter.getHeaderCodec(), HeaderUtils.resolveHeaderCodec(forwarding)); + assertNull(HeaderUtils.resolveHeaderCodec(mock(DataConverter.class))); + } + + @Test + public void wrapperRejectsMissingNullAndEmptyCodecStages() { + DataConverterWithCodec nullProvider = mock(DataConverterWithCodec.class); + DataConverterWithCodec emptyProvider = mock(DataConverterWithCodec.class); + when(emptyProvider.getPayloadCodec()).thenReturn(new ChainCodec(Collections.emptyList())); + assertThrows(NullPointerException.class, () -> new HeaderEncodingDataConverter(null)); + for (DataConverterWithCodec unavailable : + new DataConverterWithCodec[] { + nullProvider, + emptyProvider, + new CodecDataConverter(DefaultDataConverter.STANDARD_INSTANCE, Collections.emptyList()), + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, + Collections.singletonList(new ChainCodec(Collections.emptyList()))) + }) { + assertThrows( + IllegalArgumentException.class, () -> new HeaderEncodingDataConverter(unavailable)); + } + } + + @Test + public void schedulePluginsCanInstallHeaderWrapperBeforeCodecResolution() { + WorkflowServiceStubs service = scheduleService(); + ScheduleClientPlugin plugin = mock(ScheduleClientPlugin.class); + when(plugin.getName()).thenReturn("header-codec"); + PayloadCodec codec = mock(PayloadCodec.class); + PayloadCodecException codecFailure = new PayloadCodecException("plugin header codec"); + when(codec.withContext(any())).thenReturn(codec); + when(codec.encode(any())).thenThrow(codecFailure); + DataConverter converter = + new HeaderEncodingDataConverter( + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, Collections.singletonList(codec))); + doAnswer( + call -> { + ScheduleClientOptions.Builder builder = call.getArgument(0); + return builder.setDataConverter(converter); + }) + .when(plugin) + .configureScheduleClient(any()); + ScheduleClientOptions options = ScheduleClientOptions.newBuilder().setPlugins(plugin).build(); + Schedule schedule = + Schedule.newBuilder() + .setSpec(ScheduleSpec.newBuilder().build()) + .setAction( + ScheduleActionStartWorkflow.newBuilder() + .setWorkflowType("workflow") + .setOptions( + WorkflowOptions.newBuilder() + .setWorkflowId("workflow") + .setTaskQueue("q") + .build()) + .setHeader( + new Header( + Collections.singletonMap( + "trace", + DefaultDataConverter.STANDARD_INSTANCE.toPayload("context").get()))) + .build()) + .build(); + assertSame( + codecFailure, + assertThrows( + PayloadCodecException.class, + () -> + ScheduleClient.newInstance(service, options) + .createSchedule("schedule", schedule, ScheduleOptions.newBuilder().build()))); + } + + @Test + public void workflowPluginsCanInstallHeaderWrapper() { + HeaderEncodingDataConverter converter = + new HeaderEncodingDataConverter( + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, + Collections.singletonList(new PrefixPayloadCodec()))); + WorkflowClientPlugin plugin = mock(WorkflowClientPlugin.class); + when(plugin.getName()).thenReturn("header-codec"); + doAnswer( + call -> + ((WorkflowClientOptions.Builder) call.getArgument(0)).setDataConverter(converter)) + .when(plugin) + .configureWorkflowClient(any()); + WorkflowClient client = + WorkflowClient.newInstance( + scheduleService(), WorkflowClientOptions.newBuilder().setPlugins(plugin).build()); + assertSame( + converter.getHeaderCodec(), + HeaderUtils.resolveHeaderCodec(client.getOptions().getDataConverter())); + } + + @Test + public void wrapperDelegatesValuesFailuresAndSerializationContexts() { + DataConverterWithCodec delegate = mock(DataConverterWithCodec.class); + DataConverter contextual = mock(DataConverter.class); + PayloadCodec codec = mock(PayloadCodec.class); + PayloadCodec scopedCodec = new PrefixPayloadCodec(); + WorkflowSerializationContext context = + new WorkflowSerializationContext("namespace", "workflow"); + when(delegate.getPayloadCodec()).thenReturn(codec); + when(delegate.withContext(context)).thenReturn(contextual); + when(codec.withContext(context)).thenReturn(scopedCodec); + HeaderEncodingDataConverter converter = + new HeaderEncodingDataConverter(delegate).withContext(context); + assertSame(scopedCodec, converter.getHeaderCodec()); + Payload payload = Payload.getDefaultInstance(); + Optional payloads = Optional.of(Payloads.getDefaultInstance()); + Class[] types = {String.class}; + java.lang.reflect.Type[] genericTypes = {String.class}; + when(contextual.toPayload("value")).thenReturn(Optional.of(payload)); + when(contextual.fromPayload(payload, String.class, String.class)).thenReturn("value"); + when(contextual.toPayloads("value")).thenReturn(payloads); + when(contextual.fromPayloads(0, payloads, String.class, String.class)).thenReturn("value"); + Object[] values = {"value"}; + when(contextual.fromPayloads(payloads, types, genericTypes)).thenReturn(values); + assertEquals(Optional.of(payload), converter.toPayload("value")); + assertEquals("value", converter.fromPayload(payload, String.class, String.class)); + assertSame(payloads, converter.toPayloads("value")); + assertEquals("value", converter.fromPayloads(0, payloads, String.class, String.class)); + assertSame(values, converter.fromPayloads(payloads, types, genericTypes)); + RuntimeException exception = new RuntimeException("failure"); + Failure failure = Failure.getDefaultInstance(); + when(contextual.exceptionToFailure(exception)).thenReturn(failure); + when(contextual.failureToException(failure)).thenReturn(exception); + assertSame(failure, converter.exceptionToFailure(exception)); + assertSame(exception, converter.failureToException(failure)); + } + + private static WorkflowServiceStubs scheduleService() { + WorkflowServiceStubs service = mock(WorkflowServiceStubs.class); + WorkflowServiceStubsOptions options = mock(WorkflowServiceStubsOptions.class); + when(service.getOptions()).thenReturn(options); + when(options.getMetricsScope()).thenReturn(new NoopScope()); + return service; + } +} 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..99e1b281c2 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; @@ -53,10 +54,11 @@ public class WorkflowIdSignedPayloadsTest { new ManualCompletionActivityImpl(); private static final DataConverter codecDataConverter = - new CodecDataConverter( - DefaultDataConverter.STANDARD_INSTANCE, - Collections.singletonList(new PayloadEncoderWithWorkflowIdSignature()), - true); + new HeaderEncodingDataConverter( + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, + Collections.singletonList(new PayloadEncoderWithWorkflowIdSignature()), + true)); @Rule public SDKTestWorkflowRule testWorkflowRule = @@ -67,7 +69,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 +84,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 +324,40 @@ public Object execute(EncodedValues args) { } } + private static class HeaderContext implements ContextPropagator { + private static final Payload VALUE = + DefaultDataConverter.STANDARD_INSTANCE.toPayload("context").get(); + + @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..6ebaca7571 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityTaskExecutorsTest.java @@ -0,0 +1,62 @@ +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.converter.HeaderEncodingDataConverter; +import io.temporal.common.interceptors.WorkerInterceptor; +import io.temporal.internal.common.HeaderUtils; +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.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, + HeaderUtils.resolveHeaderCodec( + enabled ? new HeaderEncodingDataConverter(converter) : converter)) + .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..bc8447bb8e --- /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.converter.HeaderEncodingDataConverter; +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(enabled ? new HeaderEncodingDataConverter(converter) : 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 = new Header(Collections.singletonMap("_tracer-data", payload)); + 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/RootWorkflowClientInvokerLinkPropagationTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerLinkPropagationTest.java index bbb151534f..7dad37c38b 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerLinkPropagationTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerLinkPropagationTest.java @@ -77,7 +77,8 @@ public void setUp() { options.getQueryRejectCondition(), options.getContextPropagators(), options.getDataConverter(), - new WorkerFactoryRegistry()); + new WorkerFactoryRegistry(), + null); Scope metricsScope = new RootScopeBuilder().reportEvery(com.uber.m3.util.Duration.ofMillis(10)); nexusCtx = new InternalNexusOperationContext( 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..7eae383fdf 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,11 @@ 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.converter.HeaderEncodingDataConverter; import io.temporal.common.interceptors.Header; import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.QueryInput; import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalInput; @@ -207,6 +212,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 HeaderEncodingDataConverter( + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, Collections.singletonList(codec), false)); + Header header = headerWith("trace", "context"); + 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 +249,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..83c73bf709 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,13 @@ 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.HeaderEncodingDataConverter; 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 +36,47 @@ 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 = + enabled ? new HeaderEncodingDataConverter(codecConverter) : codecConverter; + ScheduleProtoUtil util = + new ScheduleProtoUtil(null, "namespace", Collections.emptyList(), converter); + Header header = + new Header( + Collections.singletonMap( + "_tracer-data", DefaultDataConverter.STANDARD_INSTANCE.toPayload("trace").get())); + 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(header.getValues(), 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/common/HeaderUtilsTest.java b/temporal-sdk/src/test/java/io/temporal/internal/common/HeaderUtilsTest.java new file mode 100644 index 0000000000..81e9d7b426 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/common/HeaderUtilsTest.java @@ -0,0 +1,184 @@ +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.verify; +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.converter.CodecDataConverter; +import io.temporal.common.converter.CodecDataConverterTest.PrefixPayloadCodec; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DataConverterWithCodec; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.HeaderEncodingDataConverter; +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.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 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 = new Header(values); + Header overrides = new Header(Collections.singletonMap("_tracer-data", serialized)); + CountingCodec codec = new CountingCodec(); + DataConverter valuesConverter = mock(DataConverter.class); + CodecDataConverter configured = + new CodecDataConverter(valuesConverter, Collections.singletonList(codec)); + DataConverter enabled = + new ExternalStorageDataConverter( + ClientDataConverterFactory.forClient(new HeaderEncodingDataConverter(configured)), + null) + .withContext(new WorkflowSerializationContext("namespace", "workflow")); + PayloadCodec disabledCodec = HeaderUtils.resolveHeaderCodec(configured); + PayloadCodec enabledCodec = HeaderUtils.resolveHeaderCodec(enabled); + + io.temporal.api.common.v1.Header plain = HeaderUtils.toHeaderGrpc(header, overrides); + assertEquals(plain, HeaderUtils.toHeaderGrpc(header, overrides, disabledCodec)); + assertSame(plain, HeaderUtils.decode(plain, disabledCodec)); + 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 incompatibleLegacyHeadersDoNotFallBackToPlaintext() { + HeaderUtils.decode( + HeaderUtils.toHeaderGrpc( + new Header( + Collections.singletonMap( + "legacy", DefaultDataConverter.STANDARD_INSTANCE.toPayload("context").get())), + null), + new PrefixPayloadCodec()); + } + + @Test + public void contextualizationCannotSilentlyDisableTheCodec() { + PayloadCodec codec = mock(PayloadCodec.class); + WorkflowSerializationContext context = + new WorkflowSerializationContext("namespace", "workflow"); + assertThrows(IllegalArgumentException.class, () -> HeaderUtils.withContext(codec, context)); + when(codec.withContext(any())).thenReturn(new ChainCodec(Collections.emptyList())); + assertThrows(IllegalArgumentException.class, () -> HeaderUtils.withContext(codec, context)); + } + + @Test + public void combinedConverterRejectsMissingContextualChainComponents() { + for (boolean nested : new boolean[] {false, true}) { + for (boolean sibling : new boolean[] {false, true}) { + PayloadCodec component = mock(PayloadCodec.class); + PayloadCodec source = + nested ? new ChainCodec(Collections.singletonList(component)) : component; + DataConverter valuesConverter = mock(DataConverter.class); + CodecDataConverter converter = + new CodecDataConverter( + valuesConverter, + sibling + ? Arrays.asList(source, new PrefixPayloadCodec()) + : Collections.singletonList(source)); + WorkflowSerializationContext context = + new WorkflowSerializationContext("namespace", "workflow"); + HeaderEncodingDataConverter enabled = new HeaderEncodingDataConverter(converter); + PayloadCodec headers = HeaderUtils.resolveHeaderCodec(enabled); + assertThrows( + IllegalArgumentException.class, () -> HeaderUtils.withContext(headers, context)); + assertThrows(IllegalArgumentException.class, () -> enabled.withContext(context)); + when(component.withContext(any())).thenReturn(new ChainCodec(Collections.emptyList())); + assertThrows( + IllegalArgumentException.class, () -> HeaderUtils.withContext(headers, context)); + assertThrows(IllegalArgumentException.class, () -> enabled.withContext(context)); + verifyNoInteractions(valuesConverter); + } + } + } + + @Test + public void rejectsEmptyNestedComponentsAlongsideValidCodecs() { + ChainCodec stage = + new ChainCodec( + Arrays.asList(new ChainCodec(Collections.emptyList()), new PrefixPayloadCodec())); + DataConverterWithCodec direct = mock(DataConverterWithCodec.class); + when(direct.getPayloadCodec()).thenReturn(stage); + assertThrows(IllegalArgumentException.class, () -> new HeaderEncodingDataConverter(direct)); + + PayloadCodec component = mock(PayloadCodec.class); + WorkflowSerializationContext context = + new WorkflowSerializationContext("namespace", "workflow"); + when(component.withContext(context)).thenReturn(stage); + HeaderEncodingDataConverter converter = + new HeaderEncodingDataConverter( + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, Collections.singletonList(component))); + assertThrows(IllegalArgumentException.class, () -> converter.withContext(context)); + verify(component).withContext(context); + + List payloads = + Collections.singletonList(DefaultDataConverter.STANDARD_INSTANCE.toPayload("value").get()); + assertEquals(payloads, stage.decode(stage.encode(payloads))); + } + + @Test(expected = PayloadCodecException.class) + public void rejectsCodecsThatChangeHeaderCardinality() { + HeaderUtils.toHeaderGrpc( + new Header(Collections.singletonMap("key", Payload.getDefaultInstance())), + null, + new CountingCodec() { + @Override + public List encode(@Nonnull List payloads) { + return Collections.emptyList(); + } + }); + } + + private 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().setData(payloads.get(0).toByteString()).build()); + } + + @Override + @Nonnull + public List decode(@Nonnull List payloads) { + decoded++; + try { + ByteString data = payloads.get(0).getData(); + return Collections.singletonList(Payload.parseFrom(data)); + } catch (Exception e) { + throw new PayloadCodecException(e); + } + } + } +} 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..ee3a1ee1e5 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,27 @@ 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.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 +50,64 @@ 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, + DefaultDataConverter.STANDARD_INSTANCE, + Collections.emptyList(), + codec); + context.setReplayContext(mockReplayWorkflowContext); + 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(), + new Header( + Collections.singletonMap("key", Payload.getDefaultInstance()))))); + 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<>(); diff --git a/temporal-testing/src/main/java/io/temporal/internal/sync/DummySyncWorkflowContext.java b/temporal-testing/src/main/java/io/temporal/internal/sync/DummySyncWorkflowContext.java index 8ca8ba9d1c..0b00ab9e6b 100644 --- a/temporal-testing/src/main/java/io/temporal/internal/sync/DummySyncWorkflowContext.java +++ b/temporal-testing/src/main/java/io/temporal/internal/sync/DummySyncWorkflowContext.java @@ -33,6 +33,7 @@ public static SyncWorkflowContext newDummySyncWorkflowContext() { new UpdateDispatcher(DefaultDataConverter.STANDARD_INSTANCE), null, DefaultDataConverter.STANDARD_INSTANCE, + null, null); context.setReplayContext(new DummyReplayWorkflowContext()); context.initHeadOutboundCallsInterceptor(context); diff --git a/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java b/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java index d05409e082..517b492952 100644 --- a/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java +++ b/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java @@ -31,6 +31,7 @@ import io.temporal.internal.activity.ActivityExecutionContextFactoryImpl; import io.temporal.internal.activity.ActivityTaskHandlerImpl; import io.temporal.internal.client.WorkflowClientInternal; +import io.temporal.internal.common.HeaderUtils; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.payload.storage.ExternalStorageDataConverter; import io.temporal.internal.payload.storage.ExternalStorageRunner; @@ -131,7 +132,8 @@ public TestActivityEnvironmentInternal(@Nullable TestEnvironmentOptions options) dataConverter, activityExecutionContextFactory, testEnvironmentOptions.getWorkerFactoryOptions().getWorkerInterceptors(), - workflowClient.getOptions().getContextPropagators()); + workflowClient.getOptions().getContextPropagators(), + HeaderUtils.resolveHeaderCodec(dataConverter)); } private class HeartbeatInterceptingService extends WorkflowServiceGrpc.WorkflowServiceImplBase {