Adding Partition Execution Logging for DimensionalTimeSliceCrawler and 2 minute Partition Execution Delay for Each Partition Batch

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,8 +87,26 @@ public Instant crawl(LeaderPartition leaderPartition, EnhancedSourceCoordinator
 
     @Override
     public void executePartition(DimensionalTimeSliceWorkerProgressState state, Buffer<Record<Event>> buffer, AcknowledgementSet acknowledgementSet) {
+        log.info("Processing partition - DimensionType: {}, TimeRange: {} to {}", 
+                 state.getDimensionType(), state.getStartTime(), state.getEndTime());
+        
         partitionWaitTimeTimer.record(Duration.between(state.getPartitionCreationTime(), Instant.now()));
+        
+        Instant executionStart = Instant.now();
         partitionProcessLatencyTimer.record(() -> client.executePartition(state, buffer, acknowledgementSet));
+        
+        // Enforce 2 minute minimum execution duration
+        Duration actualExecutionTime = Duration.between(executionStart, Instant.now());
+        Duration minimumDuration = Duration.ofMinutes(2); // configurable
+        
+        if (actualExecutionTime.compareTo(minimumDuration) < 0) {
+            Duration remainingDelay = minimumDuration.minus(actualExecutionTime);
+            try {
+                Thread.sleep(remainingDelay.toMillis());
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
     }
 
     private Instant createPartitions(LeaderPartition leaderPartition,

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

@@ -237,4 +237,4 @@ private void updateLeaderProgressState(LeaderPartition leaderPartition,
     void setNoAckTimeout(Duration timeout) {
         this.noAckTimeout = timeout;
     }
-}
+}