ShardId is not passed to ShardConsumer, resulting in logs saying shar…

…d is null on shutdown (opensearch-project#3683) Signed-off-by: Taylor Gray <[email protected]>
kkondaka · Nov 16, 2023 · 1c28285 · 1c28285
1 parent 92224c2
commit 1c28285
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 7 deletions.
diff --git a/...rc/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/stream/ShardConsumer.java b/...rc/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/stream/ShardConsumer.java
@@ -233,9 +233,7 @@ public void run() {
             }
 
             if (System.currentTimeMillis() - lastCheckpointTime > DEFAULT_CHECKPOINT_INTERVAL_MILLS) {
-                if (shardId != null) {
-                    LOG.info("{} records written to buffer for shard {}", recordsWrittenToBuffer, shardId);
-                }
+                LOG.debug("{} records written to buffer for shard {}", recordsWrittenToBuffer, shardId);
                 checkpointer.checkpoint(sequenceNumber);
                 lastCheckpointTime = System.currentTimeMillis();
             }
@@ -289,10 +287,6 @@ public void run() {
             acknowledgementSet.complete();
         }
 
-        if (shardId != null) {
-            LOG.info("Completed writing shard {} to buffer after reaching the end of the shard", shardId);
-        }
-
         if (waitForExport) {
             waitForExport();
         }

diff --git a/.../java/org/opensearch/dataprepper/plugins/source/dynamodb/stream/ShardConsumerFactory.java b/.../java/org/opensearch/dataprepper/plugins/source/dynamodb/stream/ShardConsumerFactory.java
@@ -96,6 +96,7 @@ public Runnable createConsumer(final StreamPartition streamPartition,
                 .tableInfo(tableInfo)
                 .checkpointer(checkpointer)
                 .shardIterator(shardIterator)
+                .shardId(streamPartition.getShardId())
                 .lastShardIterator(lastShardIterator)
                 .startTime(startTime)
                 .waitForExport(waitForExport)