Add support for configuring actor reminder storage partitions

pom.xml

@@ -14,7 +14,7 @@
 
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    <grpc.version>1.33.1</grpc.version>
+    <grpc.version>1.39.0</grpc.version>
     <protobuf.version>3.13.0</protobuf.version>
     <dapr.proto.baseurl>https://raw.githubusercontent.com/dapr/dapr/v1.2.0-rc.3/dapr/proto</dapr.proto.baseurl>
     <os-maven-plugin.version>1.6.2</os-maven-plugin.version>

sdk-actors/pom.xml

@@ -74,6 +74,7 @@
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-testing</artifactId>
+      <version>1.39.0</version>
       <scope>test</scope>
     </dependency>
   </dependencies>

sdk-actors/src/main/java/io/dapr/actors/runtime/ActorObjectSerializer.java

@@ -53,7 +53,6 @@ public byte[] serialize(Object state) throws IOException {
     return super.serialize(state);
   }
 
-
   /**
    * Faster serialization for params of Actor's timer.
    *
@@ -136,6 +135,9 @@ private byte[] serialize(ActorRuntimeConfig config) throws IOException {
       if (config.getDrainBalancedActors() != null) {
         generator.writeBooleanField("drainBalancedActors", config.getDrainBalancedActors());
       }
+      if (config.getRemindersStoragePartitions() != null) {
+        generator.writeNumberField("remindersStoragePartitions", config.getRemindersStoragePartitions());
+      }
       generator.writeEndObject();
       generator.close();
       writer.flush();
@@ -213,7 +215,7 @@ private ActorReminderParams deserializeActorReminder(byte[] value) throws IOExce
   private static Duration extractDurationOrNull(JsonNode node, String name) {
     JsonNode valueNode = node.get(name);
     if (valueNode == null) {
-      return  null;
+      return null;
     }
 
     return DurationUtils.convertDurationFromDaprFormat(valueNode.asText());

sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntimeConfig.java

@@ -26,6 +26,8 @@ public class ActorRuntimeConfig {
 
   private volatile Boolean drainBalancedActors;
 
+  private volatile Integer remindersStoragePartitions;
+
   /**
    * Instantiates a new config for the Actor Runtime.
    */
@@ -34,6 +36,7 @@ public class ActorRuntimeConfig {
 
   /**
    * Adds a registered actor to the list of registered actors.
+   * 
    * @param actorTypeName Actor type that was registered.
    * @return This instance.
    */
@@ -135,4 +138,24 @@ public ActorRuntimeConfig setDrainBalancedActors(Boolean drainBalancedActors) {
     return this;
   }
 
+  /**
+   * Gets the number of storage partitions for Actor reminders.
+   *
+   * @return The number of Actor reminder storage partitions.
+   */
+  public Integer getRemindersStoragePartitions() {
+    return remindersStoragePartitions;
+  }
+
+  /**
+   * Sets the number of storage partitions for Actor reminders.
+   *
+   * @param remindersStoragePartitions The number of storage partitions for Actor reminders.
+   * @return This instance.
+   */
+  public ActorRuntimeConfig setRemindersStoragePartitions(Integer remindersStoragePartitions) {
+    this.remindersStoragePartitions = remindersStoragePartitions;
+    return this;
+  }
+
 }

sdk-actors/src/test/java/io/dapr/actors/runtime/ActorRuntimeTest.java

