Fix for NPE when method has base type return type like int. (#244)

* Fixed workflow start through WorkflowInvocationHandler when return type is a base type

* Fixed return of the dynamic proxies for activities, child workflows, external workflows and continue as new

Author: mfateev

src/main/java/com/uber/cadence/internal/common/InternalUtils.java

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.internal.common;
 
+import com.google.common.base.Defaults;
 import com.uber.cadence.TaskList;
 import com.uber.cadence.TaskListKind;
 import com.uber.cadence.internal.worker.Shutdownable;
@@ -118,6 +119,13 @@ public static long awaitTermination(long timeoutMillis, Runnable toTerminate) {
     return remainingTimeout;
   }
 
+  public static Object getValueOrDefault(Object value, Class<?> valueClass) {
+    if (value != null) {
+      return value;
+    }
+    return Defaults.defaultValue(valueClass);
+  }
+
   /** Prohibit instantiation */
   private InternalUtils() {}
 }

src/main/java/com/uber/cadence/internal/sync/ActivityInvocationHandler.java

@@ -17,6 +17,8 @@
 
 package com.uber.cadence.internal.sync;
 
+import static com.uber.cadence.internal.common.InternalUtils.getValueOrDefault;
+
 import com.uber.cadence.activity.ActivityMethod;
 import com.uber.cadence.activity.ActivityOptions;
 import com.uber.cadence.common.MethodRetry;
@@ -91,6 +93,6 @@ public Object invoke(Object proxy, Method method, Object[] args) {
         throw Workflow.wrap(e);
       }
     }
-    return function.apply(args);
+    return getValueOrDefault(function.apply(args), method.getReturnType());
   }
 }

src/main/java/com/uber/cadence/internal/sync/ChildWorkflowInvocationHandler.java

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.internal.sync;
 
+import static com.uber.cadence.internal.common.InternalUtils.getValueOrDefault;
 import static com.uber.cadence.internal.common.InternalUtils.getWorkflowMethod;
 import static com.uber.cadence.internal.common.InternalUtils.getWorkflowType;
 
@@ -70,7 +71,9 @@ public Object invoke(Object proxy, Method method, Object[] args) {
               + "from @WorkflowMethod, @QueryMethod or @SignalMethod");
     }
     if (workflowMethod != null) {
-      return stub.execute(method.getReturnType(), method.getGenericReturnType(), args);
+      return getValueOrDefault(
+          stub.execute(method.getReturnType(), method.getGenericReturnType(), args),
+          method.getReturnType());
     }
     if (queryMethod != null) {
       throw new UnsupportedOperationException(

src/main/java/com/uber/cadence/internal/sync/ContinueAsNewWorkflowInvocationHandler.java

@@ -17,6 +17,8 @@
 
 package com.uber.cadence.internal.sync;
 
+import static com.uber.cadence.internal.common.InternalUtils.getValueOrDefault;
+
 import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.workflow.ContinueAsNewOptions;
 import com.uber.cadence.workflow.QueryMethod;
@@ -60,6 +62,6 @@ public Object invoke(Object proxy, Method method, Object[] args) {
     String workflowType = InternalUtils.getWorkflowType(method, workflowMethod);
     WorkflowInternal.continueAsNew(
         Optional.of(workflowType), Optional.of(options), args, decisionContext);
-    return null;
+    return getValueOrDefault(null, method.getReturnType());
   }
 }

src/main/java/com/uber/cadence/internal/sync/ExternalWorkflowInvocationHandler.java

@@ -17,6 +17,8 @@
 
 package com.uber.cadence.internal.sync;
 
+import static com.uber.cadence.internal.common.InternalUtils.getValueOrDefault;
+
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.workflow.ExternalWorkflowStub;
@@ -62,7 +64,7 @@ public Object invoke(Object proxy, Method method, Object[] args) {
               + "created through Workflow.newExternalWorkflowStub");
     }
     if (queryMethod != null) {
-      return queryWorkflow(method, queryMethod, args);
+      return getValueOrDefault(queryWorkflow(method, queryMethod, args), method.getReturnType());
     }
     if (signalMethod != null) {
       signalWorkflow(method, signalMethod, args);

src/main/java/com/uber/cadence/internal/sync/WorkflowInvocationHandler.java

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.internal.sync;
 
+import static com.uber.cadence.internal.common.InternalUtils.getValueOrDefault;
 import static com.uber.cadence.internal.common.InternalUtils.getWorkflowMethod;
 import static com.uber.cadence.internal.common.InternalUtils.getWorkflowType;
 
@@ -179,18 +180,20 @@ public Object invoke(Object proxy, Method method, Object[] args) {
               + " must contain at most one annotation "
               + "from @WorkflowMethod, @QueryMethod or @SignalMethod");
     }
