Retry getting result from passive in case of replication delay (#473)

* Pull in latest idls
* Retry getting result in passive cluster in case of replication lag
* Review comments

Author: meiliang86

src/main/idls

@@ -25,17 +25,16 @@
 import com.uber.cadence.SearchAttributes;
 import com.uber.cadence.TaskList;
 import com.uber.cadence.TaskListKind;
-import com.uber.cadence.EntityNotExistsError;
 import com.uber.cadence.converter.DataConverter;
 import com.uber.cadence.converter.JsonDataConverter;
 import com.uber.cadence.internal.worker.Shutdownable;
 import com.uber.cadence.workflow.WorkflowMethod;
 import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
-import java.util.Arrays;
 import java.util.ArrayList;
-import java.util.List;
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -183,17 +182,15 @@ public static History DeserializeFromBlobToHistoryEvents(
     return new History().setEvents(events);
   }
 
-  // This method deserialize the history event data to blob data
-  public static List<DataBlob> DeserializeFromHistoryEventToBlobData(List<HistoryEvent> events)
-      throws EntityNotExistsError {
-
-    List<DataBlob> blobs = new ArrayList<DataBlob>();
+  // This method serializes history event to blob data
+  public static List<DataBlob> SerializeFromHistoryEventToBlobData(List<HistoryEvent> events) {
+    List<DataBlob> blobs = new ArrayList<>(events.size());
     for (HistoryEvent event : events) {
       DataBlob blob = new DataBlob();
       try {
         blob.setData(serializer.serialize(event));
       } catch (org.apache.thrift.TException err) {
-        throw new EntityNotExistsError("Deserialize blob data to history event failed with unknown error");
+        throw new RuntimeException("Serialize to blob data failed", err);
       }
       blobs.add(blob);
     }

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

@@ -100,6 +100,9 @@ public class WorkflowExecutionUtils {
           .setDoNotRetry(BadRequestError.class, EntityNotExistsError.class)
           .build();
 
+  // Wait period for passive cluster to retry getting workflow result in case of replication delay.
+  private static final long ENTITY_NOT_EXIST_RETRY_WAIT_MILLIS = 500;
+
   /**
    * Returns result of a workflow instance execution or throws an exception if workflow did not
    * complete successfully.
@@ -178,7 +181,7 @@ private static byte[] getResultFromCloseEvent(
   }
 
   /** Returns an instance closing event, potentially waiting for workflow to complete. */
-  public static HistoryEvent getInstanceCloseEvent(
+  private static HistoryEvent getInstanceCloseEvent(
       IWorkflowService service,
       String domain,
       WorkflowExecution workflowExecution,
@@ -191,31 +194,53 @@ public static HistoryEvent getInstanceCloseEvent(
     long start = System.currentTimeMillis();
     HistoryEvent event;
     do {
+      if (timeout != 0 && System.currentTimeMillis() - start > unit.toMillis(timeout)) {
+        throw new TimeoutException(
+            "WorkflowId="
+                + workflowExecution.getWorkflowId()
+                + ", runId="
+                + workflowExecution.getRunId()
+                + ", timeout="
+                + timeout
+                + ", unit="
+                + unit);
+      }
+
       GetWorkflowExecutionHistoryRequest r = new GetWorkflowExecutionHistoryRequest();
       r.setDomain(domain);
       r.setExecution(workflowExecution);
       r.setHistoryEventFilterType(HistoryEventFilterType.CLOSE_EVENT);
       r.setNextPageToken(pageToken);
       r.setWaitForNewEvent(true);
+      r.setSkipArchival(true);
       try {
         response =
             Retryer.retryWithResult(retryParameters, () -> service.GetWorkflowExecutionHistory(r));
       } catch (EntityNotExistsError e) {
+        if (e.activeCluster != null
+            && e.currentCluster != null
+            && !e.activeCluster.equals(e.currentCluster)) {
+          // Current cluster is passive cluster. Execution might not exist because of replication
+          // lag. If we are still within timeout, wait for a little bit and retry.
+          if (timeout != 0
+              && System.currentTimeMillis() + ENTITY_NOT_EXIST_RETRY_WAIT_MILLIS - start
+                  > unit.toMillis(timeout)) {
+            throw e;
+          }
+
+          try {
+            Thread.sleep(ENTITY_NOT_EXIST_RETRY_WAIT_MILLIS);
+          } catch (InterruptedException ie) {
+            // Throw entity not exist here.
+            throw e;
+          }
+          continue;
+        }
         throw e;
       } catch (TException e) {
         throw CheckedExceptionWrapper.wrap(e);
       }
-      if (timeout != 0 && System.currentTimeMillis() - start > unit.toMillis(timeout)) {
-        throw new TimeoutException(
-            "WorkflowId="
-                + workflowExecution.getWorkflowId()
-                + ", runId="
-                + workflowExecution.getRunId()
-                + ", timeout="
-                + timeout
-                + ", unit="
-                + unit);
-      }
+
       pageToken = response.getNextPageToken();
       History history = response.getHistory();
       if (history != null && history.getEvents().size() > 0) {

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

@@ -31,7 +31,6 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
-import org.apache.thrift.TException;
 
 interface TestWorkflowStore {
 
@@ -156,7 +155,8 @@ void sendQueryTask(ExecutionId executionId, TaskListId taskList, PollForDecision
       throws EntityNotExistsError;
 
   GetWorkflowExecutionHistoryResponse getWorkflowExecutionHistory(
-      ExecutionId executionId, GetWorkflowExecutionHistoryRequest getRequest) throws EntityNotExistsError;
+      ExecutionId executionId, GetWorkflowExecutionHistoryRequest getRequest)
+      throws EntityNotExistsError;
 
   void getDiagnostics(StringBuilder result);
 

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowStore.java

@@ -18,10 +18,9 @@
 package com.uber.cadence.internal.testservice;
 
 import com.uber.cadence.BadRequestError;
+import com.uber.cadence.DataBlob;
 import com.uber.cadence.EntityNotExistsError;
 import com.uber.cadence.EventType;
-import com.uber.cadence.DataBlob;
-import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.GetWorkflowExecutionHistoryRequest;
 import com.uber.cadence.GetWorkflowExecutionHistoryResponse;
 import com.uber.cadence.History;
@@ -35,6 +34,7 @@
 import com.uber.cadence.StickyExecutionAttributes;
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowExecutionInfo;
+import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.internal.common.WorkflowExecutionUtils;
 import com.uber.cadence.internal.testservice.RequestContext.Timer;
 import java.time.Duration;
@@ -49,7 +49,6 @@
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
-import org.apache.thrift.TException;
 
 class TestWorkflowStoreImpl implements TestWorkflowStore {
 
@@ -338,7 +337,8 @@ public void sendQueryTask(
 
   @Override
   public GetWorkflowExecutionHistoryResponse getWorkflowExecutionHistory(
-      ExecutionId executionId, GetWorkflowExecutionHistoryRequest getRequest) throws EntityNotExistsError {
+      ExecutionId executionId, GetWorkflowExecutionHistoryRequest getRequest)
+      throws EntityNotExistsError {
     HistoryStore history;
     // Used to eliminate the race condition on waitForNewEvents
     long expectedNextEventId;
@@ -348,19 +348,20 @@ public GetWorkflowExecutionHistoryResponse getWorkflowExecutionHistory(
       if (!getRequest.isWaitForNewEvent()
           && getRequest.getHistoryEventFilterType() != HistoryEventFilterType.CLOSE_EVENT) {
         List<HistoryEvent> events = history.getEventsLocked();
-        List<DataBlob> blobs = InternalUtils.DeserializeFromHistoryEventToBlobData(events);
+        List<DataBlob> blobs = InternalUtils.SerializeFromHistoryEventToBlobData(events);
         // Copy the list as it is mutable. Individual events assumed immutable.
         ArrayList<HistoryEvent> eventsCopy = new ArrayList<>(events);
         return new GetWorkflowExecutionHistoryResponse()
-            .setHistory(new History().setEvents(eventsCopy)).setRawHistory(blobs);
+            .setHistory(new History().setEvents(eventsCopy))
+            .setRawHistory(blobs);
       }
       expectedNextEventId = history.getNextEventIdLocked();
     } finally {
       lock.unlock();
     }
     List<HistoryEvent> events =
         history.waitForNewEvents(expectedNextEventId, getRequest.getHistoryEventFilterType());
-    List<DataBlob> blobs = InternalUtils.DeserializeFromHistoryEventToBlobData(events);
+    List<DataBlob> blobs = InternalUtils.SerializeFromHistoryEventToBlobData(events);
     GetWorkflowExecutionHistoryResponse result = new GetWorkflowExecutionHistoryResponse();
     if (events != null) {
       result.setHistory(new History().setEvents(events));