port over changeset from our fork

Author: AngerM-DD

build.gradle

@@ -53,7 +53,7 @@ dependencies {
     errorproneJavac('com.google.errorprone:javac:9+181-r4173-1')
     errorprone('com.google.errorprone:error_prone_core:2.3.4')
 
-    compile group: 'com.uber.tchannel', name: 'tchannel-core', version: '0.8.5'
+    compile group: 'com.uber.tchannel', name: 'tchannel-core', version: '0.8.30'
     compile group: 'org.slf4j', name: 'slf4j-api', version: '1.7.25'
     compile group: 'org.apache.thrift', name: 'libthrift', version: '0.9.3'
     compile group: 'com.google.code.gson', name: 'gson', version: '2.8.6'
@@ -62,6 +62,7 @@ dependencies {
     compile group: 'com.cronutils', name: 'cron-utils', version: '9.0.0'
     compile group: 'io.micrometer', name: 'micrometer-core', version: '1.1.2'
     compile group: 'javax.annotation', name: 'javax.annotation-api', version: '1.3.2'
+    compile group: 'io.opentelemetry', name: 'opentelemetry-sdk', version: '1.1.0'
 
     testCompile group: 'junit', name: 'junit', version: '4.12'
     testCompile group: 'com.googlecode.junit-toolbox', name: 'junit-toolbox', version: '2.4'

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

@@ -30,9 +30,11 @@
 import com.uber.cadence.common.MethodRetry;
 import com.uber.cadence.common.RetryOptions;
 import com.uber.cadence.context.ContextPropagator;
+import com.uber.cadence.context.OpenTelemetryContextPropagator;
 import com.uber.cadence.internal.common.OptionsUtils;
 import com.uber.cadence.workflow.WorkflowMethod;
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -66,6 +68,7 @@ public static WorkflowOptions merge(
         .setMemo(o.getMemo())
         .setSearchAttributes(o.getSearchAttributes())
         .setContextPropagators(o.getContextPropagators())
+        .setDefaultContextPropagators(o.useDefaultContextPropagators)
         .validateBuildWithDefaults();
   }
 
@@ -91,6 +94,8 @@ public static final class Builder {
 
     private List<ContextPropagator> contextPropagators;
 
+    private Boolean useDefaultContextPropagators;
+
     public Builder() {}
 
     public Builder(WorkflowOptions o) {
@@ -107,6 +112,7 @@ public Builder(WorkflowOptions o) {
       this.memo = o.memo;
       this.searchAttributes = o.searchAttributes;
       this.contextPropagators = o.contextPropagators;
+      this.useDefaultContextPropagators = o.useDefaultContextPropagators;
     }
 
     /**
@@ -214,6 +220,13 @@ public Builder setContextPropagators(List<ContextPropagator> contextPropagators)
       return this;
     }
 
+    /** Specifies that the default context propagators should not be used. */
+    public Builder setDefaultContextPropagators(Boolean useDefaultContextPropagators) {
+      this.useDefaultContextPropagators =
+          (useDefaultContextPropagators == null || useDefaultContextPropagators);
+      return this;
+    }
+
     public WorkflowOptions build() {
       return new WorkflowOptions(
           workflowId,
@@ -225,7 +238,8 @@ public WorkflowOptions build() {
           cronSchedule,
           memo,
           searchAttributes,
-          contextPropagators);
+          contextPropagators,
+          useDefaultContextPropagators);
     }
 
     /**
@@ -261,6 +275,20 @@ public WorkflowOptions validateBuildWithDefaults() {
         cron.validate();
       }
 
+      if (useDefaultContextPropagators == null || useDefaultContextPropagators) {
+        // Add OpenTelemetry propagator if not already present.
+        if (contextPropagators != null) {
+          contextPropagators = new ArrayList(contextPropagators);
+        } else {
+          contextPropagators = new ArrayList<>();
+        }
+
+        OpenTelemetryContextPropagator otelPropagator = new OpenTelemetryContextPropagator();
+        if (!contextPropagators.contains(otelPropagator)) {
+          contextPropagators.add(otelPropagator);
+        }
+      }
+
       return new WorkflowOptions(
           workflowId,
           policy,
@@ -272,7 +300,8 @@ public WorkflowOptions validateBuildWithDefaults() {
           cronSchedule,
           memo,
           searchAttributes,
-          contextPropagators);
+          contextPropagators,
+          useDefaultContextPropagators);
     }
   }
 
@@ -296,6 +325,8 @@ public WorkflowOptions validateBuildWithDefaults() {
 
   private List<ContextPropagator> contextPropagators;
 
+  private Boolean useDefaultContextPropagators;
+
   private WorkflowOptions(
       String workflowId,
       WorkflowIdReusePolicy workflowIdReusePolicy,
@@ -306,7 +337,8 @@ private WorkflowOptions(
       String cronSchedule,
       Map<String, Object> memo,
       Map<String, Object> searchAttributes,
-      List<ContextPropagator> contextPropagators) {
+      List<ContextPropagator> contextPropagators,
+      Boolean useDefaultContextPropagators) {
     this.workflowId = workflowId;
     this.workflowIdReusePolicy = workflowIdReusePolicy;
     this.executionStartToCloseTimeout = executionStartToCloseTimeout;
@@ -317,6 +349,7 @@ private WorkflowOptions(
     this.memo = memo;
     this.searchAttributes = searchAttributes;
     this.contextPropagators = contextPropagators;
+    this.useDefaultContextPropagators = useDefaultContextPropagators;
   }
 
   public String getWorkflowId() {
@@ -373,7 +406,9 @@ public boolean equals(Object o) {
         && Objects.equals(cronSchedule, that.cronSchedule)
         && Objects.equals(memo, that.memo)
         && Objects.equals(searchAttributes, that.searchAttributes)
-        && Objects.equals(contextPropagators, that.contextPropagators);
+        && Objects.equals(contextPropagators, that.contextPropagators)
+        && (useDefaultContextPropagators == null || useDefaultContextPropagators)
+            == (that.useDefaultContextPropagators == null || that.useDefaultContextPropagators);
   }
 
   @Override
@@ -388,7 +423,8 @@ public int hashCode() {
         cronSchedule,
         memo,
         searchAttributes,
-        contextPropagators);
+        contextPropagators,
+        useDefaultContextPropagators);
   }
 
   @Override
@@ -418,6 +454,8 @@ public String toString() {
         + searchAttributes
         + ", contextPropagators='"
         + contextPropagators
+        + ", useDefaultContextPropagators='"
+        + useDefaultContextPropagators
         + '\''
         + '}';
   }

src/main/java/com/uber/cadence/context/ContextPropagator.java

@@ -23,6 +23,9 @@
  * Context Propagators are used to propagate information from workflow to activity, workflow to
  * child workflow, and workflow to child thread (using {@link com.uber.cadence.workflow.Async}).
  *
+ * <p>It is important to note that all threads share one ContextPropagator instance, so your
+ * implementation <b>must</b> be thread-safe and store any state in ThreadLocal variables.
+ *
  * <p>A sample <code>ContextPropagator</code> that copies all {@link org.slf4j.MDC} entries starting
  * with a given prefix along the code path looks like this:
  *
@@ -136,4 +139,31 @@ public interface ContextPropagator {
 
   /** Sets the current context */
   void setCurrentContext(Object context);
+
+  /**
+   * This is a lifecycle method, called after the context has been propagated to the
+   * workflow/activity thread but the workflow/activity has not yet started.
+   */
+  default void setUp() {
+    // No-op
+  }
+
+  /**
+   * This is a lifecycle method, called after the workflow/activity has completed. If the method
+   * finished without exception, {@code successful} will be true. Otherwise, it will be false and
+   * {@link #onError(Throwable)} will have already been called.
+   */
+  default void finish() {
+    // No-op
+  }
+
+  /**
+   * This is a lifecycle method, called when the workflow/activity finishes by throwing an unhandled
+   * exception. {@link #finish()} is called after this method.
+   *
+   * @param t The unhandled exception that caused the workflow/activity to terminate
+   */
+  default void onError(Throwable t) {
+    // No-op
+  }
 }

src/main/java/com/uber/cadence/internal/context/ContextThreadLocal.java

@@ -24,10 +24,14 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-/** This class holds the current set of context propagators */
+/** This class holds the current set of context propagators. */
 public class ContextThreadLocal {
 
+  private static final Logger log = LoggerFactory.getLogger(ContextThreadLocal.class);
+
   private static WorkflowThreadLocal<List<ContextPropagator>> contextPropagators =
       WorkflowThreadLocal.withInitial(
           new Supplier<List<ContextPropagator>>() {
@@ -37,7 +41,7 @@ public List<ContextPropagator> get() {
             }
           });
 
-  /** Sets the list of context propagators for the thread */
+  /** Sets the list of context propagators for the thread. */
   public static void setContextPropagators(List<ContextPropagator> propagators) {
     if (propagators == null || propagators.isEmpty()) {
       return;
@@ -57,6 +61,11 @@ public static Map<String, Object> getCurrentContextForPropagation() {
     return contextData;
   }
 
+  /**
+   * Injects the context data into the thread for each configured context propagator.
+   *
+   * @param contextData The context data received from the server.
+   */
   public static void propagateContextToCurrentThread(Map<String, Object> contextData) {
     if (contextData == null || contextData.isEmpty()) {
       return;
@@ -67,4 +76,44 @@ public static void propagateContextToCurrentThread(Map<String, Object> contextDa
       }
     }
   }
+
+  /** Calls {@link ContextPropagator#setUp()} for each propagator. */
+  public static void setUpContextPropagators() {
+    for (ContextPropagator propagator : contextPropagators.get()) {
+      try {
+        propagator.setUp();
+      } catch (Throwable t) {
+        // Don't let an error in one propagator block the others
+        log.error("Error calling setUp() on a contextpropagator", t);
+      }
+    }
+  }
+
+  /**
+   * Calls {@link ContextPropagator#onError(Throwable)} for each propagator.
+   *
+   * @param t The Throwable that caused the workflow/activity to finish.
+   */
+  public static void onErrorContextPropagators(Throwable t) {
+    for (ContextPropagator propagator : contextPropagators.get()) {
+      try {
+        propagator.onError(t);
+      } catch (Throwable t1) {
+        // Don't let an error in one propagator block the others
+        log.error("Error calling onError() on a contextpropagator", t1);
+      }
+    }
+  }
+
+  /** Calls {@link ContextPropagator#finish()} for each propagator. */
+  public static void finishContextPropagators() {
+    for (ContextPropagator propagator : contextPropagators.get()) {
+      try {
+        propagator.finish();
+      } catch (Throwable t) {
+        // Don't let an error in one propagator block the others
+        log.error("Error calling finish() on a contextpropagator", t);
+      }
+    }
+  }
 }

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

@@ -23,6 +23,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 final class WorkflowContext {
 
@@ -166,7 +167,18 @@ Map<String, Object> getPropagatedContexts() {
 
     Map<String, Object> contextData = new HashMap<>();
     for (ContextPropagator propagator : contextPropagators) {
-      contextData.put(propagator.getName(), propagator.deserializeContext(headerData));
+      // Only send the context propagator the fields that belong to them
+      // Change the map from MyPropagator:foo -> bar to foo -> bar
+      Map<String, byte[]> filteredData =
+          headerData
+              .entrySet()
+              .stream()
+              .filter(e -> e.getKey().startsWith(propagator.getName()))
+              .collect(
+                  Collectors.toMap(
+                      e -> e.getKey().substring(propagator.getName().length() + 1),
+                      Map.Entry::getValue));
+      contextData.put(propagator.getName(), propagator.deserializeContext(filteredData));
     }
 
     return contextData;

src/main/java/com/uber/cadence/internal/sync/SyncDecisionContext.java

@@ -72,6 +72,7 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -449,7 +450,18 @@ private Map<String, byte[]> extractContextsAndConvertToBytes(
     }
     Map<String, byte[]> result = new HashMap<>();
     for (ContextPropagator propagator : contextPropagators) {
-      result.putAll(propagator.serializeContext(propagator.getCurrentContext()));
+      // Get the serialized context from the propagator
+      Map<String, byte[]> serializedContext =
+          propagator.serializeContext(propagator.getCurrentContext());
+      // Namespace each entry in case of overlaps, so foo -> bar becomes MyPropagator:foo -> bar
+      Map<String, byte[]> namespacedSerializedContext =
+          serializedContext
+              .entrySet()
+              .stream()
+              .collect(
+                  Collectors.toMap(
+                      e -> propagator.getName() + ":" + e.getKey(), Map.Entry::getValue));
+      result.putAll(namespacedSerializedContext);
     }
     return result;
   }

src/main/java/com/uber/cadence/internal/sync/SyncWorkflowWorker.java

@@ -39,10 +39,7 @@
 import java.lang.reflect.Type;
 import java.time.Duration;
 import java.util.Objects;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
@@ -59,6 +56,7 @@ public class SyncWorkflowWorker
   private final ScheduledExecutorService ldaHeartbeatExecutor = Executors.newScheduledThreadPool(4);
   private SuspendableWorker ldaWorker;
   private POJOActivityTaskHandler ldaTaskHandler;
+  private final IWorkflowService service;
 
   public SyncWorkflowWorker(
       IWorkflowService service,
@@ -74,6 +72,7 @@ public SyncWorkflowWorker(
       ThreadPoolExecutor workflowThreadPool) {
     Objects.requireNonNull(workflowThreadPool);
     this.dataConverter = workflowOptions.getDataConverter();
+    this.service = service;
 
     factory =
         new POJOWorkflowImplementationFactory(
@@ -252,4 +251,8 @@ public <R> R queryWorkflowExecution(
   public void accept(PollForDecisionTaskResponse pollForDecisionTaskResponse) {
     workflowWorker.accept(pollForDecisionTaskResponse);
   }
+
+  public CompletableFuture<Boolean> isHealthy() {
+    return service.isHealthy();
+  }
 }