Merge branch 'master' of github.com:motusllc/cadence-java-client

Author: sdwr98

.buildkite/pipeline.yml

@@ -1,4 +1,9 @@
 steps:
+  - label: "fossa analyze"
+    agents:
+      queue: "init"
+      docker: "*"
+    command: ".buildkite/scripts/fossa.sh"
   - label: ":java: Unit test with test services"
     agents:
       queue: "workers"
@@ -28,4 +33,4 @@ steps:
       - docker-compose#v3.0.0:
           run: unit-test-docker-sticky-off
           config: docker/buildkite/docker-compose.yaml
-  - wait
+  - wait

.buildkite/scripts/fossa.sh

@@ -0,0 +1,14 @@
+#!/bin/bash
+
+set -exo pipefail
+
+curl -H 'Cache-Control: no-cache' https://raw.githubusercontent.com/fossas/fossa-cli/master/install.sh | bash -s -- -b ~/
+
+~/fossa init
+~/fossa analyze
+
+# Capture the exit status
+EXIT_STATUS=$?
+
+echo "fossa script exits with status $EXIT_STATUS"
+exit $EXIT_STATUS

.fossa.yml

@@ -0,0 +1,11 @@
+version: 2
+cli:
+  server: https://app.fossa.com
+  fetcher: custom
+  project: [email protected]:uber/cadence-java-client.git
+analyze:
+  modules:
+  - name: .
+    type: gradle
+    target: ':'
+    path: .

CONTRIBUTING.md

