Fix bug that would cause ddb stream processing to start before export completed (opensearch-project#6892)

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

Author: graytaylor0

data-prepper-plugins/dynamodb-source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/export/ExportScheduler.java

@@ -197,7 +197,7 @@ private void createDataFilePartitions(final String exportArn,
 
         // Currently, we need to maintain a global state to track the overall progress.
         // So that we can easily tell if all the export files are loaded
-        LoadStatus loadStatus = new LoadStatus(totalFiles.get(), 0, totalRecords.get(), 0);
+        LoadStatus loadStatus = new LoadStatus(dataFileInfo.size(), 0, totalRecords.get(), 0);
         enhancedSourceCoordinator.createPartition(new GlobalState(exportArn, Optional.of(loadStatus.toMap())));
 
         dataFileInfo.forEach((key, size) -> {

data-prepper-plugins/dynamodb-source/src/test/java/org/opensearch/dataprepper/plugins/source/dynamodb/export/ExportSchedulerTest.java

@@ -20,6 +20,7 @@
 import org.opensearch.dataprepper.plugins.source.dynamodb.coordination.partition.GlobalState;
 import org.opensearch.dataprepper.plugins.source.dynamodb.coordination.state.ExportProgressState;
 import org.opensearch.dataprepper.plugins.source.dynamodb.model.ExportSummary;
+import org.opensearch.dataprepper.plugins.source.dynamodb.model.LoadStatus;
 import org.opensearch.dataprepper.plugins.source.dynamodb.utils.DynamoDBSourceAggregateMetrics;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
 import software.amazon.awssdk.services.dynamodb.model.DescribeExportRequest;
@@ -185,6 +186,12 @@ public void test_run_exportJob_correctly() throws InterruptedException {
         assertThat(createdPartitions.get(1), instanceOf(DataFilePartition.class));
         assertThat(createdPartitions.get(2), instanceOf(DataFilePartition.class));
 
+        // Verify the GlobalState LoadStatus has accurate totalFiles count
+        GlobalState globalState = (GlobalState) createdPartitions.get(0);
+        LoadStatus loadStatus = LoadStatus.fromMap(globalState.getProgressState().get());
+        assertThat(loadStatus.getTotalFiles(), equalTo(2));
+        assertThat(loadStatus.getLoadedFiles(), equalTo(0));
+
         // Complete the export partition
         verify(coordinator).completePartition(any(EnhancedSourcePartition.class));
         verify(exportJobSuccess).increment();