From d9675e82b8c370e2da3e7b5bc57c6c2d6a9f344c Mon Sep 17 00:00:00 2001
From: Liang Mei <meiliang@uber.com>
Date: Mon, 8 Jul 2019 10:57:55 -0700
Subject: [PATCH] Wrap data converter exception in activity failure exception

---
 .../internal/sync/SyncDecisionContext.java    | 78 ++++++++++++-------
 .../workflow/ActivityFailureException.java    |  5 ++
 2 files changed, 55 insertions(+), 28 deletions(-)

diff --git a/src/main/java/com/uber/cadence/internal/sync/SyncDecisionContext.java b/src/main/java/com/uber/cadence/internal/sync/SyncDecisionContext.java
index 8d4ebec07..90b2290fc 100644
--- a/src/main/java/com/uber/cadence/internal/sync/SyncDecisionContext.java
+++ b/src/main/java/com/uber/cadence/internal/sync/SyncDecisionContext.java
@@ -26,6 +26,7 @@
 import com.uber.cadence.activity.LocalActivityOptions;
 import com.uber.cadence.common.RetryOptions;
 import com.uber.cadence.converter.DataConverter;
+import com.uber.cadence.converter.DataConverterException;
 import com.uber.cadence.internal.common.RetryParameters;
 import com.uber.cadence.internal.replay.ActivityTaskFailedException;
 import com.uber.cadence.internal.replay.ActivityTaskTimeoutException;
@@ -83,7 +84,12 @@ final class SyncDecisionContext implements WorkflowInterceptor {
   private final Map<String, Functions.Func1<byte[], byte[]>> queryCallbacks = new HashMap<>();
   private final byte[] lastCompletionResult;
 
-  public SyncDecisionContext(
+  @FunctionalInterface
+  public interface ActivityExecutor {
+    Promise<byte[]> getResult(byte[] input);
+  }
+
+  SyncDecisionContext(
       DecisionContext context,
       DataConverter converter,
       Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
@@ -127,19 +133,51 @@ public <T> Promise<T> executeActivity(
     if (retryOptions != null && !context.isServerSideActivityRetry()) {
       return WorkflowRetryerInternal.retryAsync(
           retryOptions,
-          () -> executeActivityOnce(activityName, options, args, resultClass, resultType));
-    }
-    return executeActivityOnce(activityName, options, args, resultClass, resultType);
+          () ->
+              executeActivityOnce(
+                  activityName,
+                  args,
+                  resultClass,
+                  resultType,
+                  input -> executeActivityOnce(activityName, options, input)));
+    }
+    return executeActivityOnce(
+        activityName,
+        args,
+        resultClass,
+        resultType,
+        input -> executeActivityOnce(activityName, options, input));
   }
 
   private <T> Promise<T> executeActivityOnce(
-      String name, ActivityOptions options, Object[] args, Class<T> returnClass, Type returnType) {
-    byte[] input = converter.toData(args);
-    Promise<byte[]> binaryResult = executeActivityOnce(name, options, input);
+      String name,
+      Object[] args,
+      Class<T> returnClass,
+      Type returnType,
+      ActivityExecutor executor) {
+    byte[] input;
+    try {
+      input = converter.toData(args);
+    } catch (DataConverterException e) {
+      CompletablePromise<T> result = Workflow.newPromise();
+      result.completeExceptionally(
+          new ActivityFailureException(new ActivityType().setName(name), e));
+      return result;
+    }
+
+    Promise<byte[]> binaryResult = executor.getResult(input);
+
     if (returnClass == Void.TYPE) {
       return binaryResult.thenApply((r) -> null);
     }
-    return binaryResult.thenApply((r) -> converter.fromData(r, returnClass, returnType));
+    return binaryResult.thenApply(
+        (r) -> {
+          try {
+            return converter.fromData(r, returnClass, returnType);
+          } catch (DataConverterException e) {
+            throw new ActivityFailureException(new ActivityType().setName(name), e);
+          }
+        });
   }
 
   private Promise<byte[]> executeActivityOnce(String name, ActivityOptions options, byte[] input) {
@@ -237,34 +275,18 @@ public <R> Promise<R> executeLocalActivity(
     long startTime = WorkflowInternal.currentTimeMillis();
     return WorkflowRetryerInternal.retryAsync(
         (attempt, currentStart) ->
-            executeLocalActivityOnce(
+            executeActivityOnce(
                 activityName,
-                options,
                 args,
                 resultClass,
                 resultType,
-                currentStart - startTime,
-                attempt),
+                input ->
+                    executeLocalActivityOnce(
+                        activityName, options, input, currentStart - startTime, attempt)),
         1,
         startTime);
   }
 
-  private <T> Promise<T> executeLocalActivityOnce(
-      String name,
-      LocalActivityOptions options,
-      Object[] args,
-      Class<T> returnClass,
-      Type returnType,
-      long elapsed,
-      int attempt) {
-    byte[] input = converter.toData(args);
-    Promise<byte[]> binaryResult = executeLocalActivityOnce(name, options, input, elapsed, attempt);
-    if (returnClass == Void.TYPE) {
-      return binaryResult.thenApply((r) -> null);
-    }
-    return binaryResult.thenApply((r) -> converter.fromData(r, returnClass, returnType));
-  }
-
   private Promise<byte[]> executeLocalActivityOnce(
       String name, LocalActivityOptions options, byte[] input, long elapsed, int attempt) {
     ActivityCallback callback = new ActivityCallback();
diff --git a/src/main/java/com/uber/cadence/workflow/ActivityFailureException.java b/src/main/java/com/uber/cadence/workflow/ActivityFailureException.java
index eef366eb1..3beef8e5a 100644
--- a/src/main/java/com/uber/cadence/workflow/ActivityFailureException.java
+++ b/src/main/java/com/uber/cadence/workflow/ActivityFailureException.java
@@ -29,6 +29,11 @@ public final class ActivityFailureException extends ActivityException {
   private int attempt;
   private Duration backoff;
 
+  public ActivityFailureException(ActivityType activityType, Throwable cause) {
+    super("ActivityFailureException", -1, activityType, "");
+    initCause(cause);
+  }
+
   public ActivityFailureException(
       long eventId, ActivityType activityType, String activityId, Throwable cause) {
     super("ActivityFailureException", eventId, activityType, activityId);