introduced boolean to control end to end Acknowledgment

Signed-off-by: Santhosh Gandhe <[email protected]>
opensearch-project · Jan 22, 2025 · f825a27 · f825a27
1 parent 784106e
commit f825a27
Show file tree

Hide file tree

Showing 5 changed files with 26 additions and 4 deletions.
diff --git a/.../jira-source/src/main/java/org/opensearch/dataprepper/plugins/source/jira/JiraClient.java b/.../jira-source/src/main/java/org/opensearch/dataprepper/plugins/source/jira/JiraClient.java
@@ -131,9 +131,13 @@ public void executePartition(SaasWorkerProgressState state,
                 .collect(Collectors.toList());
 
         try {
-            recordsToWrite.forEach(eventRecord -> acknowledgementSet.add(eventRecord.getData()));
-            buffer.writeAll(recordsToWrite, (int) Duration.ofSeconds(bufferWriteTimeoutInSeconds).toMillis());
-            acknowledgementSet.complete();
+            if (configuration.isAcknowledgments()) {
+                recordsToWrite.forEach(eventRecord -> acknowledgementSet.add(eventRecord.getData()));
+                buffer.writeAll(recordsToWrite, (int) Duration.ofSeconds(bufferWriteTimeoutInSeconds).toMillis());
+                acknowledgementSet.complete();
+            } else {
+                buffer.writeAll(recordsToWrite, (int) Duration.ofSeconds(bufferWriteTimeoutInSeconds).toMillis());
+            }
         } catch (Exception e) {
             throw new RuntimeException(e);
         }

diff --git a/...source/src/main/java/org/opensearch/dataprepper/plugins/source/jira/JiraSourceConfig.java b/...source/src/main/java/org/opensearch/dataprepper/plugins/source/jira/JiraSourceConfig.java
@@ -59,6 +59,13 @@ public class JiraSourceConfig implements CrawlerSourceConfig {
     @JsonProperty("backoff_time")
     private Duration backOff = DEFAULT_BACKOFF_MILLIS;
 
+    /**
+     * Boolean property indicating end to end acknowledgments state
+     */
+    @JsonProperty("acknowledgments")
+    @Getter
+    private boolean acknowledgments = false;
+
     public String getAccountUrl() {
         return this.getHosts().get(0);
     }

diff --git a/...c/main/java/org/opensearch/dataprepper/plugins/source/jira/rest/auth/JiraOauthConfig.java b/...c/main/java/org/opensearch/dataprepper/plugins/source/jira/rest/auth/JiraOauthConfig.java
@@ -154,6 +154,7 @@ public void renewCredentials() {
                             .getAccessToken().getValue();
                     this.refreshToken = (String) jiraSourceConfig.getAuthenticationConfig()
                             .getOauth2Config().getRefreshToken().getValue();
+                    this.expireTime = Instant.ofEpochMilli(System.currentTimeMillis() + (expiresInSeconds * 100L));
                 }
                 throw new RuntimeException("Failed to renew access token message:" + ex.getMessage(), ex);
             }

diff --git a/...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/CrawlerSourceConfig.java b/...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/CrawlerSourceConfig.java
@@ -6,4 +6,11 @@
 public interface CrawlerSourceConfig {
 
     int DEFAULT_NUMBER_OF_WORKERS = 1;
+
+    /**
+     * Boolean to indicate if acknowledgments enabled for this source
+     *
+     * @return boolean indicating acknowledgement state
+     */
+    boolean isAcknowledgments();
 }
diff --git a/...rch/dataprepper/plugins/source/source_crawler/coordination/scheduler/WorkerScheduler.java b/...rch/dataprepper/plugins/source/source_crawler/coordination/scheduler/WorkerScheduler.java
@@ -101,7 +101,10 @@ private void processPartition(EnhancedSourcePartition partition, Buffer<Record<E
         // Update the partition state or commit the partition as needed
         // Commit the partition to mark it as processed
         if (partition.getProgressState().isPresent()) {
-            AcknowledgementSet acknowledgementSet = createAcknowledgementSet(partition);
+            AcknowledgementSet acknowledgementSet = null;
+            if (sourceConfig.isAcknowledgments()) {
+                acknowledgementSet = createAcknowledgementSet(partition);
+            }
             crawler.executePartition((SaasWorkerProgressState) partition.getProgressState().get(), buffer, acknowledgementSet);
         }
         sourceCoordinator.completePartition(partition);