Introduce initialization exception handler in KCL V2

amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/CoordinatorConfig.java

@@ -67,6 +67,8 @@ public class CoordinatorConfig {
      */
     private ShardPrioritization shardPrioritization = new NoOpShardPrioritization();
 
+    private CoordinatorExceptionHandler coordinatorExceptionHandler = new NoOpCoordinatorExceptionHandler();
+
     private CoordinatorFactory coordinatorFactory = new SchedulerCoordinatorFactory();
 
 }

amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/CoordinatorExceptionHandler.java

@@ -0,0 +1,23 @@
+/*
+ *  Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Licensed under the Amazon Software License (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/asl/
+ *
+ *  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 software.amazon.kinesis.coordinator;
+
+/**
+ *
+ */
+public interface CoordinatorExceptionHandler {
+    void propogate(Exception e);
+}

amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/NoOpCoordinatorExceptionHandler.java

@@ -0,0 +1,30 @@
+/*
+ *  Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Licensed under the Amazon Software License (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/asl/
+ *
+ *  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 software.amazon.kinesis.coordinator;
+
+public class NoOpCoordinatorExceptionHandler implements CoordinatorExceptionHandler {
+
+    /**
+     * Empty constructor for NoOp Exception Handler
+     */
+    public NoOpCoordinatorExceptionHandler() {
+
+    }
+
+    @Override
+    public void propogate(Exception e) {
+    }
+}

amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/Scheduler.java

@@ -97,6 +97,7 @@ public class Scheduler implements Runnable {
     private final LeaseCoordinator leaseCoordinator;
     private final ShardSyncTaskManager shardSyncTaskManager;
     private final ShardPrioritization shardPrioritization;
+    private final CoordinatorExceptionHandler coordinatorExceptionHandler;
     private final boolean cleanupLeasesUponShardCompletion;
     private final boolean skipShardSyncAtWorkerInitializationIfLeasesExist;
     private final GracefulShutdownCoordinator gracefulShutdownCoordinator;
@@ -165,6 +166,7 @@ public Scheduler(@NonNull final CheckpointConfig checkpointConfig,
         this.shardSyncTaskManager = this.leaseManagementConfig.leaseManagementFactory()
                 .createShardSyncTaskManager(this.metricsFactory);
         this.shardPrioritization = this.coordinatorConfig.shardPrioritization();
+        this.coordinatorExceptionHandler = this.coordinatorConfig.coordinatorExceptionHandler();
         this.cleanupLeasesUponShardCompletion = this.leaseManagementConfig.cleanupLeasesUponShardCompletion();
         this.skipShardSyncAtWorkerInitializationIfLeasesExist =
                 this.coordinatorConfig.skipShardSyncAtWorkerInitializationIfLeasesExist();
@@ -193,20 +195,26 @@ public void run() {
             return;
         }
 
+        Exception exceptionToThrow = null;
         try {
             initialize();
             log.info("Initialization complete. Starting worker loop.");
         } catch (RuntimeException e) {
             log.error("Unable to initialize after {} attempts. Shutting down.", lifecycleConfig.maxInitializationAttempts(), e);
             shutdown();
+            exceptionToThrow = e;
         }
 
         while (!shouldShutdown()) {
+            exceptionToThrow = null;
             runProcessLoop();
         }
 
         finalShutdown();
         log.info("Worker loop is complete. Exiting from worker.");
+        if (exceptionToThrow != null) {
+            coordinatorExceptionHandler.propogate(exceptionToThrow);
+        }
     }
 
     private void initialize() {