Address review comments

Signed-off-by: Souvik Bose <[email protected]>
opensearch-project · Oct 8, 2024 · d6e5036 · d6e5036
1 parent 837e9d3
commit d6e5036
Showing 1 changed file with 12 additions and 13 deletions.
diff --git a/...ource/src/main/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisService.java b/...ource/src/main/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisService.java
@@ -116,20 +116,19 @@ public void start(final Buffer<Record<Event>> buffer) {
     public void shutDown() {
         LOG.info("Stop request received for Kinesis Source");
 
-        if (scheduler == null) {
-            LOG.info("Scheduler not initialized!!");
-            return;
-        }
-
-        Future<Boolean> gracefulShutdownFuture = scheduler.startGracefulShutdown();
-        LOG.info("Waiting up to {} seconds for shutdown to complete.", GRACEFUL_SHUTDOWN_WAIT_INTERVAL_SECONDS);
-        try {
-            gracefulShutdownFuture.get(GRACEFUL_SHUTDOWN_WAIT_INTERVAL_SECONDS, TimeUnit.SECONDS);
-        } catch (InterruptedException | ExecutionException | TimeoutException ex) {
-            LOG.error("Exception while executing kinesis consumer graceful shutdown, doing force shutdown", ex);
-            scheduler.shutdown();
+        if (scheduler != null) {
+            Future<Boolean> gracefulShutdownFuture = scheduler.startGracefulShutdown();
+            LOG.info("Waiting up to {} seconds for shutdown to complete.", GRACEFUL_SHUTDOWN_WAIT_INTERVAL_SECONDS);
+            try {
+                gracefulShutdownFuture.get(GRACEFUL_SHUTDOWN_WAIT_INTERVAL_SECONDS, TimeUnit.SECONDS);
+            } catch (InterruptedException | ExecutionException | TimeoutException ex) {
+                LOG.error("Exception while executing kinesis consumer graceful shutdown, doing force shutdown", ex);
+                scheduler.shutdown();
+            }
+            LOG.info("Completed, shutting down now.");
+        } else {
+            LOG.info("The Kinesis Scheduler was not initialized.");
         }
-        LOG.info("Completed, shutting down now.");
     }
 
     public Scheduler getScheduler(final Buffer<Record<Event>> buffer) {