Clean worker shutdown (#237)

Refactoring of the worker shutdown API and logic.
After the refactoring the worker shutdown is similar to the shutdown of a ThreadPoolExecutor. It is done through shutdown/shutdownNow and then awaitTermination. An activity heartbeat after shutdown/shutdownNow starts failing with ActivityWorkerShutdownException. shutdownNow also sends an interrupt to all currently executing activities..

Bumped the library version to 2.2.0

Author: mfateev

CHANGELOG.md

@@ -1,8 +1,23 @@
 # Changelog
 
-## v1.0.0 (06-04-2018)
+## v2.2.0
+- Added support for workflow and activity server side retries.
+- Clean worker shutdown. Replaced Worker shutdown(Duration) with Worker shutdown, shutdownNow and awaitTermination.
+- Fixed thread exhaustion with a large number of parallel async activities.
+
+## v2.1.3
+- Added RPC headers needed to enable sticky queries. Before this change
+queries did not used cached workflows.
+
+## v2.1.2
+- Requires minimum server release v0.4.0
+- Introduced WorkerFactory and FactoryOptions
+- Added sticky workflow execution, which is caching of a workflow object between decisions. It is enabled by default, 
+to disable use FactoryOptions.disableStickyExecution property.
+- Updated Thrift to expose new types of service exceptions: ServiceBusyError, DomainNotActiveError, LimitExceededError
+- Added metric for corrupted signal as well as metrics related to caching and evictions.
 
-### Features
+## v1.0.0 (06-04-2018)
 - POJO workflow, child workflow, activity execution.
 - Sync and Async workflow execution.
 - Query and Signal workflow execution.
@@ -12,14 +27,4 @@
 - Activity heartbeat throttling.
 - Deterministic retry of failed operation.
 
-## v2.1.2
-- Requires minimum server release v0.4.0
-- Introduced WorkerFactory and FactoryOptions
-- Added sticky workflow execution, which is caching of a workflow object between decisions. It is enabled by default, 
-to disable use FactoryOptions.disableStickyExecution property.
-- Updated Thrift to expose new types of service exceptions: ServiceBusyError, DomainNotActiveError, LimitExceededError
-- Added metric for corrupted signal as well as metrics related to caching and evictions.
 
-## v2.1.3
-- Added RPC headers needed to enable sticky queries. Before this change
-queries did not used cached workflows.

build.gradle

@@ -37,20 +37,20 @@ googleJavaFormat {
 }
 
 group = 'com.uber.cadence'
-version = '2.1.3'
+version = '2.2.0'
 
 description = """Uber Cadence Java Client"""
 
 sourceCompatibility = 1.8
 targetCompatibility = 1.8
 
 dependencies {
-    compile group: 'com.uber.tchannel', name: 'tchannel-core', version: '0.7.6'
+    compile group: 'com.uber.tchannel', name: 'tchannel-core', version: '0.8.5'
     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.2'
+    compile group: 'com.google.code.gson', name: 'gson', version: '2.8.5'
     compile group: 'com.uber.m3', name: 'tally-core', version: '0.2.3'
-    compile group: 'com.google.guava', name: 'guava', version: '25.1-jre'
+    compile group: 'com.google.guava', name: 'guava', version: '27.0.1-jre'
     testCompile group: 'junit', name: 'junit', version: '4.12'
     testCompile group: 'com.googlecode.junit-toolbox', name: 'junit-toolbox', version: '2.4'
     testCompile group: 'ch.qos.logback', name: 'logback-classic', version: '1.2.3'

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

@@ -17,14 +17,14 @@
 
 package com.uber.cadence.activity;
 
+import com.uber.cadence.client.ActivityCompletionException;
 import com.uber.cadence.internal.sync.ActivityInternal;
 import com.uber.cadence.internal.sync.WorkflowInternal;
 import com.uber.cadence.serviceclient.IWorkflowService;
 import com.uber.cadence.workflow.ActivityException;
 import com.uber.cadence.workflow.ActivityTimeoutException;
 import java.lang.reflect.Type;
 import java.util.Optional;
-import java.util.concurrent.CancellationException;
 
 /**
  * An activity is the implementation of a particular task in the business logic.
@@ -232,11 +232,11 @@ public static ActivityTask getTask() {
    *
    * @param details In case of activity timeout can be accessed through {@link
    *     ActivityTimeoutException#getDetails(Class)} method.
-   * @throws CancellationException Indicates that activity cancellation was requested by the
-   *     workflow.Should be rethrown from activity implementation to indicate successful
-   *     cancellation.
+   * @throws ActivityCompletionException Indicates that activity execution is expected to be
+   *     interrupted. The reason for interruption is indicated by a type of subclass of the
+   *     exception.
    */
-  public static void heartbeat(Object details) throws CancellationException {
+  public static <V> void heartbeat(V details) throws ActivityCompletionException {
     ActivityInternal.recordActivityHeartbeat(details);
   }
 

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

@@ -0,0 +1,39 @@
+/*
+ *  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.client;
+
+import com.uber.cadence.activity.ActivityTask;
+import com.uber.cadence.worker.Worker;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Indicates that {@link Worker.Factory#shutdown()} or {@link Worker.Factory#shutdownNow()} was
+ * called. It is OK to ignore the exception to let activity to complete. It assumes that {@link
+ * com.uber.cadence.worker.Worker.Factory#awaitTermination(long, TimeUnit)} is called with a timeout
+ * larger than the activity execution time.
+ */
+public final class ActivityWorkerShutdownException extends ActivityCompletionException {
+
+  public ActivityWorkerShutdownException(ActivityTask task) {
+    super(task);
+  }
+
+  public ActivityWorkerShutdownException() {
+    super();
+  }
+}

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

@@ -19,8 +19,11 @@
 
 import com.uber.cadence.TaskList;
 import com.uber.cadence.TaskListKind;
+import com.uber.cadence.internal.worker.Shutdownable;
 import com.uber.cadence.workflow.WorkflowMethod;
 import java.lang.reflect.Method;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 
 /** Utility functions shared by the implementation code. */
 public final class InternalUtils {
@@ -80,6 +83,41 @@ public static TaskList createNormalTaskList(String taskListName) {
     return tl;
   }
 
+  public static long awaitTermination(Shutdownable s, long timeoutMillis) {
+    if (s == null) {
+      return timeoutMillis;
+    }
+    return awaitTermination(
+        timeoutMillis,
+        () -> {
+          s.awaitTermination(timeoutMillis, TimeUnit.MILLISECONDS);
+        });
+  }
+
+  public static long awaitTermination(ExecutorService s, long timeoutMillis) {
+    if (s == null) {
+      return timeoutMillis;
+    }
+    return awaitTermination(
+        timeoutMillis,
+        () -> {
+          try {
+            s.awaitTermination(timeoutMillis, TimeUnit.MILLISECONDS);
+          } catch (InterruptedException e) {
+          }
+        });
+  }
+
+  public static long awaitTermination(long timeoutMillis, Runnable toTerminate) {
+    long started = System.currentTimeMillis();
+    toTerminate.run();
+    long remainingTimeout = timeoutMillis - (System.currentTimeMillis() - started);
+    if (remainingTimeout < 0) {
+      remainingTimeout = 0;
+    }
+    return remainingTimeout;
+  }
+
   /** Prohibit instantiation */
   private InternalUtils() {}
 }

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

@@ -27,6 +27,7 @@
 import com.uber.cadence.client.ActivityCompletionException;
 import com.uber.cadence.client.ActivityCompletionFailureException;
 import com.uber.cadence.client.ActivityNotExistsException;
+import com.uber.cadence.client.ActivityWorkerShutdownException;
 import com.uber.cadence.converter.DataConverter;
 import com.uber.cadence.serviceclient.IWorkflowService;
 import java.lang.reflect.Type;
@@ -85,6 +86,9 @@ class ActivityExecutionContextImpl implements ActivityExecutionContext {
   /** @see ActivityExecutionContext#recordActivityHeartbeat(Object) */
   @Override
   public <V> void recordActivityHeartbeat(V details) throws ActivityCompletionException {
+    if (heartbeatExecutor.isShutdown()) {
+      throw new ActivityWorkerShutdownException(task);
+    }
     lock.lock();
     try {
       // always set lastDetail. Successful heartbeat will clear it.

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

@@ -30,7 +30,7 @@ static ActivityExecutionContext getContext() {
     return CurrentActivityExecutionContext.get();
   }
 
-  public static void recordActivityHeartbeat(Object details) {
+  public static <V> void recordActivityHeartbeat(V details) {
     getContext().recordActivityHeartbeat(details);
   }
 

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

@@ -17,15 +17,17 @@
 
 package com.uber.cadence.internal.sync;
 
+import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.internal.worker.ActivityWorker;
+import com.uber.cadence.internal.worker.Lifecycle;
 import com.uber.cadence.internal.worker.SingleWorkerOptions;
 import com.uber.cadence.serviceclient.IWorkflowService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 /** Activity worker that supports POJO activity implementations. */
-public class SyncActivityWorker {
+public class SyncActivityWorker implements Lifecycle {
 
   private final ActivityWorker worker;
   private final POJOActivityTaskHandler taskHandler;
@@ -41,34 +43,43 @@ public void setActivitiesImplementation(Object... activitiesImplementation) {
     taskHandler.setActivitiesImplementation(activitiesImplementation);
   }
 
+  @Override
   public void start() {
     worker.start();
   }
 
+  @Override
+  public boolean isStarted() {
+    return worker.isStarted();
+  }
+
+  @Override
+  public boolean isShutdown() {
+    return worker.isShutdown();
+  }
+
+  @Override
+  public boolean isTerminated() {
+    return worker.isTerminated() && heartbeatExecutor.isTerminated();
+  }
+
+  @Override
   public void shutdown() {
     worker.shutdown();
     heartbeatExecutor.shutdown();
   }
 
+  @Override
   public void shutdownNow() {
     worker.shutdownNow();
     heartbeatExecutor.shutdownNow();
   }
 
-  public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
-    return worker.awaitTermination(timeout, unit)
-        && heartbeatExecutor.awaitTermination(timeout, unit);
-  }
-
-  public boolean shutdownAndAwaitTermination(long timeout, TimeUnit unit)
-      throws InterruptedException {
-    heartbeatExecutor.shutdownNow();
-    return worker.shutdownAndAwaitTermination(timeout, unit)
-        && heartbeatExecutor.awaitTermination(timeout, unit);
-  }
-
-  public boolean isRunning() {
-    return worker.isRunning();
+  @Override
+  public void awaitTermination(long timeout, TimeUnit unit) {
+    long timeoutMillis = unit.toMillis(timeout);
+    timeoutMillis = InternalUtils.awaitTermination(worker, timeoutMillis);
+    InternalUtils.awaitTermination(heartbeatExecutor, timeoutMillis);
   }
 
   public void suspendPolling() {

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

@@ -25,6 +25,7 @@
 import com.uber.cadence.internal.replay.ReplayDecisionTaskHandler;
 import com.uber.cadence.internal.worker.DecisionTaskHandler;
 import com.uber.cadence.internal.worker.SingleWorkerOptions;
+import com.uber.cadence.internal.worker.SuspendableWorker;
 import com.uber.cadence.internal.worker.WorkflowWorker;
 import com.uber.cadence.serviceclient.IWorkflowService;
 import com.uber.cadence.worker.WorkflowImplementationOptions;
@@ -39,7 +40,8 @@
 import java.util.function.Function;
 
 /** Workflow worker that supports POJO workflow implementations. */
-public class SyncWorkflowWorker implements Consumer<PollForDecisionTaskResponse> {
+public class SyncWorkflowWorker
+    implements SuspendableWorker, Consumer<PollForDecisionTaskResponse> {
 
   private final WorkflowWorker worker;
   private final POJOWorkflowImplementationFactory factory;
@@ -84,39 +86,56 @@ public <R> void addWorkflowImplementationFactory(Class<R> clazz, Func<R> factory
     this.factory.addWorkflowImplementationFactory(clazz, factory);
   }
 
+  @Override
   public void start() {
     worker.start();
   }
 
-  public void shutdown() {
-    worker.shutdown();
+  @Override
+  public boolean isStarted() {
+    return worker.isStarted();
   }
 
-  public void shutdownNow() {
-    worker.shutdownNow();
+  @Override
+  public boolean isShutdown() {
+    return worker.isShutdown();
   }
 
-  public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
-    return worker.awaitTermination(timeout, unit);
+  @Override
+  public boolean isTerminated() {
+    return worker.isTerminated();
+  }
+
+  @Override
+  public void shutdown() {
+    worker.shutdown();
   }
 
-  public boolean shutdownAndAwaitTermination(long timeout, TimeUnit unit)
-      throws InterruptedException {
-    return worker.shutdownAndAwaitTermination(timeout, unit);
+  @Override
+  public void shutdownNow() {
+    worker.shutdownNow();
   }
 
-  public boolean isRunning() {
-    return worker.isRunning();
+  @Override
+  public void awaitTermination(long timeout, TimeUnit unit) {
+    worker.awaitTermination(timeout, unit);
   }
 
+  @Override
   public void suspendPolling() {
     worker.suspendPolling();
   }
 
+  @Override
   public void resumePolling() {
     worker.resumePolling();
   }
 
+  @Override
+  public boolean isSuspended() {
+    return worker.isSuspended();
+  }
+
   public <R> R queryWorkflowExecution(
       WorkflowExecution execution,
       String queryType,

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

@@ -791,5 +791,10 @@ public PollForActivityTaskResponse PollForActivityTask(PollForActivityTaskReques
         throws BadRequestError, InternalServiceError, ServiceBusyError, TException {
       return impl.PollForActivityTask(pollRequest);
     }
+
+    @Override
+    public void close() {
+      impl.close();
+    }
   }
 }