This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 9140b90954 [Fix][Connector-V2] Fix HTTP cursor pagination no-progress
loop (#11098)
9140b90954 is described below
commit 9140b90954d43f9d4debe2b50dcf7a4595be23c2
Author: Jast <[email protected]>
AuthorDate: Thu Jun 18 08:24:46 2026 +0800
[Fix][Connector-V2] Fix HTTP cursor pagination no-progress loop (#11098)
---
.../seatunnel/http/source/HttpSourceReader.java | 8 +++-
.../http/HttpSourceReaderInternalPollNextTest.java | 48 ++++++++++++++++++++++
2 files changed, 54 insertions(+), 2 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java
index 2bc66a3796..3ed214ce67 100644
---
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java
@@ -464,6 +464,7 @@ public class HttpSourceReader extends
AbstractSingleSplitReader<SeaTunnelRow> {
// cursor pagination
if
(HttpPaginationType.CURSOR.getCode().equals(pageInfo.getPageType())) {
+ String currentCursor = pageInfo.getCursor();
// get cursor value from response JSON with fileName
String cursorResponseField =
pageInfo.getPageCursorResponseField();
ReadContext context =
JsonPath.using(jsonConfiguration).parse(data);
@@ -473,8 +474,11 @@ public class HttpSourceReader extends
AbstractSingleSplitReader<SeaTunnelRow> {
newCursor = cursorList.get(0);
}
pageInfo.setCursor(newCursor);
- // if not present cursor, then no more data
- noMoreElementFlag = Strings.isNullOrEmpty(newCursor);
+ // If the response cursor is empty or unchanged, the next
request cannot make
+ // progress.
+ noMoreElementFlag =
+ Strings.isNullOrEmpty(newCursor)
+ || Objects.equals(currentCursor, newCursor);
} else {
// if not set page pagination is default
// Determine whether the task is completed by specifying the
presence of the 'total
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/HttpSourceReaderInternalPollNextTest.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/HttpSourceReaderInternalPollNextTest.java
index 057eeb851b..13ac4dc8ac 100644
---
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/HttpSourceReaderInternalPollNextTest.java
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/HttpSourceReaderInternalPollNextTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.seatunnel.connectors.seatunnel.http;
+import org.apache.seatunnel.api.source.Boundedness;
import org.apache.seatunnel.api.source.Collector;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
@@ -42,10 +43,13 @@ import org.mockito.MockitoAnnotations;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class HttpSourceReaderInternalPollNextTest {
@@ -135,6 +139,50 @@ public class HttpSourceReaderInternalPollNextTest {
httpSourceReader.close();
}
+ @Test
+ public void testCursorPaginationStopsWhenCursorDoesNotAdvance() throws
Exception {
+
httpParameter.setBody("{\"scrollId\":\"${scrollId}\",\"capacity\":100}");
+
+ PageInfo pageInfo = new PageInfo();
+ pageInfo.setPageType(HttpPaginationType.CURSOR.getCode());
+ pageInfo.setPageCursorFieldName("scrollId");
+ pageInfo.setPageCursorResponseField("$.scrollId");
+ pageInfo.setUsePlaceholderReplacement(true);
+
+ when(context.getBoundedness()).thenReturn(Boundedness.BOUNDED);
+ AtomicInteger requestCount = new AtomicInteger();
+ when(httpClientProvider.execute(
+ anyString(), anyString(), any(), any(), any(),
anyBoolean()))
+ .thenAnswer(
+ invocation -> {
+ int currentRequest =
requestCount.incrementAndGet();
+ if (currentRequest == 1) {
+ return new HttpResponse(
+ 200,
+
"{\"scrollId\":\"cursor-1\",\"data\":[{\"key1\":\"v1\",\"key2\":\"v2\"}]}");
+ }
+ if (currentRequest == 2) {
+ return new HttpResponse(
+ 200,
"{\"scrollId\":\"cursor-1\",\"data\":[]}");
+ }
+ throw new AssertionError(
+ "Cursor pagination should stop when the
cursor does not advance");
+ });
+
+ httpSourceReader =
+ new HttpSourceReader(
+ httpParameter, context, deserializationSchema, null,
"$.data", pageInfo);
+ httpSourceReader.open();
+ httpSourceReader.setHttpClient(httpClientProvider);
+
+ httpSourceReader.internalPollNext(collector);
+
+ verify(httpClientProvider, times(2))
+ .execute(anyString(), anyString(), any(), any(), any(),
anyBoolean());
+ verify(context, times(1)).signalNoMoreElement();
+ httpSourceReader.close();
+ }
+
@AfterEach
public void tearDown() throws Exception {
mock.close();
