Merge remote-tracking branch 'upstream/master'

Author: sdwr98

.gitignore

@@ -9,4 +9,8 @@ target
 /out
 dummy
 $buildDir
-src/main/idls/*
+src/main/idls/*
+/bin
+.classpath
+.project
+.settings

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## v2.7.0
+- Add ParentClosePolicy to child workflows and also expose parent execution info for child workflows
+- Add context propagation
+- Fix various bugs around test workflow service and test mutable state implementation
+- Use thrift IDLs from uber/cadence-idl repo as a submodule
+- Various dependency updates including Docker base image and Gradle wrapper
+- Miscellaneous bug fixes 
+
 ## v2.6.3
 - Add Upsert Search Attributes
 - Support get search attributes inside workflow

README.md

@@ -31,12 +31,12 @@ Add *cadence-client* as a dependency to your *pom.xml*:
     <dependency>
       <groupId>com.uber.cadence</groupId>
       <artifactId>cadence-client</artifactId>
-      <version>2.6.3</version>
+      <version>2.7.0</version>
     </dependency>
 
 or to *build.gradle*:
 
-    compile group: 'com.uber.cadence', name: 'cadence-client', version: '2.6.3'
+    compile group: 'com.uber.cadence', name: 'cadence-client', version: '2.7.0'
 
 ## Documentation
 

build.gradle

@@ -37,7 +37,7 @@ googleJavaFormat {
 }
 
 group = 'com.uber.cadence'
-version = '2.6.3'
+version = '2.7.0'
 
 description = '''Uber Cadence Java Client'''
 

src/main/idls

@@ -18,6 +18,10 @@
 package com.uber.cadence.internal.common;
 
 import com.google.common.base.Defaults;
+import com.uber.cadence.DataBlob;
+import com.uber.cadence.History;
+import com.uber.cadence.HistoryEvent;
+import com.uber.cadence.HistoryEventFilterType;
 import com.uber.cadence.SearchAttributes;
 import com.uber.cadence.TaskList;
 import com.uber.cadence.TaskListKind;
@@ -27,10 +31,16 @@
 import com.uber.cadence.workflow.WorkflowMethod;
 import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
+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;
+import org.apache.thrift.TDeserializer;
+import org.apache.thrift.TException;
+import org.apache.thrift.TSerializer;
 
 /** Utility functions shared by the implementation code. */
 public final class InternalUtils {
@@ -143,6 +153,54 @@ public static SearchAttributes convertMapToSearchAttributes(
     return new SearchAttributes().setIndexedFields(mapOfByteBuffer);
   }
 
+  // This method deserialize the DataBlob data to the HistoriyEvent data
+  public static History DeserializeFromBlobToHistoryEvents(
+      List<DataBlob> blobData, HistoryEventFilterType historyEventFilterType) throws TException {
+
+    List<HistoryEvent> events = new ArrayList<HistoryEvent>();
+    for (DataBlob data : blobData) {
+      History history = new History();
+      try {
+        byte[] dataByte = data.getData();
+        dataByte = Arrays.copyOfRange(dataByte, 1, dataByte.length);
+        deSerializer.deserialize(history, dataByte);
+
+        if (history == null || history.getEvents() == null || history.getEvents().size() == 0) {
+          return null;
+        }
+      } catch (org.apache.thrift.TException err) {
+        throw new TException("Deserialize blob data to history event failed with unknown error");
+      }
+
+      events.addAll(history.getEvents());
+    }
+
+    if (events.size() > 0 && historyEventFilterType == HistoryEventFilterType.CLOSE_EVENT) {
+      events = events.subList(events.size() - 1, events.size());
+    }
+
+    return new History().setEvents(events);
+  }
+
+  // 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 RuntimeException("Serialize to blob data failed", err);
+      }
+      blobs.add(blob);
+    }
+
+    return blobs;
+  }
+
+  private static final TDeserializer deSerializer = new TDeserializer();
+  private static final TSerializer serializer = new TSerializer();
+
   /** Prohibit instantiation */
   private InternalUtils() {}
 }

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

