Fix encoding of shadowing activities (#600)

* Fix encoding of shadowing activities

Author: yux0

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

@@ -15,10 +15,7 @@
  */
 package com.uber.cadence.internal.shadowing;
 
-import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.activity.ActivityMethod;
-import com.uber.cadence.shadower.ReplayWorkflowActivityParams;
-import com.uber.cadence.shadower.ReplayWorkflowActivityResult;
 import com.uber.cadence.shadower.shadowerConstants;
 import com.uber.cadence.worker.WorkflowImplementationOptions;
 import com.uber.cadence.workflow.Functions;

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

@@ -22,16 +22,13 @@
 import com.uber.cadence.History;
 import com.uber.cadence.HistoryEvent;
 import com.uber.cadence.HistoryEventFilterType;
-import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.activity.Activity;
 import com.uber.cadence.common.WorkflowExecutionHistory;
 import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.internal.common.RpcRetryer;
 import com.uber.cadence.internal.common.WorkflowExecutionUtils;
 import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.serviceclient.IWorkflowService;
-import com.uber.cadence.shadower.ReplayWorkflowActivityParams;
-import com.uber.cadence.shadower.ReplayWorkflowActivityResult;
 import com.uber.cadence.testing.TestWorkflowEnvironment;
 import com.uber.cadence.worker.Worker;
 import com.uber.cadence.worker.WorkflowImplementationOptions;
@@ -99,8 +96,9 @@ public ReplayWorkflowActivityResult replay(ReplayWorkflowActivityParams request)
 
     // Retrieve process from heartbeat
     Optional<HeartbeatDetail> heartbeatDetail = Activity.getHeartbeatDetails(HeartbeatDetail.class);
+    ReplayWorkflowActivityResult heartbeatResult;
     if (heartbeatDetail.isPresent()) {
-      ReplayWorkflowActivityResult heartbeatResult = heartbeatDetail.get().getReplayResult();
+      heartbeatResult = heartbeatDetail.get().getReplayResult();
       successCount = heartbeatResult.getSucceeded();
       failedCount = heartbeatResult.getFailed();
       skippedCount = heartbeatResult.getSkipped();
@@ -114,21 +112,22 @@ public ReplayWorkflowActivityResult replay(ReplayWorkflowActivityParams request)
       successCount += oneReplayResult.getSucceeded();
       failedCount += oneReplayResult.getFailed();
       skippedCount += oneReplayResult.getSkipped();
-      ReplayWorkflowActivityResult heartbeatResult =
-          new ReplayWorkflowActivityResult()
-              .setSucceeded(successCount)
-              .setFailed(failedCount)
-              .setSkipped(skippedCount);
+      heartbeatResult = new ReplayWorkflowActivityResult();
+      heartbeatResult.setSucceeded(successCount);
+      heartbeatResult.setFailed(failedCount);
+      heartbeatResult.setSkipped(skippedCount);
       Activity.heartbeat(new HeartbeatDetail(heartbeatResult, replayIndex));
     }
-    return new ReplayWorkflowActivityResult()
-        .setSucceeded(successCount)
-        .setFailed(failedCount)
-        .setSkipped(skippedCount);
+    ReplayWorkflowActivityResult result = new ReplayWorkflowActivityResult();
+    result.setSucceeded(successCount);
+    result.setFailed(failedCount);
+    result.setSkipped(skippedCount);
+    return result;
   }
 
   public ReplayWorkflowActivityResult replayOneExecution(
       String domain, WorkflowExecution execution) {
+    ReplayWorkflowActivityResult result = new ReplayWorkflowActivityResult();
     WorkflowExecutionHistory workflowHistory;
     try {
       workflowHistory = getFullHistory(domain, execution);
@@ -139,23 +138,27 @@ public ReplayWorkflowActivityResult replayOneExecution(
               + ". Execution: "
               + execution.toString(),
           e);
-      return new ReplayWorkflowActivityResult().setSkipped(1);
+      result.setSkipped(1);
+      return result;
     }
 
     try {
       boolean isSuccess = replayWorkflowHistory(domain, execution, workflowHistory);
       if (isSuccess) {
         this.metricsScope.counter(MetricsType.REPLAY_SUCCESS_COUNTER).inc(1);
-        return new ReplayWorkflowActivityResult().setSucceeded(1);
+        result.setSucceeded(1);
+        return result;
       } else {
         this.metricsScope.counter(MetricsType.REPLAY_SKIPPED_COUNTER).inc(1);
-        return new ReplayWorkflowActivityResult().setSkipped(1);
+        result.setSkipped(1);
+        return result;
       }
     } catch (NonRetryableException e) {
       throw e;
     } catch (Exception e) {
       this.metricsScope.counter(MetricsType.REPLAY_FAILED_COUNTER).inc(1);
-      return new ReplayWorkflowActivityResult().setFailed(1);
+      result.setFailed(1);
+      return result;
     }
   }
 
