do not dispatch activities locally if taskListActivitiesPerSecond is set (#567)

Author: mkolodezny

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

@@ -27,6 +27,7 @@
 import com.uber.cadence.internal.worker.DecisionTaskHandler;
 import com.uber.cadence.internal.worker.LocalActivityWorker;
 import com.uber.cadence.internal.worker.LocallyDispatchedActivityWorker;
+import com.uber.cadence.internal.worker.LocallyDispatchedActivityWorker.Task;
 import com.uber.cadence.internal.worker.SingleWorkerOptions;
 import com.uber.cadence.internal.worker.SuspendableWorker;
 import com.uber.cadence.internal.worker.WorkflowWorker;
@@ -44,20 +45,21 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 
-/** Workflow worker that supports POJO workflow implementations. */
+/**
+ * Workflow worker that supports POJO workflow implementations.
+ */
 public class SyncWorkflowWorker
     implements SuspendableWorker, Consumer<PollForDecisionTaskResponse> {
 
   private final WorkflowWorker workflowWorker;
   private final LocalActivityWorker laWorker;
-  private final LocallyDispatchedActivityWorker ldaWorker;
-
   private final POJOWorkflowImplementationFactory factory;
   private final DataConverter dataConverter;
   private final POJOActivityTaskHandler laTaskHandler;
-  private final POJOActivityTaskHandler ldaTaskHandler;
   private final ScheduledExecutorService heartbeatExecutor = Executors.newScheduledThreadPool(4);
   private final ScheduledExecutorService ldaHeartbeatExecutor = Executors.newScheduledThreadPool(4);
+  private LocallyDispatchedActivityWorker ldaWorker;
+  private POJOActivityTaskHandler ldaTaskHandler;
 
   public SyncWorkflowWorker(
       IWorkflowService service,
@@ -97,15 +99,21 @@ public SyncWorkflowWorker(
             stickyDecisionScheduleToStartTimeout,
             service,
             laWorker.getLocalActivityTaskPoller());
-    ldaTaskHandler =
-        new POJOActivityTaskHandler(
-            service,
-            domain,
-            locallyDispatchedActivityOptions.getDataConverter(),
-            ldaHeartbeatExecutor);
-    ldaWorker =
-        new LocallyDispatchedActivityWorker(
-            service, domain, taskList, locallyDispatchedActivityOptions, ldaTaskHandler);
+
+    Function<Task, Boolean> locallyDispatchedActivityTaskPoller = null;
+    // do not dispatch locally if TaskListActivitiesPerSecond is set
+    if (locallyDispatchedActivityOptions.getTaskListActivitiesPerSecond() == 0) {
+      ldaTaskHandler =
+          new POJOActivityTaskHandler(
+              service,
+              domain,
+              locallyDispatchedActivityOptions.getDataConverter(),
+              ldaHeartbeatExecutor);
+      ldaWorker =
+          new LocallyDispatchedActivityWorker(
+              service, domain, taskList, locallyDispatchedActivityOptions, ldaTaskHandler);
+      locallyDispatchedActivityTaskPoller = ldaWorker.getLocallyDispatchedActivityTaskPoller();
+    }
 
     workflowWorker =
         new WorkflowWorker(
@@ -114,7 +122,7 @@ public SyncWorkflowWorker(
             taskList,
             workflowOptions,
             taskHandler,
-            ldaWorker.getLocallyDispatchedActivityTaskPoller(),
+            locallyDispatchedActivityTaskPoller,
             stickyTaskListName);
   }
 
@@ -137,7 +145,9 @@ public void setLocalActivitiesImplementation(Object... activitiesImplementation)
   }
 
   public void setActivitiesImplementationToDispatchLocally(Object... activitiesImplementation) {
-    this.ldaTaskHandler.setActivitiesImplementation(activitiesImplementation);
+    if (this.ldaTaskHandler != null) {
+      this.ldaTaskHandler.setActivitiesImplementation(activitiesImplementation);
+    }
   }
 
   @Override
@@ -147,69 +157,84 @@ public void start() {
     // to start LocalActivity Worker.
     if (workflowWorker.isStarted()) {
       laWorker.start();
-      ldaWorker.start();
+      if (ldaWorker != null) {
+        ldaWorker.start();
+      }
     }
   }
 
   @Override
   public boolean isStarted() {
-    return workflowWorker.isStarted() && laWorker.isStarted() && ldaWorker.isStarted();
+    return workflowWorker.isStarted() && laWorker.isStarted() && (ldaWorker == null || ldaWorker
+        .isStarted());
   }
 
   @Override
   public boolean isShutdown() {
-    return workflowWorker.isShutdown() && laWorker.isShutdown() && ldaWorker.isShutdown();
+    return workflowWorker.isShutdown() && laWorker.isShutdown() && (ldaWorker == null || ldaWorker
+        .isShutdown());
   }
 
   @Override
   public boolean isTerminated() {
     return workflowWorker.isTerminated()
         && laWorker.isTerminated()
         && ldaHeartbeatExecutor.isTerminated()
-        && ldaWorker.isTerminated();
+        && (ldaWorker == null || ldaWorker.isTerminated());
   }
 
   @Override
   public void shutdown() {
     laWorker.shutdown();
     ldaHeartbeatExecutor.shutdown();
-    ldaWorker.shutdown();
+    if (ldaWorker != null) {
+      ldaWorker.shutdown();
+    }
     workflowWorker.shutdown();
   }
 
   @Override
   public void shutdownNow() {
     laWorker.shutdownNow();
     ldaHeartbeatExecutor.shutdownNow();
-    ldaWorker.shutdownNow();
+    if (ldaWorker != null) {
+      ldaWorker.shutdownNow();
+    }
     workflowWorker.shutdownNow();
   }
 
   @Override
   public void awaitTermination(long timeout, TimeUnit unit) {
     long timeoutMillis = InternalUtils.awaitTermination(laWorker, unit.toMillis(timeout));
     timeoutMillis = InternalUtils.awaitTermination(ldaHeartbeatExecutor, timeoutMillis);
-    timeoutMillis = InternalUtils.awaitTermination(ldaWorker, timeoutMillis);
+    if (ldaWorker != null) {
+      timeoutMillis = InternalUtils.awaitTermination(ldaWorker, timeoutMillis);
+    }
     InternalUtils.awaitTermination(workflowWorker, timeoutMillis);
   }
 
   @Override
   public void suspendPolling() {
     workflowWorker.suspendPolling();
     laWorker.suspendPolling();
-    ldaWorker.suspendPolling();
+    if (ldaWorker != null) {
+      ldaWorker.suspendPolling();
+    }
   }
 
   @Override
   public void resumePolling() {
     workflowWorker.resumePolling();
     laWorker.resumePolling();
-    ldaWorker.resumePolling();
+    if (ldaWorker != null) {
+      ldaWorker.resumePolling();
+    }
   }
 
   @Override
   public boolean isSuspended() {
-    return workflowWorker.isSuspended() && laWorker.isSuspended() && ldaWorker.isSuspended();
+    return workflowWorker.isSuspended() && laWorker.isSuspended() && (ldaWorker == null || ldaWorker
+        .isSuspended());
   }
 
   public <R> R queryWorkflowExecution(

src/main/java/com/uber/cadence/internal/worker/SingleWorkerOptions.java

@@ -173,7 +173,7 @@ PollerOptions getPollerOptions() {
     return pollerOptions;
   }
 
-  double getTaskListActivitiesPerSecond() {
+  public double getTaskListActivitiesPerSecond() {
     return taskListActivitiesPerSecond;
   }
 

src/main/java/com/uber/cadence/internal/worker/WorkflowWorker.java

@@ -253,39 +253,41 @@ private void sendReply(
               RespondDecisionTaskCompletedResponse taskCompletedResponse = null;
               List<Task> activityTasks = new ArrayList<>();
               try {
-                for (Decision decision : taskCompleted.getDecisions()) {
-                  ScheduleActivityTaskDecisionAttributes attr =
-                      decision.getScheduleActivityTaskDecisionAttributes();
-                  if (attr != null && taskList.equals(attr.getTaskList().getName())) {
-                    // assume the activity type is in registry otherwise the activity would be
-                    // failed and retried from server
-                    Task activityTask =
-                        new Task(
-                            attr.getActivityId(),
-                            attr.getActivityType(),
-                            attr.bufferForInput(),
-                            attr.getScheduleToCloseTimeoutSeconds(),
-                            attr.getStartToCloseTimeoutSeconds(),
-                            attr.getHeartbeatTimeoutSeconds(),
-                            task.getWorkflowType(),
-                            domain,
-                            attr.getHeader(),
-                            task.getWorkflowExecution());
-                    if (ldaTaskPoller.apply(activityTask)) {
-                      options
-                          .getMetricsScope()
-                          .counter(MetricsType.ACTIVITY_LOCAL_DISPATCH_SUCCEED_COUNTER)
-                          .inc(1);
-                      decision
-                          .getScheduleActivityTaskDecisionAttributes()
-                          .setRequestLocalDispatch(true);
-                      activityTasks.add(activityTask);
-                    } else {
-                      // all pollers are busy - no room to optimize
-                      options
-                          .getMetricsScope()
-                          .counter(MetricsType.ACTIVITY_LOCAL_DISPATCH_FAILED_COUNTER)
-                          .inc(1);
+                if (ldaTaskPoller != null) {
+                  for (Decision decision : taskCompleted.getDecisions()) {
+                    ScheduleActivityTaskDecisionAttributes attr =
+                        decision.getScheduleActivityTaskDecisionAttributes();
+                    if (attr != null && taskList.equals(attr.getTaskList().getName())) {
+                      // assume the activity type is in registry otherwise the activity would be
+                      // failed and retried from server
+                      Task activityTask =
+                          new Task(
+                              attr.getActivityId(),
+                              attr.getActivityType(),
+                              attr.bufferForInput(),
+                              attr.getScheduleToCloseTimeoutSeconds(),
+                              attr.getStartToCloseTimeoutSeconds(),
+                              attr.getHeartbeatTimeoutSeconds(),
+                              task.getWorkflowType(),
+                              domain,
+                              attr.getHeader(),
+                              task.getWorkflowExecution());
+                      if (ldaTaskPoller.apply(activityTask)) {
+                        options
+                            .getMetricsScope()
+                            .counter(MetricsType.ACTIVITY_LOCAL_DISPATCH_SUCCEED_COUNTER)
+                            .inc(1);
+                        decision
+                            .getScheduleActivityTaskDecisionAttributes()
+                            .setRequestLocalDispatch(true);
+                        activityTasks.add(activityTask);
+                      } else {
+                        // all pollers are busy - no room to optimize
+                        options
+                            .getMetricsScope()
+                            .counter(MetricsType.ACTIVITY_LOCAL_DISPATCH_FAILED_COUNTER)
+                            .inc(1);
+                      }
                     }
                   }
                 }