Add opt-in header codec processing

CHANGELOG.md

@@ -25,6 +25,11 @@ to docs, or any other relevant information.
   `WorkflowOutboundCallsInterceptorBase` need to add them, and Base subclasses override them to see calls with options.
 
 ### Added
+- Tracing options and `ContextPropagator.encodeHeaders()` can opt into header processing using
+  the configured client's payload codec. Only opted-in producers' output is encoded, while
+  application-supplied wire-ready headers are unchanged. Receiving consumers use private decoded
+  views. Existing propagators default to false; custom converter wrappers can expose their codec
+  through `PayloadCodecProvider`.
 - Applications can use the experimental `@TransferTypeConvertible` annotation and
   `TransferTypeConverter` interface to convert top-level model values to transfer types. The SDK
   applies this conversion to workflow, activity, Nexus, and schedule payloads, including failure
@@ -71,6 +76,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,38 @@ If you cannot use protobuf-java 3.25 >=, you can try `temporal-shaded` which inc
 
 We'd love your help in improving the Temporal Java SDK. Please review our [contribution guidelines](CONTRIBUTING.md).
 
+## Automatic payload-header codecs
+
+User-supplied header payloads are wire-ready, including any codec encoding.
+Tracing and context propagators can independently opt into processing their own
+headers with the codec from the configured client's data converter:
+
+```java
+DataConverter payloadConverter = DefaultDataConverter.newDefaultInstance();
+DataConverter converter = new CodecDataConverter(payloadConverter, codecs);
+WorkflowClientOptions options =
+    WorkflowClientOptions.newBuilder()
+        .setDataConverter(converter)
+        .build();
+```
+
+Set `OpenTracingOptions.Builder.setEncodeHeaders(true)` to encode only that integration's
+tracing headers, retaining standard tracing serialization. The OpenTelemetry plugin and worker
+configuration builders expose the same option. Context propagators opt in by overriding
+`ContextPropagator.encodeHeaders()` to return true; the default is false.
+
+Opted-in propagators must produce serialized **but not already codec-encoded** payloads.
+The SDK applies the configured client's codec to that producer's output, not to other headers.
+Receiving integrations and propagators must opt in to read encoded headers and use a compatible codec.
+They receive private decoded views of the original wire headers; codecs must pass through
+encodings they do not recognize. Ordinary interceptors continue to receive wire-ready payloads.
+Retained schedule headers remain wire-ready and are not re-encoded when an action is reused.
+
+`CodecDataConverter` exposes its codec through `PayloadCodecProvider`; custom converter wrappers
+must forward that capability. Without a configured codec, processing is an identity operation.
+No value conversion is applied to header payloads. Nexus string headers and gRPC metadata are
+unaffected. Codec failures are propagated, never retried as plaintext.
+
 ## Development
 
 ### Development Environment

contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryPlugin.java