+    Object result;
     if (workflowMethod != null) {
-      return startWorkflow(method, args);
-    }
-    if (queryMethod != null) {
-      return queryWorkflow(method, queryMethod, args);
-    }
-    if (signalMethod != null) {
+      result = startWorkflow(method, args);
+    } else if (queryMethod != null) {
+      result = queryWorkflow(method, queryMethod, args);
+    } else if (signalMethod != null) {
       signalWorkflow(method, signalMethod, args);
-      return null;
+      result = null;
+    } else {
+      throw new IllegalArgumentException(
+          method + " is not annotated with @WorkflowMethod or @QueryMethod");
     }
-    throw new IllegalArgumentException(
-        method + " is not annotated with @WorkflowMethod or @QueryMethod");
+    // Returning null for a built in type leads to NullPointerException
+    return getValueOrDefault(result, method.getReturnType());
   }
 
   private void signalWorkflow(Method method, SignalMethod signalMethod, Object[] args) {

src/test/java/com/uber/cadence/workflow/WorkflowTest.java

@@ -866,8 +866,7 @@ public String execute(String taskList) {
         testActivities.activityWithDelay(100000, true);
         fail("unreachable");
       } catch (CancellationException e) {
-        Workflow.newDetachedCancellationScope(
-            () -> assertEquals("a1", testActivities.activity1("a1")));
+        Workflow.newDetachedCancellationScope(() -> assertEquals(1, testActivities.activity1(1)));
       }
       try {
         Workflow.sleep(Duration.ofHours(1));
@@ -954,14 +953,14 @@ public String execute(String taskList) {
       TestActivities testActivities =
           Workflow.newActivityStub(TestActivities.class, newActivityOptions2());
       Promise<String> a = Async.function(testActivities::activity);
-      Promise<String> a1 = Async.function(testActivities::activity1, "1");
+      Promise<Integer> a1 = Async.function(testActivities::activity1, 1);
       Promise<String> a2 = Async.function(testActivities::activity2, "1", 2);
       Promise<String> a3 = Async.function(testActivities::activity3, "1", 2, 3);
       Promise<String> a4 = Async.function(testActivities::activity4, "1", 2, 3, 4);
       Promise<String> a5 = Async.function(testActivities::activity5, "1", 2, 3, 4, 5);
       Promise<String> a6 = Async.function(testActivities::activity6, "1", 2, 3, 4, 5, 6);
       assertEquals("activity", a.get());
-      assertEquals("1", a1.get());
+      assertEquals(1, (int) a1.get());
       assertEquals("12", a2.get());
       assertEquals("123", a3.get());
       assertEquals("1234", a4.get());
@@ -1133,6 +1132,12 @@ private void assertResult(String expected, WorkflowExecution execution) {
     assertEquals(expected, result);
   }
 
+  private void assertResult(int expected, WorkflowExecution execution) {
+    int result =
+        workflowClient.newUntypedWorkflowStub(execution, Optional.empty()).getResult(int.class);
+    assertEquals(expected, result);
+  }
+
   private void waitForProc(WorkflowExecution execution) {
     workflowClient.newUntypedWorkflowStub(execution, Optional.empty()).getResult(Void.class);
   }
@@ -1150,8 +1155,8 @@ public void testStart() {
 
     if (!useExternalService) {
       // Use worker that polls on a task list configured through @WorkflowMethod annotation of func1
-      assertResult("1", WorkflowClient.start(stubF1::func1, "1"));
-      assertEquals("1", stubF1.func1("1")); // Check that duplicated start just returns the result.
+      assertResult(1, WorkflowClient.start(stubF1::func1, 1));
+      assertEquals(1, stubF1.func1(1)); // Check that duplicated start just returns the result.
     }
     // Check that duplicated start is not allowed for AllowDuplicate IdReusePolicy
     TestMultiargsWorkflowsFunc2 stubF2 =
@@ -1227,8 +1232,8 @@ public void testExecute() throws ExecutionException, InterruptedException {
     assertEquals("func", WorkflowClient.execute(stubF::func).get());
     TestMultiargsWorkflowsFunc1 stubF1 =
         workflowClient.newWorkflowStub(TestMultiargsWorkflowsFunc1.class, workflowOptions);
-    assertEquals("1", WorkflowClient.execute(stubF1::func1, "1").get());
-    assertEquals("1", stubF1.func1("1")); // Check that duplicated start just returns the result.
+    assertEquals(1, (int) WorkflowClient.execute(stubF1::func1, 1).get());
+    assertEquals(1, stubF1.func1(1)); // Check that duplicated start just returns the result.
     TestMultiargsWorkflowsFunc2 stubF2 =
         workflowClient.newWorkflowStub(TestMultiargsWorkflowsFunc2.class, workflowOptions);
     assertEquals("12", WorkflowClient.execute(stubF2::func2, "1", 2).get());
@@ -1294,7 +1299,7 @@ public String execute(String taskList) {
       assertEquals("func", Async.function(stubF::func).get());
       TestMultiargsWorkflowsFunc1 stubF1 =
           Workflow.newChildWorkflowStub(TestMultiargsWorkflowsFunc1.class, workflowOptions);
-      assertEquals("1", Async.function(stubF1::func1, "1").get());
+      assertEquals(1, (int) Async.function(stubF1::func1, 1).get());
       TestMultiargsWorkflowsFunc2 stubF2 =
           Workflow.newChildWorkflowStub(TestMultiargsWorkflowsFunc2.class, workflowOptions);
       assertEquals("12", Async.function(stubF2::func2, "1", 2).get());
@@ -2889,7 +2894,7 @@ public interface TestActivities {
     String activity();
 
     @ActivityMethod(name = "customActivity1")
-    String activity1(String input);
+    int activity1(int input);
 
     String activity2(String a1, int a2);
 
@@ -2994,7 +2999,7 @@ public String activity() {
     }
 
     @Override
-    public String activity1(String a1) {
+    public int activity1(int a1) {
       invocations.add("activity1");
       return a1;
     }
@@ -3150,7 +3155,7 @@ public interface TestMultiargsWorkflowsFunc1 {
       workflowIdReusePolicy = WorkflowIdReusePolicy.RejectDuplicate,
       executionStartToCloseTimeoutSeconds = 10
     )
-    String func1(String input);
+    int func1(int input);
   }
 
   public interface TestMultiargsWorkflowsFunc2 {
@@ -3249,7 +3254,7 @@ public String func() {
     }
 
     @Override
-    public String func1(String a1) {
+    public int func1(int a1) {
       return a1;
     }
 
@@ -3375,7 +3380,7 @@ public String execute(String taskList) {
       Workflow.sleep(Duration.ofSeconds(1));
       String result;
       if (workflowTime == time) {
-        result = testActivities.activity1("activity1");
+        result = "activity" + testActivities.activity1(1);
       } else {
         result = testActivities.activity2("activity2", 2);
       }
@@ -3452,27 +3457,27 @@ public String execute(String taskList) {
       int version = Workflow.getVersion("test_change", Workflow.DEFAULT_VERSION, 1);
       String result = "";
       if (version == Workflow.DEFAULT_VERSION) {
-        result += testActivities.activity1("activity1");
+        result += "activity" + testActivities.activity1(1);
       } else {
         result += testActivities.activity2("activity2", 2); // This is executed.
       }
 
       // Test version change in non-replay code.
       version = Workflow.getVersion("test_change", 1, 2);
       if (version == 1) {
-        result += testActivities.activity1("activity1"); // This is executed.
+        result += "activity" + testActivities.activity1(1); // This is executed.
       } else {
         result += testActivities.activity2("activity2", 2);
       }
 
       // Test adding a version check in replay code.
       if (!getVersionExecuted.contains(taskList + "-test_change_2")) {
-        result += testActivities.activity1("activity1"); // This is executed in non-replay mode.
+        result += "activity" + testActivities.activity1(1); // This is executed in non-replay mode.
         getVersionExecuted.add(taskList + "-test_change_2");
       } else {
         int version2 = Workflow.getVersion("test_change_2", Workflow.DEFAULT_VERSION, 1);
         if (version2 == Workflow.DEFAULT_VERSION) {
-          result += testActivities.activity1("activity1"); // This is executed in replay mode.
+          result += "activity" + testActivities.activity1(1); // This is executed in replay mode.
         } else {
           result += testActivities.activity2("activity2", 2);
         }
@@ -3482,7 +3487,7 @@ public String execute(String taskList) {
       Workflow.sleep(1000);
       version = Workflow.getVersion("test_change", 1, 2);
       if (version == 1) {
-        result += testActivities.activity1("activity1"); // This is executed.
+        result += "activity" + testActivities.activity1(1); // This is executed.
       } else {
         result += testActivities.activity2("activity2", 2);
       }
@@ -3523,7 +3528,7 @@ public String execute(String taskList) {
       if (!getVersionExecuted.contains(taskList)) {
         int version = Workflow.getVersion("test_change", Workflow.DEFAULT_VERSION, 1);
         if (version == Workflow.DEFAULT_VERSION) {
-          result = testActivities.activity1("activity1");
+          result = "activity" + testActivities.activity1(1);
         } else {
           result = testActivities.activity2("activity2", 2); // This is executed in non-replay mode.
         }
@@ -3594,7 +3599,7 @@ public String execute(String taskList) {
       int version = Workflow.getVersion("test_change", Workflow.DEFAULT_VERSION, 1);
       String result = "";
       if (version == Workflow.DEFAULT_VERSION) {
-        result += testActivities.activity1("activity1");
+        result += "activity" + testActivities.activity1(1);
       } else {
         result += testActivities.activity2("activity2", 2); // This is executed.
       }
@@ -3638,7 +3643,7 @@ public void execute(String taskList) {
       TestActivities activities =
           Workflow.newActivityStub(TestActivities.class, newActivityOptions1(taskList));
       if (!Workflow.isReplaying()) {
-        activities.activity1("foo");
+        activities.activity1(1);
       }
     }
   }