Expose test environment option for shadowing (#604)

Author: yux0

src/main/java/com/uber/cadence/internal/shadowing/ReplayWorkflowActivityImpl.java

@@ -29,6 +29,7 @@
 import com.uber.cadence.internal.common.WorkflowExecutionUtils;
 import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.serviceclient.IWorkflowService;
+import com.uber.cadence.testing.TestEnvironmentOptions;
 import com.uber.cadence.testing.TestWorkflowEnvironment;
 import com.uber.cadence.worker.Worker;
 import com.uber.cadence.worker.WorkflowImplementationOptions;
@@ -51,9 +52,17 @@ public final class ReplayWorkflowActivityImpl implements ReplayWorkflowActivity
 
   public ReplayWorkflowActivityImpl(
       IWorkflowService serviceClient, Scope metricsScope, String taskList) {
+    this(serviceClient, metricsScope, taskList, new TestEnvironmentOptions.Builder().build());
+  }
+
+  public ReplayWorkflowActivityImpl(
+      IWorkflowService serviceClient,
+      Scope metricsScope,
+      String taskList,
+      TestEnvironmentOptions testOptions) {
     this.serviceClient = Objects.requireNonNull(serviceClient);
     this.metricsScope = Objects.requireNonNull(metricsScope);
-    worker = TestWorkflowEnvironment.newInstance().newWorker(taskList);
+    worker = TestWorkflowEnvironment.newInstance(testOptions).newWorker(taskList);
   }
 
   @Override
@@ -131,7 +140,7 @@ public ReplayWorkflowActivityResult replayOneExecution(
     WorkflowExecutionHistory workflowHistory;
     try {
       workflowHistory = getFullHistory(domain, execution);
-    } catch (Exception e) {
+    } catch (Throwable e) {
       log.error(
           "skipped workflow execution with domain: "
               + domain

src/main/java/com/uber/cadence/testing/WorkflowShadower.java

@@ -43,11 +43,38 @@ public final class WorkflowShadower {
   private final ScanWorkflowActivity scanWorkflow;
   private final ReplayWorkflowActivity replayWorkflow;
 
+  /**
+   * WorkflowShadower fetches the workflow history from remote and run replay test locally.
+   *
+   * @param service is the target service to fetch workflow history.
+   * @param options is the shadowing options.
+   * @param taskList is the task list used in the workflows.
+   */
   public WorkflowShadower(IWorkflowService service, ShadowingOptions options, String taskList) {
-
     this(service, options, taskList, new NoopScope());
   }
 
+  /**
+   * WorkflowShadower fetches the workflow history from remote and run replay test locally.
+   *
+   * @param service is the target service to fetch workflow history.
+   * @param options is the shadowing options.
+   * @param taskList is the task list used in the workflows.
+   * @param metricsScope uses to emit replay metrics.
+   * @param testOptions uses to set customized data converter, interceptor and context propagator.
+   */
+  public WorkflowShadower(
+      IWorkflowService service,
+      ShadowingOptions options,
+      String taskList,
+      Scope metricsScope,
+      TestEnvironmentOptions testOptions) {
+    this(
+        options,
+        new ScanWorkflowActivityImpl(service),
+        new ReplayWorkflowActivityImpl(service, metricsScope, taskList, testOptions));
+  }
+
   public WorkflowShadower(
       IWorkflowService service, ShadowingOptions options, String taskList, Scope metricsScope) {
     this(

src/main/java/com/uber/cadence/worker/ShadowingWorker.java

@@ -36,6 +36,7 @@
 import com.uber.cadence.serviceclient.IWorkflowService;
 import com.uber.cadence.shadower.WorkflowParams;
 import com.uber.cadence.shadower.shadowerConstants;
+import com.uber.cadence.testing.TestEnvironmentOptions;
 import com.uber.cadence.workflow.Functions;
 import com.uber.m3.tally.Scope;
 import com.uber.m3.util.ImmutableMap;
@@ -55,11 +56,37 @@ public final class ShadowingWorker implements Suspendable {
   private final ShadowingOptions shadowingOptions;
   private final AtomicBoolean started = new AtomicBoolean();
 
+  /**
+   * ShadowingWorker starts a shadowing workflow to replay the target workflows.
+   *
+   * @param client is the target endpoint to fetch workflow history.
+   * @param taskList is the task list used in the workflows.
+   * @param options is worker option.
+   * @param shadowingOptions is the shadowing options.
+   */
   public ShadowingWorker(
       WorkflowClient client,
       String taskList,
       WorkerOptions options,
       ShadowingOptions shadowingOptions) {
+    this(client, taskList, options, shadowingOptions, new TestEnvironmentOptions.Builder().build());
+  }
+
+  /**
+   * ShadowingWorker starts a shadowing workflow to replay the target workflows.
+   *
+   * @param client is the target endpoint to fetch workflow history.
+   * @param taskList is the task list used in the workflows.
+   * @param options is worker option.
+   * @param shadowingOptions is the shadowing options.
+   * @param testOptions uses to set customized data converter, interceptor and context propagator.
+   */
+  public ShadowingWorker(
+      WorkflowClient client,
+      String taskList,
+      WorkerOptions options,
+      ShadowingOptions shadowingOptions,
+      TestEnvironmentOptions testOptions) {
     options = MoreObjects.firstNonNull(options, WorkerOptions.defaultInstance());
     this.shadowingOptions = Objects.requireNonNull(shadowingOptions);
     this.taskList = shadowingOptions.getDomain() + "-" + taskList;
@@ -75,7 +102,8 @@ public ShadowingWorker(
                     MetricsTag.TASK_LIST,
                     this.taskList));
     ScanWorkflowActivity scanActivity = new ScanWorkflowActivityImpl(client.getService());
-    replayActivity = new ReplayWorkflowActivityImpl(client.getService(), metricsScope, taskList);
+    replayActivity =
+        new ReplayWorkflowActivityImpl(client.getService(), metricsScope, taskList, testOptions);
 
     SingleWorkerOptions activityOptions =
         SingleWorkerOptions.newBuilder()

src/main/java/com/uber/cadence/worker/WorkflowStatus.java

@@ -17,6 +17,7 @@
 
 import com.uber.cadence.WorkflowExecutionCloseStatus;
 
+/** The WorkflowStatus is only used in shadowing option. */
 public enum WorkflowStatus {
   OPEN("OPEN"),
   CLOSED("CLOSED"),

src/test/java/com/uber/cadence/internal/shadowing/ReplayWorkflowActivityTest.java

@@ -41,10 +41,12 @@
 import com.uber.cadence.WorkflowExecutionStartedEventAttributes;
 import com.uber.cadence.WorkflowType;
 import com.uber.cadence.common.WorkflowExecutionHistory;
+import com.uber.cadence.converter.JsonDataConverter;
 import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.internal.testing.WorkflowTestingTest;
 import com.uber.cadence.serviceclient.IWorkflowService;
 import com.uber.cadence.testing.TestActivityEnvironment;
+import com.uber.cadence.testing.TestEnvironmentOptions;
 import com.uber.m3.tally.RootScopeBuilder;
 import com.uber.m3.tally.Scope;
 import com.uber.m3.util.Duration;
@@ -70,7 +72,11 @@ public class ReplayWorkflowActivityTest {
   public void init() {
     mockServiceClient = mock(IWorkflowService.class);
     metricsScope = new RootScopeBuilder().reportEvery(Duration.ofMillis(1000));
-    activity = new ReplayWorkflowActivityImpl(mockServiceClient, metricsScope, "test");
+    TestEnvironmentOptions testOptions =
+        new TestEnvironmentOptions.Builder()
+            .setDataConverter(JsonDataConverter.getInstance())
+            .build();
+    activity = new ReplayWorkflowActivityImpl(mockServiceClient, metricsScope, "test", testOptions);
     activity.registerWorkflowImplementationTypes(WorkflowTestingTest.EmptyWorkflowImpl.class);
 
     domain = UUID.randomUUID().toString();