@@ -7,6 +7,7 @@ This doc is intended for contributors to `cadence-java-client` (hopefully that's
 ## Development Environment
 
 * Java 8.
+* Thrift 0.9.3
 * Gradle build tool
 * Docker
 

build.gradle

@@ -82,7 +82,7 @@ task updateDlsSubmodule(type: Exec) {
 
 compileThrift {
     dependsOn updateDlsSubmodule
-    sourceItems "src/main/idls/thrift/cadence.thrift","src/main/idls/thrift/shared.thrift"
+    sourceItems "${projectDir}/src/main/idls/thrift/cadence.thrift","${projectDir}/src/main/idls/thrift/shared.thrift"
 
     nowarn true
 }

src/main/idls

@@ -21,7 +21,9 @@
 
 import com.uber.cadence.common.MethodRetry;
 import com.uber.cadence.common.RetryOptions;
+import com.uber.cadence.context.ContextPropagator;
 import java.time.Duration;
+import java.util.List;
 import java.util.Objects;
 
 /** Options used to configure how an local activity is invoked. */
@@ -50,12 +52,14 @@ public static LocalActivityOptions merge(
             ActivityOptions.mergeDuration(
                 a.scheduleToCloseTimeoutSeconds(), o.getScheduleToCloseTimeout()))
         .setRetryOptions(RetryOptions.merge(r, o.getRetryOptions()))
+        .setContextPropagators(o.getContextPropagators())
         .validateAndBuildWithDefaults();
   }
 
   public static final class Builder {
     private Duration scheduleToCloseTimeout;
     private RetryOptions retryOptions;
+    private List<ContextPropagator> contextPropagators;
 
     public Builder() {}
 
@@ -83,25 +87,32 @@ public Builder setRetryOptions(RetryOptions retryOptions) {
       return this;
     }
 
+    public Builder setContextPropagators(List<ContextPropagator> contextPropagators) {
+      this.contextPropagators = contextPropagators;
+      return this;
+    }
+
     public LocalActivityOptions build() {
-      return new LocalActivityOptions(scheduleToCloseTimeout, retryOptions);
+      return new LocalActivityOptions(scheduleToCloseTimeout, retryOptions, contextPropagators);
     }
 
     public LocalActivityOptions validateAndBuildWithDefaults() {
       RetryOptions ro = null;
       if (retryOptions != null) {
         ro = new RetryOptions.Builder(retryOptions).validateBuildWithDefaults();
       }
-      return new LocalActivityOptions(roundUpToSeconds(scheduleToCloseTimeout), ro);
+      return new LocalActivityOptions(roundUpToSeconds(scheduleToCloseTimeout), ro, contextPropagators);
     }
   }
 
   private final Duration scheduleToCloseTimeout;
   private final RetryOptions retryOptions;
+  private final List<ContextPropagator> contextPropagators;
 
-  private LocalActivityOptions(Duration scheduleToCloseTimeout, RetryOptions retryOptions) {
+  private LocalActivityOptions(Duration scheduleToCloseTimeout, RetryOptions retryOptions, List<ContextPropagator> contextPropagators) {
     this.scheduleToCloseTimeout = scheduleToCloseTimeout;
     this.retryOptions = retryOptions;
+    this.contextPropagators = contextPropagators;
   }
 
   public Duration getScheduleToCloseTimeout() {
@@ -112,6 +123,10 @@ public RetryOptions getRetryOptions() {
     return retryOptions;
   }
 
+  public List<ContextPropagator> getContextPropagators() {
+    return contextPropagators;
+  }
+
   @Override
   public String toString() {
     return "LocalActivityOptions{"

src/main/java/com/uber/cadence/activity/LocalActivityOptions.java

@@ -137,6 +137,11 @@ public Builder setWorkflowId(String workflowId) {
      * <ul>
      *   RejectDuplicate doesn't allow new run independently of the previous run closure status.
      * </ul>
+     *
+     * <ul>
+     *   TerminateIfRunning terminate current running workflow using the same workflow ID if exist,
+     *   then start a new run in one transaction
+     * </ul>
      */
     public Builder setWorkflowIdReusePolicy(WorkflowIdReusePolicy workflowIdReusePolicy) {
       this.workflowIdReusePolicy = workflowIdReusePolicy;

src/main/java/com/uber/cadence/client/WorkflowOptions.java

@@ -20,6 +20,8 @@
 import com.uber.cadence.QueryRejectCondition;
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.internal.common.QueryResponse;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Proxy;
 import java.lang.reflect.Type;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -44,19 +46,32 @@ public interface WorkflowStub {
    * @return untyped workflow stub for the same workflow instance.
    */
   static <T> WorkflowStub fromTyped(T typed) {
-    if (!(typed instanceof Supplier)) {
+    if (!(typed instanceof Proxy)) {
       throw new IllegalArgumentException(
           "arguments must be created through WorkflowClient.newWorkflowStub");
     }
+
+    InvocationHandler handler = Proxy.getInvocationHandler(typed);
+
+    if (!(handler instanceof Supplier)) {
+      throw new IllegalArgumentException(
+          "arguments must be created through WorkflowClient.newWorkflowStub");
+    }
+
     @SuppressWarnings("unchecked")
-    Supplier<WorkflowStub> supplier = (Supplier<WorkflowStub>) typed;
+    Supplier<WorkflowStub> supplier = (Supplier<WorkflowStub>) handler;
     return supplier.get();
   }
 
   void signal(String signalName, Object... args);
 
   WorkflowExecution start(Object... args);
 
+  CompletableFuture<WorkflowExecution> startAsync(Object... args);
+
+  CompletableFuture<WorkflowExecution> startAsyncWithTimeout(
+      long timeout, TimeUnit unit, Object... args);
+
   WorkflowExecution signalWithStart(String signalName, Object[] signalArgs, Object[] startArgs);
 
   Optional<String> getWorkflowType();

src/main/java/com/uber/cadence/client/WorkflowStub.java

@@ -43,7 +43,7 @@ public class Version {
    * support. This can be used for client capibility check, on Cadence server, for backward
    * compatibility Format: MAJOR.MINOR.PATCH
    */
-  public static final String FEATURE_VERSION = "1.2.0";
+  public static final String FEATURE_VERSION = "1.3.0";
 
   static {
     // Load version from version.properties generated by Gradle into build/resources/main directory.

src/main/java/com/uber/cadence/internal/Version.java

@@ -125,7 +125,7 @@ public static byte[] getWorkflowExecutionResult(
       TimeUnit unit)
       throws TimeoutException, CancellationException, WorkflowExecutionFailedException,
           WorkflowTerminatedException, WorkflowTimedOutException, EntityNotExistsError {
-    // getIntanceCloseEvent waits for workflow completion including new runs.
+    // getInstanceCloseEvent waits for workflow completion including new runs.
     HistoryEvent closeEvent =
         getInstanceCloseEvent(service, domain, workflowExecution, timeout, unit);
     return getResultFromCloseEvent(workflowExecution, workflowType, closeEvent);
@@ -213,9 +213,12 @@ private static HistoryEvent getInstanceCloseEvent(
       r.setNextPageToken(pageToken);
       r.setWaitForNewEvent(true);
       r.setSkipArchival(true);
+      RetryOptions retryOptions = getRetryOptionWithTimeout(timeout, unit);
       try {
         response =
-            Retryer.retryWithResult(retryParameters, () -> service.GetWorkflowExecutionHistory(r));
+            Retryer.retryWithResult(
+                retryOptions,
+                () -> service.GetWorkflowExecutionHistoryWithTimeout(r, unit.toMillis(timeout)));
       } catch (EntityNotExistsError e) {
         if (e.activeCluster != null
             && e.currentCluster != null
@@ -289,12 +292,14 @@ private static CompletableFuture<HistoryEvent> getInstanceCloseEventAsync(
     request.setDomain(domain);
     request.setExecution(workflowExecution);
     request.setHistoryEventFilterType(HistoryEventFilterType.CLOSE_EVENT);
+    request.setWaitForNewEvent(true);
     request.setNextPageToken(pageToken);
     CompletableFuture<GetWorkflowExecutionHistoryResponse> response =
-        getWorkflowExecutionHistoryAsync(service, request);
+        getWorkflowExecutionHistoryAsync(service, request, timeout, unit);
     return response.thenComposeAsync(
         (r) -> {
-          if (timeout != 0 && System.currentTimeMillis() - start > unit.toMillis(timeout)) {
+          long elapsedTime = System.currentTimeMillis() - start;
+          if (timeout != 0 && elapsedTime > unit.toMillis(timeout)) {
             throw CheckedExceptionWrapper.wrap(
                 new TimeoutException(
                     "WorkflowId="
@@ -310,7 +315,7 @@ private static CompletableFuture<HistoryEvent> getInstanceCloseEventAsync(
           if (history == null || history.getEvents().size() == 0) {
             // Empty poll returned
             return getInstanceCloseEventAsync(
-                service, domain, workflowExecution, pageToken, timeout, unit);
+                service, domain, workflowExecution, pageToken, timeout - elapsedTime, unit);
           }
           HistoryEvent event = history.getEvents().get(0);
           if (!isWorkflowExecutionCompletedEvent(event)) {
@@ -326,21 +331,36 @@ private static CompletableFuture<HistoryEvent> getInstanceCloseEventAsync(
                             .getWorkflowExecutionContinuedAsNewEventAttributes()
                             .getNewExecutionRunId());
             return getInstanceCloseEventAsync(
-                service, domain, nextWorkflowExecution, r.getNextPageToken(), timeout, unit);
+                service,
+                domain,
+                nextWorkflowExecution,
+                r.getNextPageToken(),
+                timeout - elapsedTime,
+                unit);
           }
           return CompletableFuture.completedFuture(event);
         });
   }
 
+  private static RetryOptions getRetryOptionWithTimeout(long timeout, TimeUnit unit) {
+    return new RetryOptions.Builder(retryParameters)
+        .setExpiration(Duration.ofSeconds(unit.toSeconds(timeout)))
+        .build();
+  }
+
   private static CompletableFuture<GetWorkflowExecutionHistoryResponse>
       getWorkflowExecutionHistoryAsync(
-          IWorkflowService service, GetWorkflowExecutionHistoryRequest r) {
+          IWorkflowService service,
+          GetWorkflowExecutionHistoryRequest r,
+          long timeout,
+          TimeUnit unit) {
+    RetryOptions retryOptions = getRetryOptionWithTimeout(timeout, unit);
     return Retryer.retryWithResultAsync(
-        retryParameters,
+        retryOptions,
         () -> {
           CompletableFuture<GetWorkflowExecutionHistoryResponse> result = new CompletableFuture<>();
           try {
-            service.GetWorkflowExecutionHistory(
+            service.GetWorkflowExecutionHistoryWithTimeout(
                 r,
                 new AsyncMethodCallback<GetWorkflowExecutionHistoryResponse>() {
                   @Override
@@ -352,7 +372,8 @@ public void onComplete(GetWorkflowExecutionHistoryResponse response) {
                   public void onError(Exception exception) {
                     result.completeExceptionally(exception);
                   }
-                });
+                },
+                unit.toMillis(timeout));
           } catch (TException e) {
             result.completeExceptionally(e);
           }

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

@@ -26,12 +26,19 @@
 import com.uber.cadence.internal.replay.QueryWorkflowParameters;
 import com.uber.cadence.internal.replay.SignalExternalWorkflowParameters;
 import com.uber.cadence.serviceclient.IWorkflowService;
+import java.util.concurrent.CompletableFuture;
 
 public interface GenericWorkflowClientExternal {
 
   WorkflowExecution startWorkflow(StartWorkflowExecutionParameters startParameters)
       throws WorkflowExecutionAlreadyStartedError;
 
+  CompletableFuture<WorkflowExecution> startWorkflowAsync(
+      StartWorkflowExecutionParameters startParameters);
+
+  CompletableFuture<WorkflowExecution> startWorkflowAsync(
+      StartWorkflowExecutionParameters startParameters, Long timeoutInMillis);
+
   void signalWorkflowExecution(SignalExternalWorkflowParameters signalParameters);
 
   WorkflowExecution signalWithStartWorkflowExecution(