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 cfc1ac664f [Improve][connector-elasticsearch-v2] Add slicing support
and e2e coverage for Elasticsearch source (#10454)
cfc1ac664f is described below
commit cfc1ac664f913ea9679f48d2faa6d9ecd7a1b47a
Author: cosmosni <[email protected]>
AuthorDate: Fri May 29 20:23:50 2026 +0800
[Improve][connector-elasticsearch-v2] Add slicing support and e2e coverage
for Elasticsearch source (#10454)
---
docs/en/connectors/source/Elasticsearch.md | 36 ++++++++-
docs/zh/connectors/source/Elasticsearch.md | 34 +++++++++
.../elasticsearch/client/EsRestClient.java | 29 ++++++--
.../elasticsearch/config/ElasticsearchConfig.java | 4 +
.../config/ElasticsearchSourceOptions.java | 8 ++
.../elasticsearch/source/ElasticsearchSource.java | 6 ++
.../source/ElasticsearchSourceReader.java | 55 ++++++++++----
.../source/ElasticsearchSourceSplitEnumerator.java | 20 +++--
.../connector/elasticsearch/ElasticsearchIT.java | 34 +++++++++
.../elasticsearch_source_with_pit_slice.conf | 85 ++++++++++++++++++++++
...elasticsearch_source_with_pit_slice_queued.conf | 85 ++++++++++++++++++++++
.../elasticsearch_source_with_scroll_slice.conf | 85 ++++++++++++++++++++++
12 files changed, 456 insertions(+), 25 deletions(-)
diff --git a/docs/en/connectors/source/Elasticsearch.md
b/docs/en/connectors/source/Elasticsearch.md
index 18101fbee9..2af91121de 100644
--- a/docs/en/connectors/source/Elasticsearch.md
+++ b/docs/en/connectors/source/Elasticsearch.md
@@ -49,6 +49,7 @@ support version >= 2.x and <= 8.x.
| pit_keep_alive | long | no | 60000 (1 minute)
|
| pit_batch_size | int | no | 100
|
| runtime_fields | array | no | -
|
+| slice_max | int | no | 1 (SCROLL: ES >= 5.0, PIT: ES
>= 7.10) |
| common-options | | no | -
|
@@ -256,6 +257,15 @@ runtime_fields = [
- Only Painless scripts are supported
- May be slower than indexed fields for large-scale queries
+### slice_max [int]
+Split a single index into multiple slices for parallel reads. Only effective
for SCROLL/PIT. Set to a value greater than 1 to enable slicing.
+
+**Version requirements:**
+- SCROLL slicing (sliced scroll) requires Elasticsearch 5.0 or higher.
+- PIT slicing requires Elasticsearch 7.10 or higher (PIT was introduced in
7.10.0).
+
+**Trade-off:** slicing improves throughput but may reduce snapshot consistency
across slices. For strong consistency, prefer PIT with a shared snapshot or set
`slice_max = 1`. For append-only or low-write workloads, slicing is usually
acceptable.
+
### common options
Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details
@@ -519,6 +529,30 @@ sink {
}
```
+Demo 9: PIT with slicing
+```hocon
+source {
+ Elasticsearch {
+ hosts = ["https://elasticsearch:9200";]
+ username = "elastic"
+ password = "elasticsearch"
+ tls_verify_certificate = false
+ tls_verify_hostname = false
+
+ index = "st_index"
+ query = {"range": {"c_int": {"gte": 10, "lte": 20}}}
+
+ search_type = DSL
+ search_api_type = PIT
+ pit_keep_alive = 60000
+ pit_batch_size = 100
+
+ # Enable slicing for parallel reads
+ slice_max = 2
+ }
+}
+```
+
## Changelog
-<ChangeLog />
\ No newline at end of file
+<ChangeLog />
diff --git a/docs/zh/connectors/source/Elasticsearch.md
b/docs/zh/connectors/source/Elasticsearch.md
index 1efd4a359c..8c3b4a6044 100644
--- a/docs/zh/connectors/source/Elasticsearch.md
+++ b/docs/zh/connectors/source/Elasticsearch.md
@@ -46,6 +46,7 @@ import ChangeLog from
'../changelog/connector-elasticsearch.md';
| tls_truststore_password | string | no | -
|
| pit_keep_alive | long | no | 60000 (1 minute)
|
| pit_batch_size | int | no | 100
|
+| slice_max | int | no | 1（SCROLL 需 ES >= 5.0，PIT 需 ES
>= 7.10） |
| runtime_fields | array | no | -
|
| common-options | | no | -
|
@@ -238,6 +239,15 @@ runtime_fields = [
- 适合临时分析、字段试验与低频查询
- 需要 Elasticsearch 7.11 及以上版本
+### slice_max [int]
+将单个索引拆分为多个切片以并行读取。仅对 SCROLL/PIT 生效，配置 > 1 时启用切片。
+
+**版本要求：**
+- SCROLL 切片（sliced scroll）需要 Elasticsearch 5.0 及以上版本。
+- PIT 切片需要 Elasticsearch 7.10 及以上版本（PIT 在 7.10.0 引入）。
+
+**取舍说明：**切片能提升吞吐，但可能降低跨切片的一致性。对一致性要求高时，建议使用 PIT（共享快照）或将 `slice_max =
1`；对追加写或写入较少的场景，开启切片通常可以接受。
+
### common options
Source 插件常用参数，具体请参考 [Source 常用选项](../common-options/source-common-options.md)
@@ -495,6 +505,30 @@ sink {
}
```
+Demo9: PIT + slicing 并行读取
+```hocon
+source {
+ Elasticsearch {
+ hosts = ["https://elasticsearch:9200";]
+ username = "elastic"
+ password = "elasticsearch"
+ tls_verify_certificate = false
+ tls_verify_hostname = false
+
+ index = "st_index"
+ query = {"range": {"c_int": {"gte": 10, "lte": 20}}}
+
+ search_type = DSL
+ search_api_type = PIT
+ pit_keep_alive = 60000
+ pit_batch_size = 100
+
+ # 开启切片并行读取
+ slice_max = 2
+ }
+}
+```
+
## 变更日志
<ChangeLog />
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java
index 6c57433ae9..6a43db1791 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java
@@ -200,7 +200,7 @@ public class EsRestClient implements Closeable {
Map<String, Object> query,
String scrollTime,
int scrollSize) {
- return searchByScroll(index, source, query, scrollTime, scrollSize,
null);
+ return searchByScroll(index, source, query, scrollTime, scrollSize,
null, null, null);
}
/**
@@ -219,7 +219,9 @@ public class EsRestClient implements Closeable {
Map<String, Object> query,
String scrollTime,
int scrollSize,
- Map<String, Object> runtimeFields) {
+ Map<String, Object> runtimeFields,
+ Integer sliceId,
+ Integer sliceMax) {
Map<String, Object> param = new HashMap<>();
param.put("query", query);
param.put("_source", source);
@@ -232,6 +234,12 @@ public class EsRestClient implements Closeable {
param.put("fields", new ArrayList<>(runtimeFields.keySet()));
}
+ if (sliceMax != null && sliceMax > 1) {
+ Map<String, Object> slice = new HashMap<>();
+ slice.put("id", sliceId == null ? 0 : sliceId);
+ slice.put("max", sliceMax);
+ param.put("slice", slice);
+ }
String endpoint = "/" + index + "/_search?scroll=" + scrollTime;
return getDocsFromScrollRequest(endpoint,
JsonUtils.toJsonString(param));
}
@@ -951,8 +959,11 @@ public class EsRestClient implements Closeable {
Map<String, Object> query,
int batchSize,
Object[] searchAfter,
- long keepAlive) {
- return searchWithPointInTime(pitId, source, query, batchSize,
searchAfter, keepAlive, null);
+ long keepAlive,
+ Integer sliceId,
+ Integer sliceMax) {
+ return searchWithPointInTime(
+ pitId, source, query, batchSize, searchAfter, keepAlive, null,
sliceId, sliceMax);
}
/**
@@ -974,7 +985,9 @@ public class EsRestClient implements Closeable {
int batchSize,
Object[] searchAfter,
long keepAlive,
- Map<String, Object> runtimeFields) {
+ Map<String, Object> runtimeFields,
+ Integer sliceId,
+ Integer sliceMax) {
Map<String, Object> requestBody = new HashMap<>();
requestBody.put("size", batchSize);
@@ -1002,6 +1015,12 @@ public class EsRestClient implements Closeable {
if (searchAfter != null && searchAfter.length > 0) {
requestBody.put("search_after", searchAfter);
}
+ if (sliceMax != null && sliceMax > 1) {
+ Map<String, Object> slice = new HashMap<>();
+ slice.put("id", sliceId == null ? 0 : sliceId);
+ slice.put("max", sliceMax);
+ requestBody.put("slice", slice);
+ }
String endpoint = "/_search";
Request request = new Request("POST", endpoint);
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchConfig.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchConfig.java
index d2758e6761..f0d12a1338 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchConfig.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchConfig.java
@@ -45,6 +45,8 @@ public class ElasticsearchConfig implements Serializable {
private int pitBatchSize;
private String pitId;
private Object[] searchAfter;
+ private int sliceId;
+ private int sliceMax = 1;
private Map<String, Object> runtimeFields;
@@ -67,6 +69,8 @@ public class ElasticsearchConfig implements Serializable {
elasticsearchConfig.setSearchAfter(searchAfter != null ?
searchAfter.clone() : null);
elasticsearchConfig.setRuntimeFields(
runtimeFields != null ? new HashMap<>(runtimeFields) : null);
+ elasticsearchConfig.setSliceId(sliceId);
+ elasticsearchConfig.setSliceMax(sliceMax);
return elasticsearchConfig;
}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchSourceOptions.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchSourceOptions.java
index 7a32497373..b5b43ff469 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchSourceOptions.java
@@ -119,4 +119,12 @@ public class ElasticsearchSourceOptions extends
ElasticsearchBaseOptions {
+ "Example: [{\"name\": \"day_of_week\",
\"type\": \"keyword\", \"script\":
\"emit(doc['timestamp'].value.dayOfWeekEnum.toString())\"}]. "
+ "Supported types: boolean, date, double,
geo_point, ip, keyword, long. "
+ "Available in Elasticsearch 7.11+");
+
+ public static final Option<Integer> SLICE_MAX =
+ Options.key("slice_max")
+ .intType()
+ .defaultValue(1)
+ .withDescription(
+ "Split a single index into multiple slices for
parallel reads. "
+ + "Only effective for Scroll/PIT, and
should be > 1 to enable slicing.");
}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSource.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSource.java
index d265b9fc96..46b01f2e89 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSource.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSource.java
@@ -181,6 +181,11 @@ public class ElasticsearchSource
String sqlQuery =
readonlyConfig.get(ElasticsearchSourceOptions.SQL_QUERY);
String scrollTime =
readonlyConfig.get(ElasticsearchSourceOptions.SCROLL_TIME);
int scrollSize =
readonlyConfig.get(ElasticsearchSourceOptions.SCROLL_SIZE);
+ int sliceMax =
readonlyConfig.get(ElasticsearchSourceOptions.SLICE_MAX);
+ if (SearchTypeEnum.SQL.equals(searchType) && sliceMax > 1) {
+ log.warn("SQL search_type does not support slicing. slice_max will
be ignored.");
+ sliceMax = 1;
+ }
long pitKeepAlive =
readonlyConfig.get(ElasticsearchSourceOptions.PIT_KEEP_ALIVE);
int pitBatchSize =
readonlyConfig.get(ElasticsearchSourceOptions.PIT_BATCH_SIZE);
@@ -208,6 +213,7 @@ public class ElasticsearchSource
elasticsearchConfig.setPitKeepAlive(pitKeepAlive);
elasticsearchConfig.setPitBatchSize(pitBatchSize);
+ elasticsearchConfig.setSliceMax(sliceMax);
return elasticsearchConfig;
}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceReader.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceReader.java
index d8cf3ba178..1faca89f15 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceReader.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceReader.java
@@ -117,11 +117,13 @@ public class ElasticsearchSourceReader
// Check if we should use PIT API
if
(SearchApiTypeEnum.PIT.equals(sourceIndexInfo.getSearchApiType())) {
log.info("Using Point-in-Time (PIT) API for index: {}",
sourceIndexInfo.getIndex());
+ logSliceInfo(sourceIndexInfo);
searchWithPointInTime(sourceIndexInfo, output, deserializer);
} else {
log.info("Using Scroll API for index: {}",
sourceIndexInfo.getIndex());
String scrollId = null;
try {
+ logSliceInfo(sourceIndexInfo);
ScrollResult scrollResult =
esRestClient.searchByScroll(
sourceIndexInfo.getIndex(),
@@ -129,7 +131,9 @@ public class ElasticsearchSourceReader
sourceIndexInfo.getQuery(),
sourceIndexInfo.getScrollTime(),
sourceIndexInfo.getScrollSize(),
- sourceIndexInfo.getRuntimeFields());
+ sourceIndexInfo.getRuntimeFields(),
+ sourceIndexInfo.getSliceId(),
+ sourceIndexInfo.getSliceMax());
scrollId = scrollResult.getScrollId();
outputFromScrollResult(scrollResult, sourceIndexInfo,
output, deserializer);
@@ -154,6 +158,16 @@ public class ElasticsearchSourceReader
}
}
+ private void logSliceInfo(ElasticsearchConfig sourceIndexInfo) {
+ if (sourceIndexInfo.getSliceMax() > 1) {
+ log.info(
+ "Elasticsearch slicing enabled. index={}, slice_id={},
slice_max={}",
+ sourceIndexInfo.getIndex(),
+ sourceIndexInfo.getSliceId(),
+ sourceIndexInfo.getSliceMax());
+ }
+ }
+
/**
* Search using Point-in-Time API.
*
@@ -166,15 +180,24 @@ public class ElasticsearchSourceReader
Collector<SeaTunnelRow> output,
SeaTunnelRowDeserializer deserializer) {
- // Create a PIT
- String pitId =
- esRestClient.createPointInTime(
- sourceIndexInfo.getIndex(),
sourceIndexInfo.getPitKeepAlive());
- sourceIndexInfo.setPitId(pitId);
- log.info(
- "Created Point-in-Time with ID: {} for index: {}",
- pitId,
- sourceIndexInfo.getIndex());
+ String pitId = sourceIndexInfo.getPitId();
+ boolean pitOwnedByReader = false;
+ if (StringUtils.isEmpty(pitId)) {
+ pitId =
+ esRestClient.createPointInTime(
+ sourceIndexInfo.getIndex(),
sourceIndexInfo.getPitKeepAlive());
+ sourceIndexInfo.setPitId(pitId);
+ pitOwnedByReader = true;
+ log.info(
+ "Created Point-in-Time with ID: {} for index: {}",
+ pitId,
+ sourceIndexInfo.getIndex());
+ } else {
+ log.info(
+ "Using shared Point-in-Time with ID: {} for index: {}",
+ pitId,
+ sourceIndexInfo.getIndex());
+ }
try {
// Initial search
@@ -186,7 +209,9 @@ public class ElasticsearchSourceReader
sourceIndexInfo.getPitBatchSize(),
null, // No search_after for first request
sourceIndexInfo.getPitKeepAlive(),
- sourceIndexInfo.getRuntimeFields());
+ sourceIndexInfo.getRuntimeFields(),
+ sourceIndexInfo.getSliceId(),
+ sourceIndexInfo.getSliceMax());
// Output the results
outputFromPitResult(pitResult, sourceIndexInfo, output,
deserializer);
@@ -206,18 +231,20 @@ public class ElasticsearchSourceReader
sourceIndexInfo.getPitBatchSize(),
sourceIndexInfo.getSearchAfter(),
sourceIndexInfo.getPitKeepAlive(),
- sourceIndexInfo.getRuntimeFields());
+ sourceIndexInfo.getRuntimeFields(),
+ sourceIndexInfo.getSliceId(),
+ sourceIndexInfo.getSliceMax());
// Output the results
outputFromPitResult(pitResult, sourceIndexInfo, output,
deserializer);
}
} finally {
// Always clean up the PIT when done
- if (pitId != null) {
+ if (pitOwnedByReader && pitId != null) {
try {
esRestClient.deletePointInTime(pitId);
} catch (Exception e) {
- log.warn("Failed to delete Point-in-Time with ID: " +
pitId, e);
+ log.warn("Failed to delete Point-in-Time with ID: {}",
pitId, e);
}
}
}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceSplitEnumerator.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceSplitEnumerator.java
index d501c649a8..ea1118734f 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceSplitEnumerator.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceSplitEnumerator.java
@@ -22,6 +22,7 @@ import org.apache.seatunnel.api.source.SourceSplitEnumerator;
import org.apache.seatunnel.common.exception.CommonErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsRestClient;
import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchConfig;
+import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.SearchTypeEnum;
import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.source.IndexDocsCount;
import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.ElasticsearchConnectorException;
@@ -148,12 +149,21 @@ public class ElasticsearchSourceSplitEnumerator
.filter(x -> x.getDocsCount() != null &&
x.getDocsCount() > 0)
.sorted(Comparator.comparingLong(IndexDocsCount::getDocsCount))
.collect(Collectors.toList());
+ int sliceMax = Math.max(1, elasticsearchConfig.getSliceMax());
+ if (SearchTypeEnum.SQL.equals(elasticsearchConfig.getSearchType())
&& sliceMax > 1) {
+ log.warn("SQL search_type does not support slicing. slice_max
will be ignored.");
+ sliceMax = 1;
+ }
for (IndexDocsCount indexDocsCount : indexDocsCounts) {
- ElasticsearchConfig cloneCfg = elasticsearchConfig.clone();
- cloneCfg.setIndex(indexDocsCount.getIndex());
- splits.add(
- new ElasticsearchSourceSplit(
-
String.valueOf(indexDocsCount.getIndex().hashCode()), cloneCfg));
+ String indexName = indexDocsCount.getIndex();
+ for (int sliceId = 0; sliceId < sliceMax; sliceId++) {
+ ElasticsearchConfig cloneCfg = elasticsearchConfig.clone();
+ cloneCfg.setIndex(indexName);
+ cloneCfg.setSliceId(sliceId);
+ cloneCfg.setSliceMax(sliceMax);
+ String splitId = sliceMax > 1 ? indexName + "#" + sliceId
: indexName;
+ splits.add(new ElasticsearchSourceSplit(splitId,
cloneCfg));
+ }
}
}
return splits;
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java
index 468c916b18..6b96499794 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java
@@ -424,6 +424,40 @@ public class ElasticsearchIT extends TestSuiteBase
implements TestResource {
log.info("Job output: {}", execResult.getStdout());
}
+ @TestTemplate
+ public void testElasticsearchWithPITSlice(TestContainer container)
+ throws IOException, InterruptedException {
+ Container.ExecResult execResult =
+
container.executeJob("/elasticsearch/elasticsearch_source_with_pit_slice.conf");
+ Assertions.assertEquals(0, execResult.getExitCode());
+ List<String> sinkData = readSinkDataWithSchema("st_index_pit_slice");
+ // for DSL is: {"range":{"c_int":{"gte":10,"lte":20}}}
+ Assertions.assertIterableEquals(mapTestDatasetForDSL(), sinkData);
+ }
+
+ @TestTemplate
+ public void testElasticsearchWithPITSliceQueued(TestContainer container)
+ throws IOException, InterruptedException {
+ Container.ExecResult execResult =
+ container.executeJob(
+
"/elasticsearch/elasticsearch_source_with_pit_slice_queued.conf");
+ Assertions.assertEquals(0, execResult.getExitCode());
+ List<String> sinkData =
readSinkDataWithSchema("st_index_pit_slice_queued");
+ // slice_max(4) > parallelism(2) forces queued splits on each reader
+ Assertions.assertIterableEquals(mapTestDatasetForDSL(), sinkData);
+ }
+
+ @TestTemplate
+ public void testElasticsearchWithScrollSlice(TestContainer container)
+ throws IOException, InterruptedException {
+ Container.ExecResult execResult =
+
container.executeJob("/elasticsearch/elasticsearch_source_with_scroll_slice.conf");
+ Assertions.assertEquals(0, execResult.getExitCode());
+ List<String> sinkData =
readSinkDataWithSchema("st_index_scroll_slice");
+ // for DSL is: {"range":{"c_int":{"gte":10,"lte":20}}}
+ Assertions.assertIterableEquals(mapTestDatasetForDSL(), sinkData);
+ }
+
@TestTemplate
public void testElasticsearchWithNestSchema(TestContainer container)
throws IOException, InterruptedException {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_source_with_pit_slice.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_source_with_pit_slice.conf
new file mode 100644
index 0000000000..93c72874fd
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_source_with_pit_slice.conf
@@ -0,0 +1,85 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+######
+###### This config file is a demonstration of using PIT API with slicing in
Elasticsearch connector
+######
+
+env {
+ parallelism = 2
+ job.mode = "BATCH"
+}
+
+source {
+ Elasticsearch {
+ hosts = ["https://elasticsearch:9200";]
+ username = "elastic"
+ password = "elasticsearch"
+ tls_verify_certificate = false
+ tls_verify_hostname = false
+
+ index = "st_index"
+ query = {"range": {"c_int": {"gte": 10, "lte": 20}}}
+
+ # Use DSL query with PIT API
+ search_type = "DSL"
+ search_api_type = "PIT"
+ pit_keep_alive = 60000 # 1 minute in milliseconds
+ pit_batch_size = 100
+
+ # Enable slicing for parallel reads
+ slice_max = 2
+
+ schema = {
+ fields {
+ c_map = "map<string, tinyint>"
+ c_array = "array<tinyint>"
+ c_string = string
+ c_boolean = boolean
+ c_tinyint = tinyint
+ c_smallint = smallint
+ c_bigint = bigint
+ c_float = float
+ c_double = double
+ c_decimal = "decimal(2, 1)"
+ c_bytes = bytes
+ c_int = int
+ c_date = date
+ c_timestamp = timestamp
+ c_null = "null"
+ }
+ }
+ }
+}
+
+transform {
+}
+
+sink {
+ Elasticsearch {
+ hosts = ["https://elasticsearch:9200";]
+ username = "elastic"
+ password = "elasticsearch"
+ tls_verify_certificate = false
+ tls_verify_hostname = false
+
+ index = "st_index_pit_slice"
+ index_type = "st"
+ "schema_save_mode"="CREATE_SCHEMA_WHEN_NOT_EXIST"
+ "data_save_mode"="APPEND_DATA"
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_source_with_pit_slice_queued.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_source_with_pit_slice_queued.conf
new file mode 100644
index 0000000000..811742fd02
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_source_with_pit_slice_queued.conf
@@ -0,0 +1,85 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+######
+###### PIT API with slice_max > parallelism to exercise queued-split path
+######
+
+env {
+ parallelism = 2
+ job.mode = "BATCH"
+}
+
+source {
+ Elasticsearch {
+ hosts = ["https://elasticsearch:9200";]
+ username = "elastic"
+ password = "elasticsearch"
+ tls_verify_certificate = false
+ tls_verify_hostname = false
+
+ index = "st_index"
+ query = {"range": {"c_int": {"gte": 10, "lte": 20}}}
+
+ # Use DSL query with PIT API
+ search_type = "DSL"
+ search_api_type = "PIT"
+ pit_keep_alive = 60000 # 1 minute in milliseconds
+ pit_batch_size = 100
+
+ # slice_max > parallelism forces queued slices on each reader
+ slice_max = 4
+
+ schema = {
+ fields {
+ c_map = "map<string, tinyint>"
+ c_array = "array<tinyint>"
+ c_string = string
+ c_boolean = boolean
+ c_tinyint = tinyint
+ c_smallint = smallint
+ c_bigint = bigint
+ c_float = float
+ c_double = double
+ c_decimal = "decimal(2, 1)"
+ c_bytes = bytes
+ c_int = int
+ c_date = date
+ c_timestamp = timestamp
+ c_null = "null"
+ }
+ }
+ }
+}
+
+transform {
+}
+
+sink {
+ Elasticsearch {
+ hosts = ["https://elasticsearch:9200";]
+ username = "elastic"
+ password = "elasticsearch"
+ tls_verify_certificate = false
+ tls_verify_hostname = false
+
+ index = "st_index_pit_slice_queued"
+ index_type = "st"
+ "schema_save_mode"="CREATE_SCHEMA_WHEN_NOT_EXIST"
+ "data_save_mode"="APPEND_DATA"
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_source_with_scroll_slice.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_source_with_scroll_slice.conf
new file mode 100644
index 0000000000..746c084a9f
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_source_with_scroll_slice.conf
@@ -0,0 +1,85 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+######
+###### This config file is a demonstration of using Scroll API with slicing in
Elasticsearch connector
+######
+
+env {
+ parallelism = 2
+ job.mode = "BATCH"
+}
+
+source {
+ Elasticsearch {
+ hosts = ["https://elasticsearch:9200";]
+ username = "elastic"
+ password = "elasticsearch"
+ tls_verify_certificate = false
+ tls_verify_hostname = false
+
+ index = "st_index"
+ query = {"range": {"c_int": {"gte": 10, "lte": 20}}}
+
+ # Use DSL query with Scroll API
+ search_type = "DSL"
+ search_api_type = "SCROLL"
+ scroll_time = "1m"
+ scroll_size = 100
+
+ # Enable slicing for parallel reads
+ slice_max = 2
+
+ schema = {
+ fields {
+ c_map = "map<string, tinyint>"
+ c_array = "array<tinyint>"
+ c_string = string
+ c_boolean = boolean
+ c_tinyint = tinyint
+ c_smallint = smallint
+ c_bigint = bigint
+ c_float = float
+ c_double = double
+ c_decimal = "decimal(2, 1)"
+ c_bytes = bytes
+ c_int = int
+ c_date = date
+ c_timestamp = timestamp
+ c_null = "null"
+ }
+ }
+ }
+}
+
+transform {
+}
+
+sink {
+ Elasticsearch {
+ hosts = ["https://elasticsearch:9200";]
+ username = "elastic"
+ password = "elasticsearch"
+ tls_verify_certificate = false
+ tls_verify_hostname = false
+
+ index = "st_index_scroll_slice"
+ index_type = "st"
+ "schema_save_mode"="CREATE_SCHEMA_WHEN_NOT_EXIST"
+ "data_save_mode"="APPEND_DATA"
+ }
+}