@@ -170,7 +173,7 @@ protected WorkflowExecutionHistory getFullHistory(String domain, WorkflowExecuti
               RpcRetryer.DEFAULT_RPC_RETRY_OPTIONS,
               () ->
                   WorkflowExecutionUtils.getHistoryPage(
-                      nextPageToken, this.serviceClient, domain, execution));
+                      nextPageToken, this.serviceClient, domain, execution.toThrift()));
       pageToken = resp.getNextPageToken();
 
       // handle raw history

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

@@ -0,0 +1,57 @@
+/*
+ *  Modifications Copyright (c) 2017-2021 Uber Technologies Inc.
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+package com.uber.cadence.internal.shadowing;
+
+import java.util.List;
+
+/**
+ * This class is the JSON serializable class of {@link
+ * com.uber.cadence.shadower.ReplayWorkflowActivityParams} Make sure this class is sync with auto
+ * generated ReplayWorkflowActivityParams
+ */
+public class ReplayWorkflowActivityParams {
+  private String domain;
+  private List<WorkflowExecution> executions;
+
+  public ReplayWorkflowActivityParams() {}
+
+  public String getDomain() {
+    return domain;
+  }
+
+  public void setDomain(String domain) {
+    this.domain = domain;
+  }
+
+  public List<WorkflowExecution> getExecutions() {
+    return executions;
+  }
+
+  public void setExecutions(List<WorkflowExecution> executions) {
+    this.executions = executions;
+  }
+
+  @Override
+  public String toString() {
+    return "ReplayWorkflowActivityParams{"
+        + "domain='"
+        + domain
+        + '\''
+        + ", executions="
+        + executions
+        + '}';
+  }
+}

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

@@ -0,0 +1,65 @@
+/*
+ *  Modifications Copyright (c) 2017-2021 Uber Technologies Inc.
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+package com.uber.cadence.internal.shadowing;
+
+/**
+ * This class is the JSON serializable class of {@link
+ * com.uber.cadence.shadower.ReplayWorkflowActivityResult} Make sure this class is sync with auto
+ * generated ReplayWorkflowActivityResult
+ */
+public class ReplayWorkflowActivityResult {
+  private int succeeded;
+  private int skipped;
+  private int failed;
+
+  public ReplayWorkflowActivityResult() {}
+
+  public int getSucceeded() {
+    return succeeded;
+  }
+
+  public void setSucceeded(int succeeded) {
+    this.succeeded = succeeded;
+  }
+
+  public int getSkipped() {
+    return skipped;
+  }
+
+  public void setSkipped(int skipped) {
+    this.skipped = skipped;
+  }
+
+  public int getFailed() {
+    return failed;
+  }
+
+  public void setFailed(int failed) {
+    this.failed = failed;
+  }
+
+  @Override
+  public String toString() {
+    return "ReplayWorkflowActivityResult{"
+        + "succeeded="
+        + succeeded
+        + ", skipped="
+        + skipped
+        + ", failed="
+        + failed
+        + '}';
+  }
+}

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

