This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d35272a7447c74156bcd09814560e07c801b1743 Author: Vova Kolmakov <[email protected]> AuthorDate: Tue Jun 30 08:30:52 2026 +0700 test(flink): de-flake testLookupJoin lookup-join IT (#19093) The streaming lookup join in ITTestHoodieDataSource.testLookupJoin reads its dimension table through HoodieLookupFunction, whose cache is populated lazily on the first probe row. A teardown / commit-visibility race can occasionally make the join emit no rows; execInsertSql awaits the insert but swallows the exception, so the empty result is committed to t2 and the read then asserts against an empty table. Both tables use uuid as the record key, so re-running the insert is an idempotent upsert. Retry the insert-and-read up to MAX_STREAM_READ_ATTEMPTS until the expected rows materialize, mirroring the existing stream-read de-flake pattern (submitAndFetchWithRetry). Co-authored-by: Vova Kolmakov <[email protected]> (cherry picked from commit 848d33e11823f1c6bcfc237d44229315e7781d30) --- .../org/apache/hudi/table/ITTestHoodieDataSource.java | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java index 399829f94189..148e14f30b2f 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java @@ -1198,9 +1198,22 @@ public class ITTestHoodieDataSource { + " join t1/*+ OPTIONS('lookup.join.cache.ttl'='2 day', 'lookup.async'='" + async + "'," + " 'lookup.join.cache.type'='" + cacheType + "') */ " + " FOR SYSTEM_TIME AS OF o.proc_time AS b on o.uuid = b.uuid"; - execInsertSql(tableEnv, sql); - List<Row> result = CollectionUtil.iterableToList( - () -> tableEnv.sqlQuery("select * from t2").execute().collect()); + + // The lookup function loads the dimension table lazily on the first probe row, so a teardown / + // commit-visibility race can occasionally make the join emit no rows. Re-running the upsert into + // the uuid-keyed table t2 is idempotent, so retry until the expected rows materialize. + final int expectedNum = TestData.DATA_SET_SOURCE_INSERT.size(); + List<Row> result = Collections.emptyList(); + for (int attempt = 1; attempt <= MAX_STREAM_READ_ATTEMPTS; attempt++) { + execInsertSql(tableEnv, sql); + result = CollectionUtil.iterableToList( + () -> tableEnv.sqlQuery("select * from t2").execute().collect()); + if (result.size() >= expectedNum) { + break; + } + LOG.warn("testLookupJoin collected {} of {} rows on attempt {}/{}; a teardown race produced an " + + "empty lookup join. Retrying.", result.size(), expectedNum, attempt, MAX_STREAM_READ_ATTEMPTS); + } assertRowsEquals(result, TestData.DATA_SET_SOURCE_INSERT); }
