Remove disable activity and workflow worker options (#540)

- Workers will now always start, and poller will start only if types are registered.
- Simplified worker start and shutdown logic.
- Fixed a bug in worker.isSuspended()
- Move common method implementations to base

Author: meiliang86

build.gradle

@@ -68,6 +68,7 @@ license {
     header rootProject.file('license-header.txt')
     skipExistingHeaders true
     exclude 'com/uber/cadence/*.java' // generated code
+    excludes(["**/*.json"])
 }
 
 task initDlsSubmodule(type: Exec) {

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

@@ -118,7 +118,11 @@ public void setLocalActivitiesImplementation(Object... activitiesImplementation)
   @Override
   public void start() {
     workflowWorker.start();
-    laWorker.start();
+    // workflowWorker doesn't start if no types are registered with it. In that case we don't need
+    // to start LocalActivity Worker.
+    if (workflowWorker.isStarted()) {
+      laWorker.start();
+    }
   }
 
   @Override

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

@@ -47,11 +47,10 @@
 import org.apache.thrift.TException;
 import org.slf4j.MDC;
 
-public final class ActivityWorker implements SuspendableWorker {
+public final class ActivityWorker extends SuspendableWorkerBase {
 
   private static final String POLL_THREAD_NAME_PREFIX = "Activity Poller taskList=";
 
-  private SuspendableWorker poller = new NoopSuspendableWorker();
   private final ActivityTaskHandler handler;
   private final IWorkflowService service;
   private final String domain;
@@ -83,63 +82,19 @@ public ActivityWorker(
   @Override
   public void start() {
     if (handler.isAnyTypeSupported()) {
-      poller =
+      SuspendableWorker poller =
           new Poller<>(
               options.getIdentity(),
               new ActivityPollTask(service, domain, taskList, options),
               new PollTaskExecutor<>(domain, taskList, options, new TaskHandlerImpl(handler)),
               options.getPollerOptions(),
               options.getMetricsScope());
       poller.start();
+      setPoller(poller);
       options.getMetricsScope().counter(MetricsType.WORKER_START_COUNTER).inc(1);
     }
   }
 
-  @Override
-  public boolean isStarted() {
-    return poller.isStarted();
-  }
-
-  @Override
-  public boolean isShutdown() {
-    return poller.isShutdown();
-  }
-
-  @Override
-  public boolean isTerminated() {
-    return poller.isTerminated();
-  }
-
-  @Override
-  public void shutdown() {
-    poller.shutdown();
-  }
-
-  @Override
-  public void shutdownNow() {
-    poller.shutdownNow();
-  }
-
-  @Override
-  public void awaitTermination(long timeout, TimeUnit unit) {
-    poller.awaitTermination(timeout, unit);
-  }
-
-  @Override
-  public void suspendPolling() {
-    poller.suspendPolling();
-  }
-
-  @Override
-  public void resumePolling() {
-    poller.resumePolling();
-  }
-
-  @Override
-  public boolean isSuspended() {
-    return poller.isSuspended();
-  }
-
   private class TaskHandlerImpl
       implements PollTaskExecutor.TaskHandler<PollForActivityTaskResponse> {
 

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

@@ -34,16 +34,14 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
 import java.util.function.LongSupplier;
 
-public final class LocalActivityWorker implements SuspendableWorker {
+public final class LocalActivityWorker extends SuspendableWorkerBase {
 
   private static final String POLL_THREAD_NAME_PREFIX = "Local Activity Poller taskList=";
 
-  private SuspendableWorker poller = new NoopSuspendableWorker();
   private final ActivityTaskHandler handler;
   private final String domain;
   private final String taskList;
@@ -71,63 +69,19 @@ public LocalActivityWorker(
   @Override
   public void start() {
     if (handler.isAnyTypeSupported()) {
-      poller =
+      SuspendableWorker poller =
           new Poller<>(
               options.getIdentity(),
               laPollTask,
               new PollTaskExecutor<>(domain, taskList, options, new TaskHandlerImpl(handler)),
               options.getPollerOptions(),
               options.getMetricsScope());
       poller.start();
+      setPoller(poller);
       options.getMetricsScope().counter(MetricsType.WORKER_START_COUNTER).inc(1);
     }
   }
 
-  @Override
-  public boolean isStarted() {
-    return poller.isStarted();
-  }
-
-  @Override
-  public boolean isShutdown() {
-    return poller.isShutdown();
-  }
-
-  @Override
-  public boolean isTerminated() {
-    return poller.isTerminated();
-  }
-
-  @Override
-  public void shutdown() {
-    poller.shutdown();
-  }
-
-  @Override
-  public void shutdownNow() {
-    poller.shutdownNow();
-  }
-
-  @Override
-  public void awaitTermination(long timeout, TimeUnit unit) {
-    poller.awaitTermination(timeout, unit);
-  }
-
-  @Override
-  public void suspendPolling() {
-    poller.suspendPolling();
-  }
-
-  @Override
-  public void resumePolling() {
-    poller.resumePolling();
-  }
-
-  @Override
-  public boolean isSuspended() {
-    return poller.isSuspended();
-  }
-
   public static class Task {
     private final ExecuteLocalActivityParameters params;
     private final Consumer<HistoryEvent> eventConsumer;

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

@@ -52,27 +52,21 @@ public void shutdownNow() {
   public void awaitTermination(long timeout, TimeUnit unit) {}
 
   @Override
-  public void start() {
-    throw new IllegalStateException("Non startable");
-  }
+  public void start() {}
 
   @Override
   public boolean isStarted() {
     return false;
   }
 
   @Override
-  public void suspendPolling() {
-    throw new IllegalStateException("Non suspendable");
-  }
+  public void suspendPolling() {}
 
   @Override
-  public void resumePolling() {
-    throw new IllegalStateException("Non resumable");
-  }
+  public void resumePolling() {}
 
   @Override
   public boolean isSuspended() {
-    return false;
+    return true;
   }
 }

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

@@ -0,0 +1,76 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package com.uber.cadence.internal.worker;
+
+import java.util.concurrent.TimeUnit;
+
+public abstract class SuspendableWorkerBase implements SuspendableWorker {
+  private SuspendableWorker poller = new NoopSuspendableWorker();
+
+  final void setPoller(SuspendableWorker poller) {
+    this.poller = poller;
+  }
+
+  @Override
+  public boolean isStarted() {
+    return poller.isStarted();
+  }
+
+  @Override
+  public boolean isShutdown() {
+    return poller.isShutdown();
+  }
+
+  @Override
+  public boolean isTerminated() {
+    return poller.isTerminated();
+  }
+
+  @Override
+  public void shutdown() {
+    poller.shutdown();
+  }
+
+  @Override
+  public void shutdownNow() {
+    poller.shutdownNow();
+  }
+
+  @Override
+  public void awaitTermination(long timeout, TimeUnit unit) {
+    if (!poller.isStarted()) {
+      return;
+    }
+    poller.awaitTermination(timeout, unit);
+  }
+
+  @Override
+  public void suspendPolling() {
+    poller.suspendPolling();
+  }
+
+  @Override
+  public void resumePolling() {
+    poller.resumePolling();
+  }
+
+  @Override
+  public boolean isSuspended() {
+    return poller.isSuspended();
+  }
+}

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

@@ -46,18 +46,16 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.function.Consumer;
 import org.apache.thrift.TException;
 import org.slf4j.MDC;
 
-public final class WorkflowWorker
-    implements SuspendableWorker, Consumer<PollForDecisionTaskResponse> {
+public final class WorkflowWorker extends SuspendableWorkerBase
+    implements Consumer<PollForDecisionTaskResponse> {
 
   private static final String POLL_THREAD_NAME_PREFIX = "Workflow Poller taskList=";
 
-  private SuspendableWorker poller = new NoopSuspendableWorker();
   private PollTaskExecutor<PollForDecisionTaskResponse> pollTaskExecutor;
   private final DecisionTaskHandler handler;
   private final IWorkflowService service;
@@ -96,7 +94,7 @@ public void start() {
     if (handler.isAnyTypeSupported()) {
       pollTaskExecutor =
           new PollTaskExecutor<>(domain, taskList, options, new TaskHandlerImpl(handler));
-      poller =
+      SuspendableWorker poller =
           new Poller<>(
               options.getIdentity(),
               new WorkflowPollTask(
@@ -105,25 +103,11 @@ public void start() {
               options.getPollerOptions(),
               options.getMetricsScope());
       poller.start();
+      setPoller(poller);
       options.getMetricsScope().counter(MetricsType.WORKER_START_COUNTER).inc(1);
     }
   }
 
-  @Override
-  public boolean isStarted() {
-    return poller.isStarted();
-  }
-
-  @Override
-  public boolean isShutdown() {
-    return poller.isShutdown();
-  }
-
-  @Override
-  public boolean isTerminated() {
-    return poller.isTerminated();
-  }
-
   public byte[] queryWorkflowExecution(WorkflowExecution exec, String queryType, byte[] args)
       throws Exception {
     GetWorkflowExecutionHistoryResponse historyResponse =
@@ -187,40 +171,6 @@ private byte[] queryWorkflowExecution(
     throw new RuntimeException("Query returned wrong response: " + result);
   }
 
-  @Override
-  public void shutdown() {
-    poller.shutdown();
-  }
-
-  @Override
-  public void shutdownNow() {
-    poller.shutdownNow();
-  }
-
-  @Override
-  public void awaitTermination(long timeout, TimeUnit unit) {
-    if (!poller.isStarted()) {
-      return;
-    }
-
-    poller.awaitTermination(timeout, unit);
-  }
-
-  @Override
-  public void suspendPolling() {
-    poller.suspendPolling();
-  }
-
-  @Override
-  public void resumePolling() {
-    poller.resumePolling();
-  }
-
-  @Override
-  public boolean isSuspended() {
-    return poller.isSuspended();
-  }
-
   @Override
   public void accept(PollForDecisionTaskResponse pollForDecisionTaskResponse) {
     pollTaskExecutor.process(pollForDecisionTaskResponse);