KCLRecordProcessor handles multiple shutdown calls gracefully

In some occasions KCL library calls shutdown multiple times for the
same shards KCLRecordProcessor. We have to make sure it doesn't fail
because closed dependencies.

source/src/main/java/com/trustpilot/connector/dynamodb/kcl/KclRecordProcessor.java

@@ -208,18 +208,19 @@ public void shutdown(ShutdownInput shutdownInput) {
      * Called on ShutdownReason.TERMINATE
      */
     private void onTerminate(ShutdownInput shutdownInput) throws InvalidStateException, ShutdownException, InterruptedException {
-        if (!lastProcessedSeqNo.isEmpty()) {
-            ShardInfo processorRegister = shardRegister.get(this.shardId);
-
-            while (!processorRegister.getLastCommittedRecordSeqNo().equals(this.lastProcessedSeqNo)) {
-                LOGGER.info(
-                        "Shard ended. Waiting for all data table: {} from shard: {} to be committed. " +
-                                "lastCommittedRecordSeqNo: {} lastProcessedSeqNo: {}",
-                        tableName,
-                        shardId,
-                        processorRegister.getLastCommittedRecordSeqNo(),
-                        this.lastProcessedSeqNo);
-                Thread.sleep(1000);
+        if (lastProcessedSeqNo != null && !lastProcessedSeqNo.isEmpty()) {
+            ShardInfo processorRegister = shardRegister.getOrDefault(this.shardId, null);
+            if (processorRegister != null) {
+                while (!processorRegister.getLastCommittedRecordSeqNo().equals(this.lastProcessedSeqNo)) {
+                    LOGGER.info(
+                            "Shard ended. Waiting for all data table: {} from shard: {} to be committed. " +
+                                    "lastCommittedRecordSeqNo: {} lastProcessedSeqNo: {}",
+                            tableName,
+                            shardId,
+                            processorRegister.getLastCommittedRecordSeqNo(),
+                            this.lastProcessedSeqNo);
+                    Thread.sleep(500);
+                }
             }
         }
 
@@ -231,7 +232,10 @@ private void onTerminate(ShutdownInput shutdownInput) throws InvalidStateExcepti
                 "Shard ended. All data committed. Checkpoint and proceed to next one. Table: {} ShardID: {}",
                 tableName,
                 shardId);
-        shutdownInput.getCheckpointer().checkpoint();
+        IRecordProcessorCheckpointer checkpointer = shutdownInput.getCheckpointer();
+        if (checkpointer != null) {
+            checkpointer.checkpoint();
+        }
     }
 
     /**

source/src/test/java/com/trustpilot/connector/dynamodb/kcl/KclRecordProcessorTests.java

@@ -264,6 +264,27 @@ void shutdownWaitsForLastRecordToBeCommittedOnShardEndBeforeCheckpoint() throws
         verify(checkpointer, only()).checkpoint();
     }
 
+        @Test
+    void shutdownSucceddsIfCalledMultipleTimes() throws InvalidStateException, ShutdownException {
+        // Arrange
+        IRecordProcessorCheckpointer checkpointer = Mockito.mock(IRecordProcessorCheckpointer.class);
+        ProcessRecordsInput processRecordsInput = getProcessRecordsInput("SQ1").withCheckpointer(checkpointer);
+        shardRegister.get(shardId).setLastCommittedRecordSeqNo("SQ1");
+
+        // Act
+        processor.processRecords(processRecordsInput);
+        ShutdownInput shutdownInput = new ShutdownInput()
+                .withShutdownReason(ShutdownReason.TERMINATE)
+                .withCheckpointer(checkpointer);
+
+        processor.shutdown(shutdownInput);
+        processor.shutdown(shutdownInput);
+
+        // Assert
+        verify(checkpointer, times(2)).checkpoint();
+        assertFalse(shardRegister.containsKey(shardId));
+    }
+
     @Test
     void shutdownCheckpointsIfNoDataWasReceivedFromThisShard() throws InvalidStateException, ShutdownException {
         // Arrange