Fixing bug in exception handling from actor impl. (#247)

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

@@ -285,8 +285,10 @@ class ActorManager<T extends AbstractActor> {
           // Actor methods must have a one or no parameter, which is guaranteed at this point.
           return method.invoke(actor, input);
         }
+      } catch (RuntimeException e) {
+        throw e;
       } catch (Exception e) {
-        return Mono.error(e);
+        throw new RuntimeException(e);
       }
     });
   }

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

@@ -8,13 +8,13 @@ package io.dapr.actors.runtime;
 import io.dapr.actors.ActorId;
 import io.dapr.actors.ActorType;
 import io.dapr.serializer.DefaultObjectSerializer;
-import org.junit.Assert;
-import org.junit.Test;
-import reactor.core.publisher.Mono;
-
 import java.io.IOException;
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
+import reactor.core.publisher.Mono;
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
@@ -35,6 +35,12 @@ public class ActorManagerTest {
     int getCount();
 
     void incrementCount(int delta);
+
+    void throwsException();
+
+    Mono<Void> throwsExceptionHotMono();
+
+    Mono<Void> throwsExceptionMono();
   }
 
   public static class NotRemindableActor extends AbstractActor {
@@ -63,6 +69,21 @@ public class ActorManagerTest {
       this.timeCount = timeCount + delta;
     }
 
+    @Override
+    public void throwsException() {
+      throw new IllegalArgumentException();
+    }
+
+    @Override
+    public Mono<Void> throwsExceptionHotMono() {
+      throw new IllegalArgumentException();
+    }
+
+    @Override
+    public Mono<Void> throwsExceptionMono() {
+      return Mono.error(new IllegalArgumentException());
+    }
+
     public MyActorImpl(ActorRuntimeContext runtimeContext, ActorId id) {
       super(runtimeContext, id);
       super.registerActorTimer(
@@ -107,6 +128,63 @@ public class ActorManagerTest {
       this.context.getObjectSerializer().deserialize(response, String.class));
   }
 
+  @Test
+  public void activateThenInvokeWithActorImplException() throws Exception {
+    ActorId actorId = newActorId();
+    this.manager.activateActor(actorId).block();
+
+    Assertions.assertThrows(RuntimeException.class, () -> {
+      this.manager.invokeMethod(actorId, "throwsException", null).block();
+    });
+  }
+
+  @Test
+  public void activateThenInvokeWithActorImplExceptionButNotSubscribed() throws Exception {
+    ActorId actorId = newActorId();
+    this.manager.activateActor(actorId).block();
+
+    // Nothing happens because we don't call block().
+    this.manager.invokeMethod(actorId, "throwsException", null);
+  }
+
+  @Test
+  public void activateThenInvokeWithActorImplHotMonoException() throws Exception {
+    ActorId actorId = newActorId();
+    this.manager.activateActor(actorId).block();
+
+    Assertions.assertThrows(RuntimeException.class, () -> {
+      this.manager.invokeMethod(actorId, "throwsExceptionHotMono", null).block();
+    });
+  }
+
+  @Test
+  public void activateThenInvokeWithActorImplHotMonoExceptionNotSubscribed() throws Exception {
+    ActorId actorId = newActorId();
+    this.manager.activateActor(actorId).block();
+
+    // Nothing happens because we don't call block().
+    this.manager.invokeMethod(actorId, "throwsExceptionHotMono", null);
+  }
+
+  @Test
+  public void activateThenInvokeWithActorImplMonoException() throws Exception {
+    ActorId actorId = newActorId();
+    this.manager.activateActor(actorId).block();
+
+    Assertions.assertThrows(RuntimeException.class, () -> {
+      this.manager.invokeMethod(actorId, "throwsExceptionMono", null).block();
+    });
+  }
+
+  @Test
+  public void activateThenInvokeWithActorImplMonoExceptionNotSubscribed() throws Exception {
+    ActorId actorId = newActorId();
+    this.manager.activateActor(actorId).block();
+
+    // Nothing happens because we don't call block().
+    this.manager.invokeMethod(actorId, "throwsExceptionMono", null);
+  }
+
   @Test(expected = IllegalArgumentException.class)
   public void activateInvokeDeactivateThenInvoke() throws Exception {
     ActorId actorId = newActorId();