pass the maxPoll, timeout from config to poller (opensearch-project#17995)

Signed-off-by: Harsh Kothari <techarsh@amazon.com>

Author: yupeng9

server/src/main/java/org/opensearch/indices/pollingingest/DefaultStreamPoller.java

@@ -31,9 +31,6 @@
 public class DefaultStreamPoller implements StreamPoller {
     private static final Logger logger = LogManager.getLogger(DefaultStreamPoller.class);
 
-    public static final long MAX_POLL_SIZE = 1000;
-    public static final int POLL_TIMEOUT = 1000;
-
     private volatile State state = State.NONE;
 
     // goal state
@@ -53,6 +50,9 @@ public class DefaultStreamPoller implements StreamPoller {
     private ResetState resetState;
     private final String resetValue;
 
+    private long maxPollSize;
+    private int pollTimeout;
+
     private Set<IngestionShardPointer> persistedPointers;
 
     private final CounterMetric totalPolledCount = new CounterMetric();
@@ -84,7 +84,9 @@ public DefaultStreamPoller(
             resetState,
             resetValue,
             errorStrategy,
-            initialState
+            initialState,
+            maxPollSize,
+            pollTimeout
         );
     }
 
@@ -96,14 +98,18 @@ public DefaultStreamPoller(
         ResetState resetState,
         String resetValue,
         IngestionErrorStrategy errorStrategy,
-        State initialState
+        State initialState,
+        long maxPollSize,
+        int pollTimeout
     ) {
         this.consumer = Objects.requireNonNull(consumer);
         this.resetState = resetState;
         this.resetValue = resetValue;
         this.initialBatchStartPointer = startPointer;
         this.state = initialState;
         this.persistedPointers = persistedPointers;
+        this.maxPollSize = maxPollSize;
+        this.pollTimeout = pollTimeout;
         if (!this.persistedPointers.isEmpty()) {
             maxPersistedPointer = this.persistedPointers.stream().max(IngestionShardPointer::compareTo).get();
         }
@@ -197,13 +203,13 @@ protected void startPoll() {
                 List<IngestionShardConsumer.ReadResult<? extends IngestionShardPointer, ? extends Message>> results;
 
                 if (includeBatchStartPointer) {
-                    results = consumer.readNext(initialBatchStartPointer, true, MAX_POLL_SIZE, POLL_TIMEOUT);
+                    results = consumer.readNext(initialBatchStartPointer, true, maxPollSize, pollTimeout);
                     includeBatchStartPointer = false;
                 } else if (failedShardPointer != null) {
                     results = consumer.readNext(failedShardPointer, true, MAX_POLL_SIZE, POLL_TIMEOUT);
                     failedShardPointer = null;
                 } else {
-                    results = consumer.readNext(MAX_POLL_SIZE, POLL_TIMEOUT);
+                    results = consumer.readNext(maxPollSize, pollTimeout);
                 }
 
                 if (results.isEmpty()) {

server/src/test/java/org/opensearch/indices/pollingingest/DefaultStreamPollerTests.java

@@ -72,7 +72,9 @@ public void setUp() throws Exception {
             StreamPoller.ResetState.NONE,
             "",
             errorStrategy,
-            StreamPoller.State.NONE
+            StreamPoller.State.NONE,
+            1000,
+            1000
         );
         partitionedBlockingQueueContainer.startProcessorThreads();
     }
@@ -131,7 +133,9 @@ public void testSkipProcessed() throws InterruptedException {
             StreamPoller.ResetState.NONE,
             "",
             errorStrategy,
-            StreamPoller.State.NONE
+            StreamPoller.State.NONE,
+            1000,
+            1000
         );
 
         CountDownLatch latch = new CountDownLatch(2);
@@ -169,7 +173,9 @@ public void testResetStateEarliest() throws InterruptedException {
             StreamPoller.ResetState.EARLIEST,
             "",
             errorStrategy,
-            StreamPoller.State.NONE
+            StreamPoller.State.NONE,
+            1000,
+            1000
         );
         CountDownLatch latch = new CountDownLatch(2);
         doAnswer(invocation -> {
@@ -193,7 +199,9 @@ public void testResetStateLatest() throws InterruptedException {
             StreamPoller.ResetState.LATEST,
             "",
             errorStrategy,
-            StreamPoller.State.NONE
+            StreamPoller.State.NONE,
+            1000,
+            1000
         );
 
         poller.start();
@@ -213,7 +221,9 @@ public void testResetStateRewindByOffset() throws InterruptedException {
             StreamPoller.ResetState.REWIND_BY_OFFSET,
             "1",
             errorStrategy,
-            StreamPoller.State.NONE
+            StreamPoller.State.NONE,
+            1000,
+            1000
         );
         CountDownLatch latch = new CountDownLatch(1);
         doAnswer(invocation -> {
@@ -289,7 +299,9 @@ public void testDropErrorIngestionStrategy() throws TimeoutException, Interrupte
             StreamPoller.ResetState.NONE,
             "",
             errorStrategy,
-            StreamPoller.State.NONE
+            StreamPoller.State.NONE,
+            1000,
+            1000
         );
         poller.start();
         Thread.sleep(sleepTime);
@@ -340,7 +352,9 @@ public void testBlockErrorIngestionStrategy() throws TimeoutException, Interrupt
             StreamPoller.ResetState.NONE,
             "",
             errorStrategy,
-            StreamPoller.State.NONE
+            StreamPoller.State.NONE,
+            1000,
+            1000
         );
         poller.start();
         Thread.sleep(sleepTime);
@@ -369,7 +383,9 @@ public void testProcessingErrorWithBlockErrorIngestionStrategy() throws TimeoutE
             StreamPoller.ResetState.NONE,
             "",
             mockErrorStrategy,
-            StreamPoller.State.NONE
+            StreamPoller.State.NONE,
+            1000,
+            1000
         );
         poller.start();
         Thread.sleep(sleepTime);
@@ -433,7 +449,9 @@ public void testPersistedBatchStartPointer() throws TimeoutException, Interrupte
             StreamPoller.ResetState.NONE,
             "",
             errorStrategy,
-            StreamPoller.State.NONE
+            StreamPoller.State.NONE,
+            1000,
+            1000
         );
         poller.start();
         Thread.sleep(sleepTime);