Fix shutdown sequence

Fix a few issues in shutdown sequence and make sure scheduler shutdown without invoking run works Inorder for the main thread to shutdown, the scheduler's final shutdown still needs to be invoked, which only happens via the run method.
awslabs · Nov 8, 2024 · 6acb51d · 6acb51d
1 parent 037dd7f
commit 6acb51d
Show file tree

Hide file tree

Showing 4 changed files with 13 additions and 2 deletions.
diff --git a/amazon-kinesis-client/pom.xml b/amazon-kinesis-client/pom.xml
@@ -120,6 +120,11 @@
       <artifactId>http-client-spi</artifactId>
       <version>${awssdk.version}</version>
     </dependency>
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>dynamodb-enhanced</artifactId>
+      <version>${awssdk.version}</version>
+    </dependency>
 
     <dependency>
       <groupId>software.amazon.glue</groupId>

diff --git a/.../main/java/software/amazon/kinesis/coordinator/DynamicMigrationComponentsInitializer.java b/.../main/java/software/amazon/kinesis/coordinator/DynamicMigrationComponentsInitializer.java
@@ -223,6 +223,9 @@ void shutdown() {
         workerMetricsThreadPool.shutdown();
         try {
             if (!lamThreadPool.awaitTermination(SCHEDULER_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+                log.info(
+                        "LamThreadPool did not shutdown in {}s, forcefully shutting down",
+                        SCHEDULER_SHUTDOWN_TIMEOUT_SECONDS);
                 lamThreadPool.shutdownNow();
             }
         } catch (final InterruptedException e) {
@@ -232,6 +235,9 @@ void shutdown() {
 
         try {
             if (!workerMetricsThreadPool.awaitTermination(SCHEDULER_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+                log.info(
+                        "WorkerMetricsThreadPool did not shutdown in {}s, forcefully shutting down",
+                        SCHEDULER_SHUTDOWN_TIMEOUT_SECONDS);
                 workerMetricsThreadPool.shutdownNow();
             }
         } catch (final InterruptedException e) {

diff --git a/...tware/amazon/kinesis/coordinator/migration/MigrationClientVersion3xWithRollbackState.java b/...tware/amazon/kinesis/coordinator/migration/MigrationClientVersion3xWithRollbackState.java
@@ -87,7 +87,7 @@ public synchronized void enter(final ClientVersion fromClientVersion) throws Dep
     }
 
     @Override
-    public void leave() {
+    public synchronized void leave() {
         if (entered && !left) {
             log.info("Leaving {}", this);
             cancelRollbackMonitor();

diff --git a/...rc/main/java/software/amazon/kinesis/coordinator/migration/MigrationStateMachineImpl.java b/...rc/main/java/software/amazon/kinesis/coordinator/migration/MigrationStateMachineImpl.java
@@ -126,7 +126,7 @@ public void shutdown() {
         if (!stateMachineThreadPool.isShutdown()) {
             stateMachineThreadPool.shutdown();
             try {
-                if (stateMachineThreadPool.awaitTermination(THREAD_POOL_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+                if (!stateMachineThreadPool.awaitTermination(THREAD_POOL_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
                     log.info(
                             "StateMachineThreadPool did not shutdown within {} seconds, forcefully shutting down",
                             THREAD_POOL_SHUTDOWN_TIMEOUT_SECONDS);