Fix issue where custom JSOn serializer generates Base64 encoded data in state store

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

@@ -9,7 +9,6 @@
 import io.dapr.actors.ActorId;
 import io.dapr.config.Properties;
 import io.dapr.serializer.DaprObjectSerializer;
-import io.dapr.serializer.DefaultObjectSerializer;
 import io.dapr.utils.TypeRef;
 import reactor.core.publisher.Mono;
 
@@ -27,6 +26,11 @@ class DaprStateAsyncProvider {
    */
   private static final Charset CHARSET = Properties.STRING_CHARSET.get();
 
+  /**
+   * Marker to identify Json serializers.
+   */
+  public static final String JSON_CONTENT_TYPE = "application/json";
+
   /**
    * Handles special serialization cases.
    */
@@ -45,7 +49,7 @@ class DaprStateAsyncProvider {
   /**
    * Flag determining if state serializer is the default serializer instead of user provided.
    */
-  private final boolean isStateSerializerDefault;
+  private final boolean isStateSerializerJson;
 
   /**
    * Instantiates a new Actor's state provider.
@@ -56,7 +60,7 @@ class DaprStateAsyncProvider {
   DaprStateAsyncProvider(DaprClient daprClient, DaprObjectSerializer stateSerializer) {
     this.daprClient = daprClient;
     this.stateSerializer = stateSerializer;
-    this.isStateSerializerDefault = stateSerializer.getClass() == DefaultObjectSerializer.class;
+    this.isStateSerializerJson = JSON_CONTENT_TYPE.equals(stateSerializer.getContentType());
   }
 
   <T> Mono<T> load(String actorType, ActorId actorId, String stateName, TypeRef<T> type) {
@@ -69,7 +73,7 @@ <T> Mono<T> load(String actorType, ActorId actorId, String stateName, TypeRef<T>
         }
 
         T response = this.stateSerializer.deserialize(s, type);
-        if (this.isStateSerializerDefault && (response instanceof byte[])) {
+        if (this.isStateSerializerJson && (response instanceof byte[])) {
           if (s.length == 0) {
             return Mono.empty();
           }
@@ -138,7 +142,7 @@ Mono<Void> apply(String actorType, ActorId actorId, ActorStateChange... stateCha
         try {
           byte[] data = this.stateSerializer.serialize(stateChange.getValue());
           if (data != null) {
-            if (this.isStateSerializerDefault && !(stateChange.getValue() instanceof byte[])) {
+            if (this.isStateSerializerJson && !(stateChange.getValue() instanceof byte[])) {
               // DefaultObjectSerializer is a JSON serializer, so we just pass it on.
               value = new String(data, CHARSET);
             } else {

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

@@ -7,13 +7,15 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.dapr.actors.ActorId;
+import io.dapr.client.ObjectSerializer;
 import io.dapr.serializer.DaprObjectSerializer;
 import io.dapr.serializer.DefaultObjectSerializer;
 import io.dapr.utils.TypeRef;
 import org.junit.Assert;
 import org.junit.Test;
 import reactor.core.publisher.Mono;
 
+import java.io.IOException;
 import java.util.Arrays;
 import java.util.Objects;
 
@@ -27,10 +29,29 @@ public class DaprStateAsyncProviderTest {
 
   private static final DaprObjectSerializer SERIALIZER = new DefaultObjectSerializer();
 
-  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
 
   private static final double EPSILON = 1e-10;
 
+  class CustomJsonSerializer implements DaprObjectSerializer{
+    private final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    @Override
+    public byte[] serialize(Object o) throws IOException {
+      return OBJECT_MAPPER.writeValueAsBytes(o);
+    }
+
+    @Override
+    public <T> T deserialize(byte[] data, TypeRef<T> type) throws IOException {
+      return OBJECT_MAPPER.readValue(data, OBJECT_MAPPER.constructType(type.getType()));
+    }
+
+    @Override
+    public String getContentType() {
+      return "application/json";
+    }
+  }
+
   /**
    * Class used to test JSON serialization.
    */
@@ -136,6 +157,49 @@ public void happyCaseApply() {
     verify(daprClient).saveStateTransactionally(eq("MyActor"), eq("123"), any());
   }
 
+  @Test
+  public void happyCaseApplyWithCustomJsonSerializer() {
+    DaprClient daprClient = mock(DaprClient.class);
+    when(daprClient
+            .saveStateTransactionally(
+                    eq("MyActor"),
+                    eq("123"),
+                    argThat(operations -> {
+                      if (operations == null) {
+                        return false;
+                      }
+
+                      if (operations.size() != 1) {
+                        return false;
+                      }
+                      ActorStateOperation operation = operations.get(0);
+                        if (operation.getOperationType() == null) {
+                          return false;
+                        }
+                        if (operation.getKey() == null) {
+                          return false;
+                        }
+
+                        String opName = operation.getOperationType();
+                        String key = operation.getKey();
+                        Object value = operation.getValue();
+
+                        return "upsert".equals(opName) &&
+                                "object".equals(key) &&
+                                "{\"id\":1000,\"name\":\"Roxane\"}".equals(value);
+                    })))
+            .thenReturn(Mono.empty());
+
+    DaprStateAsyncProvider provider = new DaprStateAsyncProvider(daprClient, new CustomJsonSerializer());
+    provider.apply("MyActor",
+            new ActorId("123"),
+            createInsertChange("object", new Customer().setId(1000).setName("Roxane")))
+            .block();
+
+    verify(daprClient).saveStateTransactionally(eq("MyActor"), eq("123"), any());
+  }
+
+
   @Test
   public void happyCaseLoad() throws Exception {
     DaprClient daprClient = mock(DaprClient.class);