This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12711-affbe05ff724cf8bbbb1bae109d15391cc1669ad in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 6b92e9ed7826538c280765c6bed5bc0e2c0c6f88 Author: Ryan Yang <[email protected]> AuthorDate: Sun Oct 11 09:40:37 2026 +0000 [Test][Connector-V2][Elasticsearch] Poll for index visibility in ElasticsearchIT slice tests (#12711) --- .../connector/elasticsearch/ElasticsearchIT.java | 33 ++++++++++++++++++---- 1 file changed, 27 insertions(+), 6 deletions(-) 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 79050d7be9..39c5cb3997 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 @@ -490,9 +490,16 @@ public class ElasticsearchIT extends TestSuiteBase implements TestResource { 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); + Awaitility.await() + .atMost(INDEX_REFRESH_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(1, TimeUnit.SECONDS) + .ignoreExceptions() + .untilAsserted( + () -> + Assertions.assertIterableEquals( + mapTestDatasetForDSL(), + readSinkDataWithSchema("st_index_pit_slice"))); } @TestTemplate @@ -502,9 +509,16 @@ public class ElasticsearchIT extends TestSuiteBase implements TestResource { 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); + Awaitility.await() + .atMost(INDEX_REFRESH_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(1, TimeUnit.SECONDS) + .ignoreExceptions() + .untilAsserted( + () -> + Assertions.assertIterableEquals( + mapTestDatasetForDSL(), + readSinkDataWithSchema("st_index_pit_slice_queued"))); } @TestTemplate @@ -513,9 +527,16 @@ public class ElasticsearchIT extends TestSuiteBase implements TestResource { 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); + Awaitility.await() + .atMost(INDEX_REFRESH_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(1, TimeUnit.SECONDS) + .ignoreExceptions() + .untilAsserted( + () -> + Assertions.assertIterableEquals( + mapTestDatasetForDSL(), + readSinkDataWithSchema("st_index_scroll_slice"))); } @TestTemplate
