Merge branch 'master' into actor_ttl

Author: artursouza

pom.xml

@@ -15,7 +15,7 @@
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <grpc.version>1.64.0</grpc.version>
-    <protobuf.version>3.25.0</protobuf.version>
+    <protobuf.version>3.25.5</protobuf.version>
     <protocCommand>protoc</protocCommand>
     <dapr.proto.baseurl>https://raw.githubusercontent.com/dapr/dapr/v1.14.4/dapr/proto</dapr.proto.baseurl>
     <dapr.sdk.version>1.13.0-SNAPSHOT</dapr.sdk.version>

sdk-actors/src/main/java/io/dapr/actors/client/ActorClient.java

@@ -22,6 +22,9 @@
 import io.grpc.ManagedChannelBuilder;
 import reactor.core.publisher.Mono;
 
+import java.util.Collections;
+import java.util.Map;
+
 /**
  * Holds a client for Dapr sidecar communication. ActorClient should be reused.
  */
@@ -59,7 +62,7 @@ public ActorClient(ResiliencyOptions resiliencyOptions) {
    * @param overrideProperties Override properties.
    */
   public ActorClient(Properties overrideProperties) {
-    this(buildManagedChannel(overrideProperties), null, overrideProperties.getValue(Properties.API_TOKEN));
+    this(overrideProperties, null);
   }
 
   /**
@@ -69,21 +72,38 @@ public ActorClient(Properties overrideProperties) {
    * @param resiliencyOptions Client resiliency options.
    */
   public ActorClient(Properties overrideProperties, ResiliencyOptions resiliencyOptions) {
-    this(buildManagedChannel(overrideProperties), resiliencyOptions, overrideProperties.getValue(Properties.API_TOKEN));
+    this(overrideProperties, null, resiliencyOptions);
+  }
+
+  /**
+   * Instantiates a new channel for Dapr sidecar communication.
+   *
+   * @param overrideProperties Override properties.
+   * @param metadata gRPC metadata or HTTP headers for actor invocation.
+   * @param resiliencyOptions Client resiliency options.
+   */
+  public ActorClient(Properties overrideProperties, Map<String, String> metadata, ResiliencyOptions resiliencyOptions) {
+    this(buildManagedChannel(overrideProperties),
+        metadata,
+        resiliencyOptions,
+        overrideProperties.getValue(Properties.API_TOKEN));
   }
 
   /**
    * Instantiates a new channel for Dapr sidecar communication.
    *
    * @param grpcManagedChannel gRPC channel.
+   * @param metadata gRPC metadata or HTTP headers for actor invocation.
    * @param resiliencyOptions Client resiliency options.
+   * @param daprApiToken Dapr API token.
    */
   private ActorClient(
       ManagedChannel grpcManagedChannel,
+      Map<String, String> metadata,
       ResiliencyOptions resiliencyOptions,
       String daprApiToken) {
     this.grpcManagedChannel = grpcManagedChannel;
-    this.daprClient = buildDaprClient(grpcManagedChannel, resiliencyOptions, daprApiToken);
+    this.daprClient = buildDaprClient(grpcManagedChannel, metadata, resiliencyOptions, daprApiToken);
   }
 
   /**
@@ -137,10 +157,12 @@ private static ManagedChannel buildManagedChannel(Properties overrideProperties)
    */
   private static DaprClient buildDaprClient(
       Channel grpcManagedChannel,
+      Map<String, String> metadata,
       ResiliencyOptions resiliencyOptions,
       String daprApiToken) {
     return new DaprClientImpl(
         DaprGrpc.newStub(grpcManagedChannel),
+        metadata == null ? null : Collections.unmodifiableMap(metadata),
         resiliencyOptions,
         daprApiToken);
   }

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

@@ -34,6 +34,7 @@
 import reactor.core.publisher.MonoSink;
 import reactor.util.context.ContextView;
 
+import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Consumer;
 
@@ -57,19 +58,30 @@ class DaprClientImpl implements DaprClient {
    */
   private final DaprClientGrpcInterceptors grpcInterceptors;
 
+  /**
+   * Metadata for actor invocation requests.
+   */
+  private final Map<String, String> metadata;
+
   /**
    * Internal constructor.
    *
    * @param grpcClient Dapr's GRPC client.
+   * @param metadata gRPC metadata or HTTP headers for actor server to receive.
    * @param resiliencyOptions Client resiliency options (optional).
    * @param daprApiToken Dapr API token (optional).
    */
-  DaprClientImpl(DaprGrpc.DaprStub grpcClient, ResiliencyOptions resiliencyOptions, String daprApiToken) {
+  DaprClientImpl(
+      DaprGrpc.DaprStub grpcClient,
+      Map<String, String> metadata,
+      ResiliencyOptions resiliencyOptions,
+      String daprApiToken) {
     this.client = grpcClient;
     this.grpcInterceptors = new DaprClientGrpcInterceptors(daprApiToken,
         new TimeoutPolicy(resiliencyOptions == null ? null : resiliencyOptions.getTimeout()));
     this.retryPolicy = new RetryPolicy(
         resiliencyOptions == null ? null : resiliencyOptions.getMaxRetries());
+    this.metadata = metadata == null ? Map.of() : metadata;
   }
 
   /**
@@ -82,6 +94,7 @@ public Mono<byte[]> invoke(String actorType, String actorId, String methodName,
             .setActorType(actorType)
             .setActorId(actorId)
             .setMethod(methodName)
+            .putAllMetadata(this.metadata)
             .setData(jsonPayload == null ? ByteString.EMPTY : ByteString.copyFrom(jsonPayload))
             .build();
     return Mono.deferContextual(

sdk-actors/src/test/java/io/dapr/actors/client/DaprGrpcClientTest.java

@@ -106,7 +106,7 @@ public void setup() throws IOException {
         InProcessChannelBuilder.forName(serverName).directExecutor().build());
 
     // Create a HelloWorldClient using the in-process channel;
-    client = new DaprClientImpl(DaprGrpc.newStub(channel), null, null);
+    client = new DaprClientImpl(DaprGrpc.newStub(channel), null, null, null);
   }
 
   @Test

sdk-tests/src/test/java/io/dapr/it/DaprRun.java

@@ -187,11 +187,19 @@ public DaprClientBuilder newDaprClientBuilder() {
   }
 
   public ActorClient newActorClient() {
-    return this.newActorClient(null);
+    return this.newActorClient(null, null);
+  }
+
+  public ActorClient newActorClient(Map<String, String> metadata) {
+    return this.newActorClient(metadata, null);
   }
 
   public ActorClient newActorClient(ResiliencyOptions resiliencyOptions) {
-    return new ActorClient(new Properties(this.getPropertyOverrides()), resiliencyOptions);
+    return this.newActorClient(null, resiliencyOptions);
+  }
+
+  public ActorClient newActorClient(Map<String, String> metadata, ResiliencyOptions resiliencyOptions) {
+    return new ActorClient(new Properties(this.getPropertyOverrides()), metadata, resiliencyOptions);
   }
 
   public void waitForAppHealth(int maxWaitMilliseconds) throws InterruptedException {

sdk-tests/src/test/java/io/dapr/it/actors/ActorExceptionIT.java

@@ -16,12 +16,17 @@
 import io.dapr.actors.ActorId;
 import io.dapr.actors.client.ActorProxyBuilder;
 import io.dapr.it.BaseIT;
+import io.dapr.it.DaprRun;
 import io.dapr.it.actors.app.MyActor;
 import io.dapr.it.actors.app.MyActorService;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Map;
+
 import static io.dapr.it.Retry.callWithRetry;
 import static io.dapr.it.TestUtils.assertThrowsDaprExceptionSubstring;
 
@@ -30,23 +35,24 @@ public class ActorExceptionIT extends BaseIT {
 
   private static Logger logger = LoggerFactory.getLogger(ActorExceptionIT.class);
 
-  @Test
-  public void exceptionTest() throws Exception {
+  private static DaprRun run;
+
+  @BeforeAll
+  public static void start() throws Exception {
     // The call below will fail if service cannot start successfully.
-    var run = startDaprApp(
+    run = startDaprApp(
         ActorExceptionIT.class.getSimpleName(),
         MyActorService.SUCCESS_MESSAGE,
         MyActorService.class,
         true,
         60000);
+  }
 
-    logger.debug("Creating proxy builder");
+  @Test
+  public void exceptionTest() throws Exception {
     ActorProxyBuilder<MyActor> proxyBuilder =
         new ActorProxyBuilder("MyActorTest", MyActor.class, deferClose(run.newActorClient()));
-    logger.debug("Creating actorId");
-    ActorId actorId1 = new ActorId("1");
-    logger.debug("Building proxy");
-    MyActor proxy = proxyBuilder.build(actorId1);
+    MyActor proxy = proxyBuilder.build(new ActorId("1"));
 
     callWithRetry(() -> {
       assertThrowsDaprExceptionSubstring(
@@ -55,4 +61,20 @@ public void exceptionTest() throws Exception {
           () ->  proxy.throwException());
     }, 10000);
   }
+
+  @Test
+  public void exceptionDueToMetadataTest() throws Exception {
+    // Setting this HTTP header via actor metadata will cause the Actor HTTP server to error.
+    Map<String, String> metadata = Map.of("Content-Length", "9999");
+    ActorProxyBuilder<MyActor> proxyBuilderMetadataOverride =
+        new ActorProxyBuilder("MyActorTest", MyActor.class, deferClose(run.newActorClient(metadata)));
+
+    MyActor proxyWithMetadata = proxyBuilderMetadataOverride.build(new ActorId("2"));
+    callWithRetry(() -> {
+      assertThrowsDaprExceptionSubstring(
+          "INTERNAL",
+          "ContentLength=9999 with Body length 13",
+          () -> proxyWithMetadata.say("hello world"));
+    }, 10000);
+  }
 }

sdk/pom.xml

@@ -52,7 +52,7 @@
     <dependency>
       <groupId>com.squareup.okhttp3</groupId>
       <artifactId>okhttp</artifactId>
-      <version>4.9.0</version>
+      <version>4.9.2</version>
     </dependency>
     <dependency>
       <groupId>org.mockito</groupId>

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

@@ -270,6 +270,14 @@ public <T> Mono<T> invokeBinding(
     return this.invokeBinding(bindingName, operation, data, metadata, TypeRef.get(clazz));
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public Mono<Void> invokeBinding(InvokeBindingRequest request) {
+    return this.invokeBinding(request, TypeRef.VOID);
+  }
+
   /**
    * {@inheritDoc}
    */

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

@@ -314,6 +314,14 @@ <T> Mono<T> invokeBinding(String bindingName, String operation, Object data, Map
   <T> Mono<T> invokeBinding(String bindingName, String operation, Object data, Map<String, String> metadata,
                             Class<T> clazz);
 
+  /**
+   * Invokes a Binding operation.
+   *
+   * @param request The binding invocation request.
+   * @return a Mono with void.
+   */
+  Mono<Void> invokeBinding(InvokeBindingRequest request);
+
   /**
    * Invokes a Binding operation.
    *

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

@@ -459,8 +459,10 @@ private <T> Subscription<T> buildSubscription(
 
       try {
         CloudEvent<T> cloudEvent = new CloudEvent<>();
-        var object =
-            DaprClientImpl.this.objectSerializer.deserialize(message.getData().toByteArray(), type);
+        T object = null;
+        if (type != null) {
+          object = DaprClientImpl.this.objectSerializer.deserialize(message.getData().toByteArray(), type);
+        }
         cloudEvent.setData(object);
         cloudEvent.setDatacontenttype(message.getDataContentType());
         cloudEvent.setId(message.getId());
@@ -528,6 +530,10 @@ public <T> Mono<T> invokeMethod(InvokeMethodRequest invokeMethodRequest, TypeRef
 
   private <T> Mono<T> getMonoForHttpResponse(TypeRef<T> type, DaprHttp.Response r) {
     try {
+      if (type == null) {
+        return Mono.empty();
+      }
+
       T object = objectSerializer.deserialize(r.getBody(), type);
       if (object == null) {
         return Mono.empty();
@@ -585,6 +591,9 @@ public <T> Mono<T> invokeBinding(InvokeBindingRequest request, TypeRef<T> type)
             }
 
             try {
+              if (type == null) {
+                return Mono.empty();
+              }
               return Mono.justOrEmpty(objectSerializer.deserialize(it.getData().toByteArray(), type));
             } catch (IOException e) {
               throw DaprException.propagate(e);
@@ -706,13 +715,18 @@ private <T> State<T> buildStateKeyValue(
       return new State<>(key, error);
     }
 
-    ByteString payload = item.getData();
-    byte[] data = payload == null ? null : payload.toByteArray();
-    T value = stateSerializer.deserialize(data, type);
     String etag = item.getEtag();
     if (etag.equals("")) {
       etag = null;
     }
+
+    T value = null;
+    if (type != null) {
+      ByteString payload = item.getData();
+      byte[] data = payload == null ? null : payload.toByteArray();
+      value = stateSerializer.deserialize(data, type);
+    }
+
     return new State<>(key, value, etag, item.getMetadataMap(), null);
   }
 
@@ -723,7 +737,11 @@ private <T> State<T> buildStateKeyValue(
       TypeRef<T> type) throws IOException {
     ByteString payload = response.getData();
     byte[] data = payload == null ? null : payload.toByteArray();
-    T value = stateSerializer.deserialize(data, type);
+    T value = null;
+    if (type != null) {
+      value = stateSerializer.deserialize(data, type);
+    }
+
     String etag = response.getEtag();
     if (etag.equals("")) {
       etag = null;
@@ -1108,7 +1126,11 @@ private <T> QueryStateItem<T> buildQueryStateKeyValue(
     }
     ByteString payload = item.getData();
     byte[] data = payload == null ? null : payload.toByteArray();
-    T value = stateSerializer.deserialize(data, type);
+    T value = null;
+    if (type != null) {
+      value = stateSerializer.deserialize(data, type);
+    }
+
     String etag = item.getEtag();
     if (etag.equals("")) {
       etag = null;