Increase acknowledgment set timeout for opensearch source (opensearch-project#6291)

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

Author: graytaylor0

data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/WorkerCommonUtils.java

@@ -28,7 +28,11 @@ public class WorkerCommonUtils {
     static final Duration BACKOFF_ON_EXCEPTION = Duration.ofSeconds(60);
 
     static final long DEFAULT_CHECKPOINT_INTERVAL_MILLS = 5 * 60_000;
-    static final Duration ACKNOWLEDGEMENT_SET_TIMEOUT = Duration.ofMinutes(20);
+
+    // Set acknowledgment timeout to very high value to handle large indexes.
+    // In case of failure the retries will be handled by source coordination
+    static final Duration ACKNOWLEDGEMENT_SET_TIMEOUT = Duration.ofHours(1000);
+    static final Duration OWNERSHIP_TIMEOUT = Duration.ofMinutes(30);
     static final Duration STARTING_BACKOFF = Duration.ofMillis(500);
     static final Duration MAX_BACKOFF = Duration.ofSeconds(60);
     static final int BACKOFF_RATE = 2;
@@ -64,7 +68,7 @@ static void completeIndexPartition(final OpenSearchSourceConfiguration openSearc
                                               final SourcePartition<OpenSearchIndexProgressState> indexPartition,
                                               final SourceCoordinator<OpenSearchIndexProgressState> sourceCoordinator) {
         if (openSearchSourceConfiguration.isAcknowledgmentsEnabled()) {
-            sourceCoordinator.updatePartitionForAcknowledgmentWait(indexPartition.getPartitionKey(), ACKNOWLEDGEMENT_SET_TIMEOUT);
+            sourceCoordinator.updatePartitionForAcknowledgmentWait(indexPartition.getPartitionKey(), OWNERSHIP_TIMEOUT);
             acknowledgementSet.complete();
         } else {
             sourceCoordinator.closePartition(

data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/PitWorkerTest.java

@@ -69,6 +69,7 @@
 import static org.opensearch.dataprepper.plugins.source.opensearch.worker.PitWorker.EXTEND_KEEP_ALIVE_TIME;
 import static org.opensearch.dataprepper.plugins.source.opensearch.worker.PitWorker.STARTING_KEEP_ALIVE;
 import static org.opensearch.dataprepper.plugins.source.opensearch.worker.WorkerCommonUtils.ACKNOWLEDGEMENT_SET_TIMEOUT;
+import static org.opensearch.dataprepper.plugins.source.opensearch.worker.WorkerCommonUtils.OWNERSHIP_TIMEOUT;
 
 @ExtendWith(MockitoExtension.class)
 public class PitWorkerTest {
@@ -316,7 +317,7 @@ void run_with_acknowledgments_enabled_creates_and_deletes_pit_and_closes_that_pa
         when(schedulingParameterConfiguration.getInterval()).thenReturn(Duration.ZERO);
         when(openSearchSourceConfiguration.getSchedulingParameterConfiguration()).thenReturn(schedulingParameterConfiguration);
 
-        doNothing().when(sourceCoordinator).updatePartitionForAcknowledgmentWait(partitionKey, ACKNOWLEDGEMENT_SET_TIMEOUT);
+        doNothing().when(sourceCoordinator).updatePartitionForAcknowledgmentWait(partitionKey, OWNERSHIP_TIMEOUT);
         doNothing().when(sourceCoordinator).closePartition(partitionKey,
                 Duration.ZERO, 1, true);