@@ -18,8 +18,6 @@
 import static com.uber.cadence.shadower.shadowerConstants.ScanWorkflowActivityName;
 
 import com.uber.cadence.activity.ActivityMethod;
-import com.uber.cadence.shadower.ScanWorkflowActivityParams;
-import com.uber.cadence.shadower.ScanWorkflowActivityResult;
 
 public interface ScanWorkflowActivity {
   @ActivityMethod(name = ScanWorkflowActivityName)

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

@@ -24,8 +24,6 @@
 import com.uber.cadence.WorkflowExecutionInfo;
 import com.uber.cadence.internal.common.RpcRetryer;
 import com.uber.cadence.serviceclient.IWorkflowService;
-import com.uber.cadence.shadower.ScanWorkflowActivityParams;
-import com.uber.cadence.shadower.ScanWorkflowActivityResult;
 import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
@@ -55,9 +53,14 @@ public ScanWorkflowActivityResult scan(ScanWorkflowActivityParams params) throws
     List<WorkflowExecution> executions =
         samplingWorkflows(resp.getExecutions(), params.getSamplingRate());
 
-    return new ScanWorkflowActivityResult()
-        .setExecutions(executions)
-        .setNextPageToken(resp.getNextPageToken());
+    ScanWorkflowActivityResult result = new ScanWorkflowActivityResult();
+    result.setExecutions(
+        executions
+            .stream()
+            .map(com.uber.cadence.internal.shadowing.WorkflowExecution::new)
+            .collect(Collectors.toList()));
+    result.setNextPageToken(resp.getNextPageToken());
+    return result;
   }
 
   protected ListWorkflowExecutionsResponse scanWorkflows(ListWorkflowExecutionsRequest request)

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

@@ -0,0 +1,91 @@
+/*
+ *  Modifications Copyright (c) 2017-2021 Uber Technologies Inc.
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+package com.uber.cadence.internal.shadowing;
+
+import java.util.Arrays;
+
+/**
+ * This class is the JSON serializable class of {@link
+ * com.uber.cadence.shadower.ScanWorkflowActivityParams} Make sure this class is sync with auto
+ * generated ScanWorkflowActivityParams
+ */
+public class ScanWorkflowActivityParams {
+  private String domain;
+  private String workflowQuery;
+  private double samplingRate;
+  private int pageSize;
+  private byte[] nextPageToken;
+
+  public ScanWorkflowActivityParams() {}
+
+  public String getDomain() {
+    return domain;
+  }
+
+  public void setDomain(String domain) {
+    this.domain = domain;
+  }
+
+  public String getWorkflowQuery() {
+    return workflowQuery;
+  }
+
+  public void setWorkflowQuery(String workflowQuery) {
+    this.workflowQuery = workflowQuery;
+  }
+
+  public double getSamplingRate() {
+    return samplingRate;
+  }
+
+  public void setSamplingRate(double samplingRate) {
+    this.samplingRate = samplingRate;
+  }
+
+  public int getPageSize() {
+    return pageSize;
+  }
+
+  public void setPageSize(int pageSize) {
+    this.pageSize = pageSize;
+  }
+
+  public byte[] getNextPageToken() {
+    return nextPageToken;
+  }
+
+  public void setNextPageToken(byte[] nextPageToken) {
+    this.nextPageToken = nextPageToken;
+  }
+
+  @Override
+  public String toString() {
+    return "ScanWorkflowActivityParams{"
+        + "domain='"
+        + domain
+        + '\''
+        + ", workflowQuery='"
+        + workflowQuery
+        + '\''
+        + ", samplingRate="
+        + samplingRate
+        + ", pageSize="
+        + pageSize
+        + ", nextPageToken="
+        + Arrays.toString(nextPageToken)
+        + '}';
+  }
+}