feat: Add configurable lease interval for crawler source (opensearch-project#6432)

This change adds support for configurable lease interval in the crawler
source plugin, allowing users to customize the leader scheduler's lease
interval instead of using a hardcoded value.

Changes:
- Added getLeaseInterval() method to CrawlerSourceConfig interface with
  default value of 1 minute
- Modified CrawlerSourcePlugin to use the configurable lease interval
  from the source configuration

Signed-off-by: Alexander Christensen <alchrisk@amazon.com>
Signed-off-by: Simon ELBAZ <elbazsimon9@gmail.com>

Author: chrisale000

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

@@ -45,4 +45,14 @@ default Duration getDurationToGiveUpRetry() {
     default Duration getDurationToDelayRetry() {
         return DEFAULT_MAX_DURATION_TO_DELAY_RETRY;
     }
+
+    /**
+     * Gets the lease interval for the leader scheduler.
+     * Defaults to 1 minute if not overridden.
+     *
+     * @return Duration for lease interval
+     */
+    default Duration getLeaseInterval() {
+        return Duration.ofMinutes(1);
+    }
 }

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

@@ -70,7 +70,8 @@ public void start(Buffer<Record<Event>> buffer) {
         boolean isPartitionCreated = coordinator.createPartition(leaderPartition);
         log.debug("Leader partition creation status: {}", isPartitionCreated);
 
-        Runnable leaderScheduler = new LeaderScheduler(coordinator, crawler);
+        LeaderScheduler leaderScheduler = new LeaderScheduler(coordinator, crawler);
+        leaderScheduler.setLeaseInterval(sourceConfig.getLeaseInterval());
         this.executorService.submit(leaderScheduler);
         //Register worker threaders
         for (int i = 0; i < sourceConfig.getNumberOfWorkers(); i++) {