Do not set ending sequence number for child shard partitions discovered after initial scan for shards

Signed-off-by: Taylor Gray <tylgry@amazon.com>

Author: graytaylor0

data-prepper-plugins/dynamodb-source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/coordination/state/LeaderProgressState.java

@@ -12,6 +12,9 @@ public class LeaderProgressState {
     @JsonProperty("streamArns")
     private List<String> streamArns;
 
+    @JsonProperty("completed_first_discovery")
+    private boolean completedFirstDiscovery = false;
+
     public boolean isInitialized() {
         return initialized;
     }
@@ -27,4 +30,13 @@ public List<String> getStreamArns() {
     public void setStreamArns(List<String> streamArns) {
         this.streamArns = streamArns;
     }
+
+    public boolean isCompletedFirstDiscovery() {
+        return completedFirstDiscovery;
+    }
+
+    public void setCompletedFirstDiscovery(boolean completedFirstDiscovery) {
+        this.completedFirstDiscovery = completedFirstDiscovery;
+    }
+
 }

data-prepper-plugins/dynamodb-source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/leader/LeaderScheduler.java

@@ -57,6 +57,7 @@ public class LeaderScheduler implements Runnable {
     private LeaderPartition leaderPartition;
 
     private List<String> streamArns;
+    private Map<String, Boolean> isExportDoneForStreamArns;
 
     public LeaderScheduler(EnhancedSourceCoordinator coordinator, DynamoDbClient dynamoDbClient, ShardManager shardManager, List<TableConfig> tableConfigs) {
         this(coordinator, dynamoDbClient, shardManager, tableConfigs, DEFAULT_LEASE_INTERVAL);
@@ -116,6 +117,7 @@ public void run() {
 
                         // Step 3: Find and create children partitions.
                         compareAndCreateChildrenPartitions(sourcePartitions);
+                        leaderPartition.getProgressState().get().setCompletedFirstDiscovery(true);
                     }
 
                 }
@@ -337,7 +339,11 @@ private void createChildStreamPartition(StreamPartition streamPartition, String
         StreamProgressState parentStreamProgressState = streamPartition.getProgressState().get();
         StreamProgressState streamProgressState = new StreamProgressState();
         streamProgressState.setStartTime(parentStreamProgressState.getStartTime());
-        streamProgressState.setEndingSequenceNumber(shardManager.getEndingSequenceNumber(childShardId));
+        if (shouldSetEndingSequenceNumber()) {
+            streamProgressState.setEndingSequenceNumber(shardManager.getEndingSequenceNumber(childShardId));
+        } else {
+            streamProgressState.setEndingSequenceNumber(null);
+        }
         streamProgressState.setWaitForExport(parentStreamProgressState.shouldWaitForExport());
         StreamPartition partition = new StreamPartition(streamPartition.getStreamArn(), childShardId, Optional.of(streamProgressState));
         coordinator.createPartition(partition);
@@ -363,4 +369,7 @@ private void createExportPartition(String tableArn, Instant exportTime, String b
         coordinator.createPartition(exportPartition);
     }
 
+    private boolean shouldSetEndingSequenceNumber() {
+        return leaderPartition.getProgressState().get().isCompletedFirstDiscovery();
+    }
 }