@@ -71,6 +71,8 @@ public class ServiceMethod {
       MetricsType.CADENCE_METRICS_PREFIX + "RespondDecisionTaskFailed";
   public static final String SIGNAL_WORKFLOW_EXECUTION =
       MetricsType.CADENCE_METRICS_PREFIX + "SignalWorkflowExecution";
+  public static final String RESET_WORKFLOW_EXECUTION =
+      MetricsType.CADENCE_METRICS_PREFIX + "ResetWorkflowExecution";
   public static final String SIGNAL_WITH_START_WORKFLOW_EXECUTION =
       MetricsType.CADENCE_METRICS_PREFIX + "SignalWithStartWorkflowExecution";
   public static final String START_WORKFLOW_EXECUTION =

src/main/java/com/uber/cadence/internal/metrics/ServiceMethod.java

@@ -39,8 +39,6 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
@@ -88,8 +86,7 @@ public void accept(Exception reason) {
   private final Map<String, ExecuteLocalActivityParameters> unstartedLaTasks = new HashMap<>();
   private final ReplayDecider replayDecider;
   private final DataConverter dataConverter;
-  private final Lock laTaskLock = new ReentrantLock();
-  private final Condition taskCondition = laTaskLock.newCondition();
+  private final Condition taskCondition;
   private boolean taskCompleted = false;
 
   ClockDecisionContext(
@@ -98,6 +95,7 @@ public void accept(Exception reason) {
       ReplayDecider replayDecider,
       DataConverter dataConverter) {
     this.decisions = decisions;
+    this.taskCondition = replayDecider.getLock().newCondition();
     mutableSideEffectHandler =
         new MarkerHandler(decisions, MUTABLE_SIDE_EFFECT_MARKER_NAME, () -> replaying);
     versionHandler = new MarkerHandler(decisions, VERSION_MARKER_NAME, () -> replaying);
@@ -272,13 +270,9 @@ private void handleLocalActivityMarker(MarkerRecordedEventAttributes attributes)
       completionHandle.accept(marker.getResult(), failure);
       setReplayCurrentTimeMilliseconds(marker.getReplayTimeMillis());
 
-      laTaskLock.lock();
-      try {
-        taskCompleted = true;
-        taskCondition.signal();
-      } finally {
-        laTaskLock.unlock();
-      }
+      taskCompleted = true;
+      // This method is already called under the lock.
+      taskCondition.signal();
     }
   }
 
@@ -311,7 +305,8 @@ int getVersion(String changeId, DataConverter converter, int minSupported, int m
   }
 
   private void validateVersion(String changeID, int version, int minSupported, int maxSupported) {
-    if (version < minSupported || version > maxSupported) {
+    if ((version < minSupported || version > maxSupported)
+        && version != WorkflowInternal.DEFAULT_VERSION) {
       throw new Error(
           String.format(
               "Version %d of changeID %s is not supported. Supported version is between %d and %d.",
@@ -341,7 +336,7 @@ boolean startUnstartedLaTasks(Duration maxWaitAllowed) {
           laTaskPoller.apply(
               new LocalActivityWorker.Task(
                   params,
-                  replayDecider,
+                  replayDecider.getLocalActivityCompletionSink(),
                   replayDecider.getDecisionTimeoutSeconds(),
                   this::currentTimeMillis,
                   this::replayTimeUpdatedAtMillis),
@@ -359,14 +354,10 @@ int numPendingLaTasks() {
   }
 
   void awaitTaskCompletion(Duration duration) throws InterruptedException {
-    laTaskLock.lock();
-    try {
-      while (!taskCompleted) {
-        taskCondition.awaitNanos(duration.toNanos());
-      }
-      taskCompleted = false;
-    } finally {
-      laTaskLock.unlock();
+    while (!taskCompleted) {
+      // This call is called from already locked object
+      taskCondition.awaitNanos(duration.toNanos());
     }
+    taskCompleted = false;
   }
 }

src/main/java/com/uber/cadence/internal/replay/ClockDecisionContext.java

@@ -43,8 +43,7 @@ public interface DecisionContext extends ReplayAware {
 
   WorkflowExecution getWorkflowExecution();
 
-  // TODO: Add to Cadence
-  //    com.uber.cadence.WorkflowExecution getParentWorkflowExecution();
+  WorkflowExecution getParentWorkflowExecution();
 
   WorkflowType getWorkflowType();
 
@@ -87,11 +86,7 @@ public interface DecisionContext extends ReplayAware {
    */
   Map<String, Object> getPropagatedContexts();
 
-  /**
-   * Returns the set of configured context propagators
-   *
-   * @return
-   */
+  /** Returns the set of configured context propagators */
   List<ContextPropagator> getContextPropagators();
 
   /**