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 ff13fb30b6 [Test][E2E] Wait for Doris multi-read sink visibility
(#11316)
ff13fb30b6 is described below
commit ff13fb30b6daa3888c0fffa33adef9fd7f0d78ca
Author: Daniel <[email protected]>
AuthorDate: Sat Jul 4 09:31:06 2026 +0800
[Test][E2E] Wait for Doris multi-read sink visibility (#11316)
Co-authored-by: DanielLeens <[email protected]>
---
.../e2e/connector/doris/DorisMultiReadIT.java | 45 +++++++++++++++++++---
1 file changed, 39 insertions(+), 6 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisMultiReadIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisMultiReadIT.java
index cbdbfc4ae4..8265e2d0e0 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisMultiReadIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisMultiReadIT.java
@@ -52,12 +52,21 @@ import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
+import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
+import static org.awaitility.Awaitility.await;
+
@Slf4j
public class DorisMultiReadIT extends AbstractDorisIT {
+ /**
+ * Multi-source Doris writes can become visible slightly after the
SeaTunnel job exits, so the
+ * sink assertions must wait for the final row count instead of querying
immediately.
+ */
+ private static final long SINK_ASSERT_TIMEOUT_MILLIS = 60_000L;
+
private static final String UNIQUE_TABLE_0 = "doris_e2e_unique_table_0";
private static final String UNIQUE_TABLE_1 = "doris_e2e_unique_table_1";
private static final String SOURCE_DB_0 = "e2e_source_0";
@@ -149,7 +158,14 @@ public class DorisMultiReadIT extends AbstractDorisIT {
protected void checkSinkData(String database, String tableName, String
sqlCondition) {
try {
+ String sourceSql =
+ String.format(
+ "select * from %s.%s %s order by F_ID ",
+ database, tableName, sqlCondition);
+ String sinkSql = String.format("select * from %s.%s order by
F_ID", sinkDB, tableName);
+
assertHasData(database, tableName);
+ awaitSinkTableReady(sourceSql, sinkSql);
assertHasData(sinkDB, tableName);
PreparedStatement sourcePre =
@@ -216,18 +232,23 @@ public class DorisMultiReadIT extends AbstractDorisIT {
sinkColumnType.toUpperCase(Locale.ROOT));
}
}
-
- String sourceSql =
- String.format(
- "select * from %s.%s %s order by F_ID ",
- database, tableName, sqlCondition);
- String sinkSql = String.format("select * from %s.%s order by
F_ID", sinkDB, tableName);
checkSourceAndSinkTableDate(sourceSql, sinkSql,
UNIQUE_TABLE_COLUMN_STRING);
} catch (Exception e) {
throw new RuntimeException("Doris connection error", e);
}
}
+ /**
+ * Wait until the Doris sink exposes the same number of rows as the
filtered source query before
+ * running schema and row-by-row assertions.
+ */
+ private void awaitSinkTableReady(String sourceSql, String sinkSql) {
+ int expectedRowCount = queryRowCount(sourceSql);
+ await().atMost(SINK_ASSERT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> Assertions.assertEquals(expectedRowCount,
queryRowCount(sinkSql)));
+ }
+
private void checkSourceAndSinkTableDate(String sourceSql, String sinkSql,
String columnsString)
throws Exception {
List<String> columnList =
@@ -278,6 +299,18 @@ public class DorisMultiReadIT extends AbstractDorisIT {
return -1;
}
+ private int queryRowCount(String sql) {
+ try (Statement statement =
+ conn.createStatement(
+ ResultSet.TYPE_SCROLL_SENSITIVE,
ResultSet.CONCUR_READ_ONLY);
+ ResultSet resultSet = statement.executeQuery(sql)) {
+ resultSet.last();
+ return resultSet.getRow();
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to query Doris table rows", e);
+ }
+ }
+
private void assertHasData(String db, String table) {
try (Statement statement = conn.createStatement()) {
String sql = String.format("select * from %s.%s limit 1", db,
table);