@@ -27,6 +27,7 @@ public class ActorRuntimeTest {
 
   public interface MyActor {
     String say();
+
     int count();
   }
 
@@ -93,14 +94,12 @@ public int count() {
 
   @BeforeClass
   public static void beforeAll() throws Exception {
-    constructor = (Constructor<ActorRuntime>) Arrays.stream(ActorRuntime.class.getDeclaredConstructors())
-      .filter(c -> c.getParameters().length == 2)
-      .map(c -> {
-        c.setAccessible(true);
-        return c;
-      })
-      .findFirst()
-      .get();
+    constructor =
+        (Constructor<ActorRuntime>) Arrays.stream(ActorRuntime.class.getDeclaredConstructors())
+            .filter(c -> c.getParameters().length == 2).map(c -> {
+              c.setAccessible(true);
+              return c;
+            }).findFirst().get();
   }
 
   @Before
@@ -116,17 +115,20 @@ public void registerActorNullClass() {
 
   @Test(expected = IllegalArgumentException.class)
   public void registerActorNullFactory() {
-    this.runtime.registerActor(MyActorImpl.class, null, new DefaultObjectSerializer(), new DefaultObjectSerializer());
+    this.runtime.registerActor(MyActorImpl.class, null, new DefaultObjectSerializer(),
+        new DefaultObjectSerializer());
   }
 
   @Test(expected = IllegalArgumentException.class)
   public void registerActorNullSerializer() {
-    this.runtime.registerActor(MyActorImpl.class, new DefaultActorFactory<>(), null, new DefaultObjectSerializer());
+    this.runtime.registerActor(MyActorImpl.class, new DefaultActorFactory<>(), null,
+        new DefaultObjectSerializer());
   }
 
   @Test(expected = IllegalArgumentException.class)
   public void registerActorNullStateSerializer() {
-    this.runtime.registerActor(MyActorImpl.class, new DefaultActorFactory<>(), new DefaultObjectSerializer(), null);
+    this.runtime.registerActor(MyActorImpl.class, new DefaultActorFactory<>(),
+        new DefaultObjectSerializer(), null);
   }
 
   @Test
@@ -157,6 +159,13 @@ public void setDrainOngoingCallTimeout() throws Exception {
         new String(this.runtime.serializeConfig()));
   }
 
+  @Test
+  public void setRemindersStoragePartitions() throws Exception {
+    this.runtime.getConfig().setRemindersStoragePartitions(12);
+    Assert.assertEquals("{\"entities\":[],\"remindersStoragePartitions\":12}",
+        new String(this.runtime.serializeConfig()));
+  }
+
   @Test
   public void invokeActor() throws Exception {
     String actorId = UUID.randomUUID().toString();
@@ -194,10 +203,8 @@ public void lazyDeactivate() throws Exception {
     deactivateCall.block();
 
     this.runtime.invoke(ACTOR_NAME, actorId, "say", null)
-      .doOnError(e -> Assert.assertTrue(e.getMessage().contains("Could not find actor")))
-      .doOnSuccess(s -> Assert.fail())
-      .onErrorReturn("".getBytes())
-      .block();
+        .doOnError(e -> Assert.assertTrue(e.getMessage().contains("Could not find actor")))
+        .doOnSuccess(s -> Assert.fail()).onErrorReturn("".getBytes()).block();
   }
 
   @Test

sdk-autogen/pom.xml

@@ -32,26 +32,25 @@
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-netty-shaded</artifactId>
+      <version>1.39.0</version>
       <scope>runtime</scope>
     </dependency>
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-protobuf</artifactId>
+      <version>1.39.0</version>
     </dependency>
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-stub</artifactId>
+      <version>1.39.0</version>
     </dependency>
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-testing</artifactId>
+      <version>1.39.0</version>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>io.grpc</groupId>
-      <artifactId>grpc-stub</artifactId>
-      <version>1.33.1</version>
-    </dependency>
   </dependencies>
 
   <build>

sdk-tests/pom.xml

@@ -28,7 +28,7 @@
     <dapr.sdk.version>1.2.0-SNAPSHOT</dapr.sdk.version>
     <protobuf.output.directory>${project.build.directory}/generated-sources</protobuf.output.directory>
     <protobuf.input.directory>${project.basedir}/proto</protobuf.input.directory>
-    <grpc.version>1.33.1</grpc.version>
+    <grpc.version>1.39.0</grpc.version>
     <protobuf.version>3.13.0</protobuf.version>
     <opentelemetry.version>0.14.0</opentelemetry.version>
   </properties>

sdk/pom.xml

@@ -107,7 +107,7 @@
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-testing</artifactId>
-      <version>1.33.1</version>
+      <version>1.39.0</version>
       <scope>test</scope>
     </dependency>
   </dependencies>