.start() pulls headers from context propagators.

Author: GregoryTravis

temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java

@@ -1,6 +1,8 @@
 package io.temporal.client;
 
 import com.uber.m3.tally.Scope;
+import io.temporal.api.common.v1.Payload;
+import io.temporal.common.context.ContextPropagator;
 import io.temporal.common.interceptors.ActivityClientCallsInterceptor;
 import io.temporal.common.interceptors.ActivityClientInterceptor;
 import io.temporal.common.interceptors.Header;
@@ -15,6 +17,9 @@
 import java.lang.reflect.Method;
 import java.lang.reflect.Type;
 import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.Stream;
 import javax.annotation.Nullable;
@@ -304,7 +309,7 @@ public UntypedActivityHandle start(
                 activityType,
                 Arrays.asList(args != null ? args : new Object[0]),
                 options,
-                Header.empty()));
+                propagatedHeader()));
     return new ActivityHandleImpl(output.getActivityId(), output.getActivityRunId(), invoker);
   }
 
@@ -410,6 +415,18 @@ public <R> ActivityHandle<R> getHandle(
     return ActivityHandle.fromUntyped(untyped, resultClass, resultType);
   }
 
+  private Header propagatedHeader() {
+    List<ContextPropagator> propagators = options.getContextPropagators();
+    if (propagators.isEmpty()) {
+      return Header.empty();
+    }
+    Map<String, Payload> result = new HashMap<>();
+    for (ContextPropagator propagator : propagators) {
+      result.putAll(propagator.serializeContext(propagator.getCurrentContext()));
+    }
+    return new Header(result);
+  }
+
   // ---- List / count ----
 
   @Override

temporal-sdk/src/test/java/io/temporal/client/ActivityClientImplTest.java

@@ -1,15 +1,24 @@
 package io.temporal.client;
 
+import static org.junit.Assert.*;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.*;
 
 import com.uber.m3.tally.Scope;
 import io.temporal.activity.ActivityInterface;
 import io.temporal.activity.ActivityMethod;
+import io.temporal.api.common.v1.Payload;
+import io.temporal.common.context.ContextPropagator;
+import io.temporal.common.interceptors.ActivityClientCallsInterceptor;
+import io.temporal.common.interceptors.ActivityClientCallsInterceptorBase;
+import io.temporal.common.interceptors.ActivityClientInterceptor;
+import io.temporal.common.interceptors.Header;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import io.temporal.serviceclient.WorkflowServiceStubsOptions;
 import io.temporal.workflow.Functions;
 import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicReference;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -27,12 +36,13 @@ public interface WrongActivity {
     void doWrong();
   }
 
+  private WorkflowServiceStubs stubs;
   private ActivityClient client;
   private StartActivityOptions options;
 
   @Before
   public void setUp() {
-    WorkflowServiceStubs stubs = mock(WorkflowServiceStubs.class);
+    stubs = mock(WorkflowServiceStubs.class);
     WorkflowServiceStubsOptions stubsOptions = mock(WorkflowServiceStubsOptions.class);
     Scope scope = mock(Scope.class);
     when(stubs.getOptions()).thenReturn(stubsOptions);
@@ -48,6 +58,39 @@ public void setUp() {
             .build();
   }
 
+  @Test
+  public void testContextPropagatorHeaderIsIncluded() {
+    Payload payload = Payload.newBuilder().build();
+    ContextPropagator propagator = mock(ContextPropagator.class);
+    when(propagator.getCurrentContext()).thenReturn("ctx");
+    when(propagator.serializeContext("ctx"))
+        .thenReturn(Collections.singletonMap("my-key", payload));
+
+    AtomicReference<Header> capturedHeader = new AtomicReference<>();
+    ActivityClientInterceptor capturingInterceptor =
+        next ->
+            new ActivityClientCallsInterceptorBase(next) {
+              @Override
+              public ActivityClientCallsInterceptor.StartActivityOutput startActivity(
+                  ActivityClientCallsInterceptor.StartActivityInput input) {
+                capturedHeader.set(input.getHeader());
+                return new ActivityClientCallsInterceptor.StartActivityOutput("fake-id", null);
+              }
+            };
+
+    ActivityClientOptions opts =
+        ActivityClientOptions.newBuilder()
+            .setContextPropagators(Collections.singletonList(propagator))
+            .setInterceptors(Collections.singletonList(capturingInterceptor))
+            .build();
+
+    ActivityClient clientWithPropagators = ActivityClient.newInstance(stubs, opts);
+    clientWithPropagators.start(ProbedActivity.class, ProbedActivity::doIt, options);
+
+    assertNotNull(capturedHeader.get());
+    assertEquals(payload, capturedHeader.get().getValues().get("my-key"));
+  }
+
   @Test(expected = NoSuchMethodError.class)
   @SuppressWarnings("unchecked")
   public void testStartWithMethodFromWrongClass() {