Add opt-in header codec processing

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).

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

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<Payload> encode(@Nonnull List<Payload> payloads) {
+      encoded++;
+      return Collections.singletonList(
+          Payload.newBuilder().setData(payloads.get(0).toByteString()).build());
+    }
+
+    @Override
+    @Nonnull
+    public List<Payload> decode(@Nonnull List<Payload> payloads) {
+      decoded++;
+      try {
+        return Collections.singletonList(Payload.parseFrom(payloads.get(0).getData()));
+      } catch (Exception e) {
+        throw new PayloadCodecException(e);
+      }
+    }
+  }
+}

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(

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();
     }

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(

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}
    *
+   * <p>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<String, Payload> serializeContext(Object context);

temporal-sdk/src/main/java/io/temporal/common/converter/CodecDataConverter.java

@@ -27,8 +27,11 @@
  * <p>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 &lt;-&gt; byte manipulation such as encryption or compression
+ *
+ * <p>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<Payload> encode(@Nonnull List<Payload> payloads) {

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 {}