Non deserializable activity arguments should cause activity failure (#381)

Author: mfateev

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

@@ -202,9 +202,9 @@ public ActivityTaskHandler.Result execute(ActivityTaskImpl task, Scope metricsSc
       ActivityExecutionContext context =
           new ActivityExecutionContextImpl(service, domain, task, dataConverter, heartbeatExecutor);
       byte[] input = task.getInput();
-      Object[] args = dataConverter.fromDataArray(input, method.getGenericParameterTypes());
       CurrentActivityExecutionContext.set(context);
       try {
+        Object[] args = dataConverter.fromDataArray(input, method.getGenericParameterTypes());
         Object result = method.invoke(activity, args);
         RespondActivityTaskCompletedRequest request = new RespondActivityTaskCompletedRequest();
         if (context.isDoNotCompleteOnReturn()) {
@@ -239,8 +239,8 @@ public ActivityTaskHandler.Result execute(ActivityTaskImpl task, Scope metricsSc
           new LocalActivityExecutionContextImpl(service, domain, task);
       CurrentActivityExecutionContext.set(context);
       byte[] input = task.getInput();
-      Object[] args = dataConverter.fromDataArray(input, method.getGenericParameterTypes());
       try {
+        Object[] args = dataConverter.fromDataArray(input, method.getGenericParameterTypes());
         Object result = method.invoke(activity, args);
         RespondActivityTaskCompletedRequest request = new RespondActivityTaskCompletedRequest();
         if (method.getReturnType() != Void.TYPE) {

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

@@ -169,6 +169,10 @@ public static ActivityStub newUntypedActivityStub(ActivityOptions options) {
     return ActivityStubImpl.newInstance(options, getWorkflowInterceptor());
   }
 
+  public static ActivityStub newUntypedLocalActivityStub(LocalActivityOptions options) {
+    return LocalActivityStubImpl.newInstance(options, getWorkflowInterceptor());
+  }
+
   @SuppressWarnings("unchecked")
   public static <T> T newChildWorkflowStub(
       Class<T> workflowInterface, ChildWorkflowOptions options) {

src/main/java/com/uber/cadence/workflow/Workflow.java

@@ -389,7 +389,7 @@ public static <T> T newActivityStub(Class<T> activityInterface) {
   }
 
   /**
-   * Creates client stub to activities that implement given interface.
+   * Creates non typed client stub to activities. Allows executing activities by their string name.
    *
    * @param options specify the activity invocation parameters.
    */
@@ -422,6 +422,16 @@ public static <T> T newLocalActivityStub(Class<T> activityInterface) {
     return WorkflowInternal.newLocalActivityStub(activityInterface, null);
   }
 
+  /**
+   * Creates non typed client stub to local activities. Allows executing activities by their string
+   * name.
+   *
+   * @param options specify the local activity invocation parameters.
+   */
+  public static ActivityStub newUntypedLocalActivityStub(LocalActivityOptions options) {
+    return WorkflowInternal.newUntypedLocalActivityStub(options);
+  }
+
   /**
    * Creates client stub that can be used to start a child workflow that implements the given
    * interface using parent options. Use {@link #newExternalWorkflowStub(Class, String)} to get a

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

@@ -4447,6 +4447,61 @@ public void testNonSerializableExceptionInActivity() {
     assertTrue(result.contains("NonSerializableException"));
   }
 
+  public interface NonDeserializableArgumentsActivity {
+
+    @ActivityMethod(scheduleToCloseTimeoutSeconds = 5)
+    void execute(int arg);
+  }
+
+  public static class NonDeserializableExceptionActivityImpl
+      implements NonDeserializableArgumentsActivity {
+
+    @Override
+    public void execute(int arg) {}
+  }
+
+  public static class TestNonSerializableArgumentsInActivityWorkflow implements TestWorkflow1 {
+
+    @Override
+    public String execute(String taskList) {
+      StringBuilder result = new StringBuilder();
+      ActivityStub activity =
+          Workflow.newUntypedActivityStub(
+              new ActivityOptions.Builder()
+                  .setScheduleToCloseTimeout(Duration.ofSeconds(5))
+                  .build());
+      ActivityStub localActivity =
+          Workflow.newUntypedLocalActivityStub(
+              new LocalActivityOptions.Builder()
+                  .setScheduleToCloseTimeout(Duration.ofSeconds(5))
+                  .build());
+      try {
+        activity.execute("NonDeserializableArgumentsActivity::execute", Void.class, "boo");
+      } catch (ActivityFailureException e) {
+        result.append(e.getCause().getClass().getSimpleName());
+      }
+      result.append("-");
+      try {
+        localActivity.execute("NonDeserializableArgumentsActivity::execute", Void.class, "boo");
+      } catch (ActivityFailureException e) {
+        result.append(e.getCause().getClass().getSimpleName());
+      }
+      return result.toString();
+    }
+  }
+
+  @Test
+  public void testNonSerializableArgumentsInActivity() {
+    worker.registerActivitiesImplementations(new NonDeserializableExceptionActivityImpl());
+    startWorkerFor(TestNonSerializableArgumentsInActivityWorkflow.class);
+    TestWorkflow1 workflowStub =
+        workflowClient.newWorkflowStub(
+            TestWorkflow1.class, newWorkflowOptionsBuilder(taskList).build());
+
+    String result = workflowStub.execute(taskList);
+    assertEquals("DataConverterException-DataConverterException", result);
+  }
+
   public interface NonSerializableExceptionChildWorkflow {
 
     @WorkflowMethod