Fix issue where we start from checkpoint for PIT with acks to instead…

… start from beginning (#3610) Signed-off-by: Taylor Gray <[email protected]>
opensearch-project · Nov 8, 2023 · f08c16c · f08c16c
1 parent 3e558c0
commit f08c16c
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 2 deletions.
diff --git a/.../src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/PitWorker.java b/.../src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/PitWorker.java
@@ -156,7 +156,8 @@ private void processIndex(final SourcePartition<OpenSearchIndexProgressState> op
         LOG.info("Starting processing for index: '{}'", indexName);
         Optional<OpenSearchIndexProgressState> openSearchIndexProgressStateOptional = openSearchIndexPartition.getPartitionState();
 
-        if (openSearchIndexProgressStateOptional.isEmpty()) {
+        // We can't checkpoint acks yet so need to restart from the beginning of index when acks are enabled for now
+        if (openSearchSourceConfiguration.isAcknowledgmentsEnabled() || openSearchIndexProgressStateOptional.isEmpty()) {
             openSearchIndexProgressStateOptional = Optional.of(initializeProgressState());
         }
 

diff --git a/...n/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/WorkerCommonUtils.java b/...n/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/WorkerCommonUtils.java
@@ -28,7 +28,7 @@ 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.ofHours(1);
+    static final Duration ACKNOWLEDGEMENT_SET_TIMEOUT = Duration.ofMinutes(20);
     static final Duration STARTING_BACKOFF = Duration.ofMillis(500);
     static final Duration MAX_BACKOFF = Duration.ofSeconds(60);
     static final int BACKOFF_RATE = 2;