@@ -37,6 +37,7 @@ public final class OpenTelemetryPlugin extends SimplePlugin {
   private final Runnable flushHook;
   private final IdGenerator idGenerator;
   private final boolean flushOnWorkerFactoryShutdown;
+  private final boolean encodeHeaders;
   private final List<Scope> metricsScopes = new ArrayList<>();
 
   private OpenTelemetry openTelemetry;
@@ -53,6 +54,7 @@ private OpenTelemetryPlugin(Builder builder) {
     this.flushHook = builder.flushHook;
     this.idGenerator = builder.idGenerator;
     this.flushOnWorkerFactoryShutdown = builder.flushOnWorkerFactoryShutdown;
+    this.encodeHeaders = builder.encodeHeaders;
   }
 
   public static Builder newBuilder() {
@@ -94,15 +96,15 @@ public void configureServiceStubs(@Nonnull WorkflowServiceStubsOptions.Builder b
   @Override
   public void configureWorkflowClient(@Nonnull WorkflowClientOptions.Builder builder) {
     OpenTracingOptions tracingOptions =
-        OpenTelemetryWorker.createOpenTracingOptions(getOpenTelemetry());
+        OpenTelemetryWorker.createOpenTracingOptions(getOpenTelemetry(), encodeHeaders);
     OpenTelemetryWorker.appendClientInterceptor(
         builder, new io.temporal.opentracing.OpenTracingClientInterceptor(tracingOptions));
   }
 
   @Override
   public void configureWorkerFactory(@Nonnull WorkerFactoryOptions.Builder builder) {
     OpenTracingOptions tracingOptions =
-        OpenTelemetryWorker.createOpenTracingOptions(getOpenTelemetry());
+        OpenTelemetryWorker.createOpenTracingOptions(getOpenTelemetry(), encodeHeaders);
     OpenTelemetryWorker.appendWorkerInterceptor(
         builder, new io.temporal.opentracing.OpenTracingWorkerInterceptor(tracingOptions));
   }
@@ -155,6 +157,17 @@ private List<Scope> drainMetricsScopes() {
 
   /** Builder for {@link OpenTelemetryPlugin}. */
   public static final class Builder {
+    private boolean encodeHeaders;
+
+    /**
+     * Uses the configured client's payload codec for tracing headers. Defaults to false;
+     * application-supplied headers and standard tracing serialization are unchanged.
+     */
+    public Builder setEncodeHeaders(boolean encodeHeaders) {
+      this.encodeHeaders = encodeHeaders;
+      return this;
+    }
+
     private final Map<String, String> env;
     private OpenTelemetry openTelemetry;
     private String endpoint;

contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryWorker.java

@@ -141,10 +141,23 @@ public static void configureTracing(
       @Nonnull WorkflowClientOptions.Builder clientOptions,
       @Nonnull WorkerFactoryOptions.Builder workerFactoryOptions,
       @Nonnull OpenTelemetry openTelemetry) {
+    configureTracing(clientOptions, workerFactoryOptions, openTelemetry, false);
+  }
+
+  /**
+   * Configures tracing with optional codec processing using the configured client's payload codec.
+   * Only tracing headers are encoded; application-supplied headers are unchanged.
+   */
+  public static void configureTracing(
+      @Nonnull WorkflowClientOptions.Builder clientOptions,
+      @Nonnull WorkerFactoryOptions.Builder workerFactoryOptions,
+      @Nonnull OpenTelemetry openTelemetry,
+      boolean encodeHeaders) {
     Objects.requireNonNull(clientOptions, "clientOptions");
     Objects.requireNonNull(workerFactoryOptions, "workerFactoryOptions");
     OpenTracingOptions tracingOptions =
-        createOpenTracingOptions(Objects.requireNonNull(openTelemetry, "openTelemetry"));
+        createOpenTracingOptions(
+            Objects.requireNonNull(openTelemetry, "openTelemetry"), encodeHeaders);
     appendClientInterceptor(clientOptions, new OpenTracingClientInterceptor(tracingOptions));
     appendWorkerInterceptor(workerFactoryOptions, new OpenTracingWorkerInterceptor(tracingOptions));
   }
@@ -205,6 +218,17 @@ public static String resolveServiceName(
   }
 
   public static final class Builder {
+    private boolean encodeHeaders;
+
+    /**
+     * Uses the configured client's payload codec for tracing headers. Defaults to false;
+     * application-supplied headers and standard tracing serialization are unchanged.
+     */
+    public Builder setEncodeHeaders(boolean encodeHeaders) {
+      this.encodeHeaders = encodeHeaders;
+      return this;
+    }
+
     private final Map<String, String> env;
     private OpenTelemetry openTelemetry;
     private String endpoint;
@@ -294,7 +318,7 @@ public void apply(
           resolvedOpenTelemetry,
           getServiceName(),
           metricsReportInterval);
-      configureTracing(clientOptions, workerFactoryOptions, resolvedOpenTelemetry);
+      configureTracing(clientOptions, workerFactoryOptions, resolvedOpenTelemetry, encodeHeaders);
       if (flushHook == null) {
         configureFlushHook(addShutdownHook, resolvedOpenTelemetry, flushTimeout);
       } else {
@@ -378,7 +402,13 @@ static Scope createMetricsScope(
   }
 
   static OpenTracingOptions createOpenTracingOptions(@Nonnull OpenTelemetry openTelemetry) {
+    return createOpenTracingOptions(openTelemetry, false);
+  }
+
+  static OpenTracingOptions createOpenTracingOptions(
+      @Nonnull OpenTelemetry openTelemetry, boolean encodeHeaders) {
     return OpenTracingOptions.newBuilder()
+        .setEncodeHeaders(encodeHeaders)
         .setTracer(
             OpenTracingShim.createTracerShim(
                 Objects.requireNonNull(openTelemetry, "openTelemetry")))

contrib/temporal-opentracing/src/main/java/io/temporal/opentracing/OpenTracingOptions.java

@@ -17,6 +17,7 @@ public class OpenTracingOptions {
   private final SpanBuilderProvider spanBuilderProvider;
   private final OpenTracingSpanContextCodec spanContextCodec;
   private final Predicate<Throwable> isErrorPredicate;
+  private final boolean encodeHeaders;
 
   public static OpenTracingOptions getDefaultInstance() {
     return DEFAULT_INSTANCE;
@@ -26,12 +27,14 @@ private OpenTracingOptions(
       Tracer tracer,
       SpanBuilderProvider spanBuilderProvider,
       OpenTracingSpanContextCodec spanContextCodec,
-      Predicate<Throwable> isErrorPredicate) {
+      Predicate<Throwable> isErrorPredicate,
+      boolean encodeHeaders) {
     if (tracer == null) throw new IllegalArgumentException("tracer shouldn't be null");
     this.tracer = tracer;
     this.spanBuilderProvider = spanBuilderProvider;
     this.spanContextCodec = spanContextCodec;
     this.isErrorPredicate = isErrorPredicate;
+    this.encodeHeaders = encodeHeaders;
   }
 
   @Nonnull
@@ -54,6 +57,10 @@ public Predicate<Throwable> getIsErrorPredicate() {
     return isErrorPredicate;
   }
 
+  public boolean isEncodeHeaders() {
+    return encodeHeaders;
+  }
+
   public static Builder newBuilder() {
     return new Builder();
   }
@@ -64,9 +71,21 @@ public static final class Builder {
     private OpenTracingSpanContextCodec spanContextCodec =
         OpenTracingSpanContextCodec.TEXT_MAP_INJECT_EXTRACT_CODEC;
     private Predicate<Throwable> isErrorPredicate = t -> true;
+    private boolean encodeHeaders;
 
     private Builder() {}
 
+    /**
+     * Uses the configured client's payload codec for this integration's tracing headers. Defaults
+     * to false. Standard tracing serialization and other producers' headers are unchanged.
+     * Configure the same setting on receiving workers. Codecs must pass through unrecognized
+     * encodings.
+     */
+    public Builder setEncodeHeaders(boolean encodeHeaders) {
+      this.encodeHeaders = encodeHeaders;
+      return this;
+    }
+
     public Builder setTracer(@Nullable Tracer tracer) {
       this.tracer = tracer;
       return this;
@@ -116,7 +135,8 @@ public OpenTracingOptions build() {
           MoreObjects.firstNonNull(tracer, GlobalTracer.get()),
           spanBuilderProvider,
           spanContextCodec,
-          isErrorPredicate);
+          isErrorPredicate,
+          encodeHeaders);
     }
   }
 }

contrib/temporal-opentracing/src/main/java/io/temporal/opentracing/internal/ContextAccessor.java

@@ -13,6 +13,7 @@
 import io.temporal.opentracing.OpenTracingOptions;
 import io.temporal.opentracing.OpenTracingSpanContextCodec;
 import java.lang.reflect.Type;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
@@ -24,9 +25,11 @@ public class ContextAccessor {
       new TypeToken<HashMap<String, String>>() {}.getType();
 
   private final OpenTracingSpanContextCodec codec;
+  private final boolean encodeHeaders;
 
   public ContextAccessor(OpenTracingOptions options) {
     this.codec = options.getSpanContextCodec();
+    this.encodeHeaders = options.isEncodeHeaders();
   }
 
   public Span writeSpanContextToHeader(
@@ -40,11 +43,13 @@ public void writeSpanContextToHeader(SpanContext spanContext, Header header, Tra
     Map<String, String> serializedSpanContext = codec.encode(spanContext, tracer);
     Optional<Payload> payload =
         DefaultDataConverter.STANDARD_INSTANCE.toPayload(serializedSpanContext);
-    header.getValues().put(TRACER_HEADER_KEY, payload.get());
+    header.putAll(Collections.singletonMap(TRACER_HEADER_KEY, payload.get()), encodeHeaders);
   }
 
   public SpanContext readSpanContextFromHeader(Header header, Tracer tracer) {
-    Payload payload = header.getValues().get(TRACER_HEADER_KEY);
+    Payload payload =
+        (encodeHeaders ? header.getCodecDecodedValues() : header.getValues())
+            .get(TRACER_HEADER_KEY);
     if (payload == null) {
       return null;
     }

contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/HeaderCodecTest.java

@@ -0,0 +1,88 @@
+package io.temporal.opentracing;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyNoInteractions;
+
+import com.google.protobuf.ByteString;
+import io.opentracing.Span;
+import io.opentracing.mock.MockTracer;
+import io.temporal.api.common.v1.Payload;
+import io.temporal.common.converter.CodecDataConverter;
+import io.temporal.common.converter.DataConverter;
+import io.temporal.common.converter.DefaultDataConverter;
+import io.temporal.common.interceptors.Header;
+import io.temporal.internal.common.HeaderUtils;
+import io.temporal.opentracing.internal.ContextAccessor;
+import io.temporal.payload.codec.PayloadCodec;
+import io.temporal.payload.codec.PayloadCodecException;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nonnull;
+import org.junit.Test;
+
+public class HeaderCodecTest {
+  @Test
+  public void tracingKeepsStandardSerializationAndUsesHeaderCodecExactlyOnce() {
+    for (boolean enabled : new boolean[] {false, true}) {
+      MockTracer tracer = new MockTracer();
+      ContextAccessor accessor =
+          new ContextAccessor(
+              OpenTracingOptions.newBuilder().setTracer(tracer).setEncodeHeaders(enabled).build());
+      Span span = tracer.buildSpan("parent").start();
+      CountingCodec codec = new CountingCodec();
+      DataConverter valuesConverter = mock(DataConverter.class);
+      CodecDataConverter converter =
+          new CodecDataConverter(valuesConverter, Collections.singletonList(codec));
+      PayloadCodec headerCodec = HeaderUtils.resolveHeaderCodec(converter);
+      Header header = Header.empty();
+      accessor.writeSpanContextToHeader(span.context(), header, tracer);
+      Payload plaintext = DefaultDataConverter.STANDARD_INSTANCE.toPayload("other producer").get();
+      header.getValues().put("other", plaintext);
+      io.temporal.api.common.v1.Header wire = HeaderUtils.toHeaderGrpc(header, null, headerCodec);
+      Header received = new Header(wire, headerCodec);
+      assertEquals(plaintext, received.getValues().get("other"));
+
+      assertEquals(
+          span.context().toTraceId(),
+          accessor.readSpanContextFromHeader(received, tracer).toTraceId());
+      assertEquals(enabled ? 1 : 0, codec.encoded);
+      assertEquals(enabled ? 1 : 0, codec.decoded);
+      assertEquals(wire.getFieldsMap(), received.getValues());
+      verifyNoInteractions(valuesConverter);
+      span.finish();
+      tracer.close();
+    }
+  }
+
+  static class CountingCodec implements PayloadCodec {
+    private int encoded;
+    private int decoded;
+
+    @Override
+    @Nonnull
+    public List<Payload> encode(@Nonnull List<Payload> payloads) {
+      encoded++;
+      return Collections.singletonList(
+          Payload.newBuilder()
+              .putMetadata("encoding", ByteString.copyFromUtf8("binary/test-tracing"))
+              .setData(payloads.get(0).toByteString())
+              .build());
+    }
+
+    @Override
+    @Nonnull
+    public List<Payload> decode(@Nonnull List<Payload> payloads) {
+      if (!ByteString.copyFromUtf8("binary/test-tracing")
+          .equals(payloads.get(0).getMetadataOrDefault("encoding", ByteString.EMPTY))) {
+        return payloads;
+      }
+      decoded++;
+      try {
+        return Collections.singletonList(Payload.parseFrom(payloads.get(0).getData()));
+      } catch (Exception e) {
+        throw new PayloadCodecException(e);
+      }
+    }
+  }
+}