(hudi) branch master updated: test(flink): de-flake testLookupJoin lookup-join IT (#19093)

[danny0405]

This is an automated email from the ASF dual-hosted git repository.

danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git


The following commit(s) were added to refs/heads/master by this push:
     new 848d33e11823 test(flink): de-flake testLookupJoin lookup-join IT 
(#19093)
848d33e11823 is described below

commit 848d33e11823f1c6bcfc237d44229315e7781d30
Author: Vova Kolmakov <wombatu...@gmail.com>
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 <wombatu...@apache.org>
---
 .../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 dbe596007f7a..b91cf3ef3324 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
@@ -1246,9 +1246,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);
   }