[Backport release-1.19] fix: Forward caller headers in service invocation without duplicates or restricted-header errors

sdk/src/main/java/io/dapr/client/AbstractDaprClient.java

@@ -127,7 +127,6 @@ public <T> Mono<T> invokeMethod(
     InvokeMethodRequest req = new InvokeMethodRequest(appId, methodName)
         .setBody(data)
         .setHttpExtension(httpExtension)
-        .setContentType(objectSerializer.getContentType())
         .setMetadata(metadata);
 
     return this.invokeMethod(req, type);

sdk/src/main/java/io/dapr/client/DaprClientImpl.java

@@ -143,6 +143,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.TreeMap;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -692,7 +693,7 @@ public <T> Mono<T> invokeMethod(InvokeMethodRequest invokeMethodRequest, TypeRef
       List<String> pathSegments = new ArrayList<>(Arrays.asList(DaprHttp.API_VERSION, "invoke", appId, "method"));
       pathSegments.addAll(Arrays.asList(methodSegments));
 
-      final Map<String, String> headers = new HashMap<>();
+      final Map<String, String> headers = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
       headers.putAll(httpExtension.getHeaders());
       if (metadata != null) {
         headers.putAll(metadata);
@@ -701,7 +702,7 @@ public <T> Mono<T> invokeMethod(InvokeMethodRequest invokeMethodRequest, TypeRef
       if (contentType != null && !contentType.isEmpty()) {
         headers.put(io.dapr.client.domain.Metadata.CONTENT_TYPE, contentType);
       } else {
-        headers.put(io.dapr.client.domain.Metadata.CONTENT_TYPE, objectSerializer.getContentType());
+        headers.putIfAbsent(io.dapr.client.domain.Metadata.CONTENT_TYPE, objectSerializer.getContentType());
       }
       Mono<DaprHttp.Response> response = Mono.deferContextual(
           context -> this.httpClient.invokeApi(httpMethod, pathSegments.toArray(new String[0]),

sdk/src/main/java/io/dapr/client/DaprHttp.java

@@ -38,6 +38,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.TreeMap;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 
@@ -69,6 +70,19 @@ public class DaprHttp implements AutoCloseable {
   private static final Set<String> ALLOWED_CONTEXT_IN_HEADERS =
       Set.of("grpc-trace-bin", "traceparent", "tracestate", "baggage");
 
+  /**
+   * Hop-by-hop headers managed by {@link HttpClient} itself, which rejects them when set explicitly.
+   * They are dropped so headers received from an inbound request can be forwarded as-is.
+   */
+  private static final Set<String> RESTRICTED_HEADERS =
+      Set.of("connection", "content-length", "expect", "host", "upgrade");
+
+  /**
+   * Headers owned by the SDK that callers cannot override.
+   */
+  private static final Set<String> SDK_OWNED_HEADERS =
+      Set.of(Headers.DAPR_USER_AGENT.toLowerCase(), Headers.DAPR_API_TOKEN);
+
   /**
    * Object mapper to parse DaprError with or without details.
    */
@@ -298,20 +312,7 @@ private CompletableFuture<Response> doInvokeApi(
     HttpRequest.Builder requestBuilder = HttpRequest.newBuilder();
 
     requestBuilder.uri(createUri(uri, pathSegments, urlParameters));
-    addHeader(requestBuilder, Headers.DAPR_USER_AGENT, Version.getSdkVersion());
-    addHeader(requestBuilder, HEADER_DAPR_REQUEST_ID, UUID.randomUUID().toString());
-    addHeader(requestBuilder, "Content-Type", getContentType(headers));
-    addHeaders(requestBuilder, headers);
-
-    if (daprApiToken != null) {
-      addHeader(requestBuilder, Headers.DAPR_API_TOKEN, daprApiToken);
-    }
-
-    if (context != null) {
-      context.stream()
-          .filter(entry -> ALLOWED_CONTEXT_IN_HEADERS.contains(entry.getKey().toString().toLowerCase()))
-          .forEach(entry -> addHeader(requestBuilder, entry.getKey().toString(), entry.getValue().toString()));
-    }
+    buildHeaders(headers, context).forEach(requestBuilder::header);
 
     HttpRequest.BodyPublisher body = getBodyPublisher(content);
 
@@ -332,10 +333,41 @@ private CompletableFuture<Response> doInvokeApi(
         .thenApply(this::createResponse);
   }
 
-  private static String getContentType(Map<String, String> headers) {
-    String result = headers != null ? headers.get(Metadata.CONTENT_TYPE) : null;
+  /**
+   * Merges SDK, tracing and caller headers into a single set, matching names case-insensitively so each
+   * header is sent once. Precedence (lowest to highest): SDK defaults, tracing context, caller headers,
+   * SDK-owned headers. Hop-by-hop headers rejected by {@link HttpClient} are dropped.
+   */
+  private Map<String, String> buildHeaders(Map<String, String> headers, ContextView context) {
+    Map<String, String> result = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+    result.put(HEADER_DAPR_REQUEST_ID, UUID.randomUUID().toString());
+    result.put(Metadata.CONTENT_TYPE, MEDIA_TYPE_APPLICATION_JSON);
 
-    return result == null ? MEDIA_TYPE_APPLICATION_JSON : result;
+    if (context != null) {
+      context.stream()
+          .filter(entry -> ALLOWED_CONTEXT_IN_HEADERS.contains(entry.getKey().toString().toLowerCase()))
+          .forEach(entry -> result.put(entry.getKey().toString(), entry.getValue().toString()));
+    }
+
+    if (headers != null) {
+      headers.forEach((name, value) -> {
+        if (name == null || value == null) {
+          return;
+        }
+
+        String key = name.toLowerCase();
+        if (!RESTRICTED_HEADERS.contains(key) && !SDK_OWNED_HEADERS.contains(key)) {
+          result.put(name, value);
+        }
+      });
+    }
+
+    result.put(Headers.DAPR_USER_AGENT, Version.getSdkVersion());
+    if (daprApiToken != null) {
+      result.put(Headers.DAPR_API_TOKEN, daprApiToken);
+    }
+
+    return result;
   }
 
   private static URI createUri(URI uri, String[] pathSegments, Map<String, List<String>> urlParameters) {
@@ -418,18 +450,6 @@ private static String encodeQueryParam(String key, String value) {
     return URLEncoder.encode(key, StandardCharsets.UTF_8) + "=" + URLEncoder.encode(value, StandardCharsets.UTF_8);
   }
 
-  private static void addHeader(HttpRequest.Builder requestBuilder, String name, String value) {
-    requestBuilder.header(name, value);
-  }
-
-  private static void addHeaders(HttpRequest.Builder requestBuilder, Map<String, String> headers) {
-    if (headers == null || headers.isEmpty()) {
-      return;
-    }
-
-    headers.forEach((k, v) -> addHeader(requestBuilder, k, v));
-  }
-
   private static HttpRequest.BodyPublisher getBodyPublisher(byte[] content) {
     return HttpRequest.BodyPublishers.ofByteArray(Objects.requireNonNullElse(content, EMPTY_BYTES));
   }

sdk/src/test/java/io/dapr/client/DaprClientHttpTest.java

@@ -46,6 +46,7 @@
 import static io.dapr.utils.TestUtils.findFreePort;
 import static io.dapr.utils.TestUtils.formatIpAddress;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
@@ -656,6 +657,95 @@ public void invokeServiceWithContext() {
     assertEquals(tracestate, request.headers().firstValue("tracestate").get());
   }
 
+  @Test
+  public void invokeServiceForwardsHttpExtensionHeadersAndMetadata() {
+    HttpRequest request = invokeAndCaptureRequest(
+        new HttpExtension(DaprHttp.HttpMethods.GET, null, Map.of("aaa", "96969")),
+        Map.of("bbb", "123"));
+
+    assertEquals(List.of("96969"), request.headers().allValues("aaa"));
+    assertEquals(List.of("123"), request.headers().allValues("bbb"));
+  }
+
+  @Test
+  public void invokeServiceDropsHopByHopHeadersWhenForwardingInboundHeaders() {
+    Map<String, String> inboundHeaders = Map.of(
+        "Host", "127.0.0.1:3000",
+        "Connection", "Keep-Alive",
+        "Content-Length", "13",
+        "x-forwarded-host", "test.local");
+
+    HttpRequest request = invokeAndCaptureRequest(
+        new HttpExtension(DaprHttp.HttpMethods.POST, null, inboundHeaders), inboundHeaders);
+
+    assertEquals(List.of("test.local"), request.headers().allValues("x-forwarded-host"));
+    assertEquals(List.of(), request.headers().allValues("host"));
+    assertEquals(List.of(), request.headers().allValues("connection"));
+    assertEquals(List.of(), request.headers().allValues("content-length"));
+  }
+
+  @Test
+  public void invokeServiceSendsCallerContentTypeOnce() {
+    HttpRequest request = invokeAndCaptureRequest(
+        new HttpExtension(DaprHttp.HttpMethods.POST, null, Map.of("Content-Type", "text/plain")), null);
+
+    assertEquals(List.of("text/plain"), request.headers().allValues("content-type"));
+  }
+
+  @Test
+  public void invokeServiceDefaultsContentTypeToSerializer() {
+    HttpRequest request = invokeAndCaptureRequest(HttpExtension.POST, null);
+
+    assertEquals(List.of("application/json"), request.headers().allValues("content-type"));
+  }
+
+  @Test
+  public void invokeServiceExplicitHeaderOverridesTracingContextOnce() {
+    String contextTraceparent = "00-0af7651916cd43dd8448eb211c80319c-b9c7c989f97918e1-01";
+    String inboundTraceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01";
+    HttpRequest request = invokeAndCaptureRequest(
+        new HttpExtension(DaprHttp.HttpMethods.POST, null, Map.of("Traceparent", inboundTraceparent)),
+        null,
+        Context.of("traceparent", contextTraceparent));
+
+    assertEquals(List.of(inboundTraceparent), request.headers().allValues("traceparent"));
+  }
+
+  @Test
+  public void invokeServiceDoesNotLetCallerOverrideSdkOwnedHeaders() {
+    DaprHttp tokenedDaprHttp = new DaprHttp(sidecarIp, 3000, "xyz", READ_TIMEOUT, httpClient);
+    daprClientHttp = buildDaprClient(tokenedDaprHttp);
+
+    HttpRequest request = invokeAndCaptureRequest(
+        new HttpExtension(DaprHttp.HttpMethods.POST, null,
+            Map.of(Headers.DAPR_API_TOKEN, "forged", "user-agent", "curl/8.0")),
+        null);
+
+    assertEquals(List.of("xyz"), request.headers().allValues(Headers.DAPR_API_TOKEN));
+    assertEquals(1, request.headers().allValues(Headers.DAPR_USER_AGENT).size());
+    assertNotEquals("curl/8.0", request.headers().firstValue(Headers.DAPR_USER_AGENT).orElse(null));
+  }
+
+  private HttpRequest invokeAndCaptureRequest(HttpExtension httpExtension, Map<String, String> metadata) {
+    return invokeAndCaptureRequest(httpExtension, metadata, Context.empty());
+  }
+
+  private HttpRequest invokeAndCaptureRequest(
+      HttpExtension httpExtension, Map<String, String> metadata, ContextView context) {
+    MockHttpResponse mockHttpResponse = new MockHttpResponse(new byte[0], HTTP_OK);
+    CompletableFuture<HttpResponse<Object>> mockResponse = CompletableFuture.completedFuture(mockHttpResponse);
+    ArgumentCaptor<HttpRequest> requestCaptor = ArgumentCaptor.forClass(HttpRequest.class);
+
+    when(httpClient.sendAsync(any(), any())).thenReturn(mockResponse);
+
+    daprClientHttp.invokeMethod("41", "neworder", "", httpExtension, metadata, byte[].class)
+        .contextWrite(it -> it.putAll(context))
+        .block();
+
+    verify(httpClient).sendAsync(requestCaptor.capture(), any());
+    return requestCaptor.getValue();
+  }
+
   @Test
   public void closeException() {
     DaprHttp daprHttp = Mockito.mock(DaprHttp.class);

sdk/src/test/java/io/dapr/client/DaprHttpTest.java

@@ -32,6 +32,7 @@
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.time.Duration;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -108,6 +109,29 @@ public void invokeApi_daprApiToken_present() throws IOException {
     assertEquals("xyz", request.headers().firstValue(Headers.DAPR_API_TOKEN).get());
   }
 
+  @Test
+  public void invokeApi_skipsHeadersWithNullNameOrValue() {
+    MockHttpResponse mockHttpResponse = new MockHttpResponse(new byte[0], HTTP_OK);
+    CompletableFuture<HttpResponse<Object>> mockResponse = CompletableFuture.completedFuture(mockHttpResponse);
+    ArgumentCaptor<HttpRequest> requestCaptor = ArgumentCaptor.forClass(HttpRequest.class);
+    Map<String, String> headers = new HashMap<>();
+    headers.put(null, "orphan-value");
+    headers.put("x-tenant-id", null);
+    headers.put("x-correlation-id", "order-1234");
+
+    when(httpClient.sendAsync(any(), any())).thenReturn(mockResponse);
+
+    DaprHttp daprHttp = new DaprHttp(sidecarIp, 3500, daprTokenApi, READ_TIMEOUT, httpClient);
+    daprHttp.invokeApi("POST", "v1.0/state".split("/"), null, (byte[]) null, headers, Context.empty()).block();
+
+    verify(httpClient).sendAsync(requestCaptor.capture(), any());
+
+    HttpRequest request = requestCaptor.getValue();
+
+    assertEquals(List.of("order-1234"), request.headers().allValues("x-correlation-id"));
+    assertEquals(List.of(), request.headers().allValues("x-tenant-id"));
+  }
+
   @Test
   public void invokeApi_daprApiToken_absent() throws IOException {
     byte[] content = serializer.serialize(EXPECTED_RESULT);