Adding Partition Execution Logging

Signed-off-by: Alexander Christensen <alchrisk@amazon.com>

Author: chrisale000

data-prepper-plugins/saas-source-plugins/source-crawler/src/main/java/org/opensearch/dataprepper/plugins/source/source_crawler/base/DimensionalTimeSliceCrawler.java

@@ -87,6 +87,8 @@ public Instant crawl(LeaderPartition leaderPartition, EnhancedSourceCoordinator
 
     @Override
     public void executePartition(DimensionalTimeSliceWorkerProgressState state, Buffer<Record<Event>> buffer, AcknowledgementSet acknowledgementSet) {
+        log.debug("Processing partition - DimensionType: {}, TimeRange: {} to {}", 
+                 state.getDimensionType(), state.getStartTime(), state.getEndTime());
         partitionWaitTimeTimer.record(Duration.between(state.getPartitionCreationTime(), Instant.now()));
         partitionProcessLatencyTimer.record(() -> client.executePartition(state, buffer, acknowledgementSet));
     }

data-prepper-plugins/saas-source-plugins/source-crawler/src/main/java/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

@@ -123,6 +123,9 @@ public Instant crawl(LeaderPartition leaderPartition,
 
     @Override
     public void executePartition(PaginationCrawlerWorkerProgressState state, Buffer buffer, AcknowledgementSet acknowledgementSet) {
+        String firstToken = !state.getItemIds().isEmpty() ? state.getItemIds().get(0) : "";
+        String lastToken = !state.getItemIds().isEmpty() ? state.getItemIds().get(state.getItemIds().size()-1) : "";
+        log.debug("Processing partition - FirstToken: {}, LastToken: {}", firstToken, lastToken);
         client.executePartition(state, buffer, acknowledgementSet);
     }
 
@@ -140,6 +143,10 @@ private List<ItemInfo> collectBatch(Iterator<ItemInfo> iterator) {
     private void processBatch(List<ItemInfo> batch,
                               LeaderPartition leaderPartition,
                               EnhancedSourceCoordinator coordinator) {
+        String firstToken = !batch.isEmpty() ? batch.get(0).getItemId() : "(No First Token in this batch)";
+        String lastToken = !batch.isEmpty() ? batch.get(batch.size()-1).getItemId() : "(No Last Token in this batch)";
+        log.debug("Processing batch - FirstToken: {}, LastToken: {}", firstToken, lastToken);
+
         if (acknowledgementsEnabled) {
             AtomicBoolean ackReceived = new AtomicBoolean(false);
             long createTimestamp = System.currentTimeMillis();
@@ -237,4 +244,4 @@ private void updateLeaderProgressState(LeaderPartition leaderPartition,
     void setNoAckTimeout(Duration timeout) {
         this.noAckTimeout = timeout;
     }
-}
+}