(celeborn) branch main updated: [CELEBORN-2418] Fix flaky Flink WordCountTest file length check

[ni...@apache.org]

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

SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git


The following commit(s) were added to refs/heads/main by this push:
     new a0444ea4c5 [CELEBORN-2418] Fix flaky Flink WordCountTest file length check
a0444ea4c5 is described below

commit a0444ea4c5e3c44efbe073b16b8ebab32c543d93
Author: SparksFyz <sp...@gmail.com>
AuthorDate: Tue Aug 11 19:10:05 2026 +0800

    [CELEBORN-2418] Fix flaky Flink WordCountTest file length check
    
    ### What changes were proposed in this pull request?
    
      Wait for asynchronous disk flushing to complete before checking the file length in Flink `WordCountTest`.
    
      This applies the same bounded `eventually` check already used by `HybridShuffleWordCountTest`.
    
      ### Why are the changes needed?
    
      `DiskFileInfo` records the logical file length before `LocalFlusher` finishes writing the physical file.
    
      The test may check the file too early and observe a temporary length of zero, causing an intermittent failure.
    
      ### Does this PR resolve a correctness bug?
    
      - [ ] Yes
    
      ### Does this PR introduce _any_ user-facing change?
    
      - [ ] Yes
    
      ### How was this patch tested?
    
      - Verified the change with `git diff --check`.
      - The fix follows the existing handling in `HybridShuffleWordCountTest`.
    
    Closes #3796 from SparksFyz/fyz/fix-flink-word-count-flaky-ut.
    
    Authored-by: SparksFyz <sp...@gmail.com>
    Signed-off-by: Nicholas Jiang <pr...@163.com>
---
 .../celeborn/tests/flink/WordCountTest.scala       | 22 +++++++++++++++-------
 1 file changed, 15 insertions(+), 7 deletions(-)

diff --git a/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/WordCountTest.scala b/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/WordCountTest.scala
index e3cf461910..ab7051d912 100644
--- a/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/WordCountTest.scala
+++ b/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/WordCountTest.scala
@@ -31,7 +31,9 @@ import org.apache.flink.util.OperatingSystem
 import org.apache.hadoop.fs.Path
 import org.apache.hadoop.hdfs.MiniDFSCluster
 import org.scalatest.BeforeAndAfterAll
+import org.scalatest.concurrent.Eventually._
 import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.time.SpanSugar._
 
 import org.apache.celeborn.common.CelebornConf
 import org.apache.celeborn.common.CelebornConf.{ACTIVE_STORAGE_TYPES, AUTH_ENABLED, HDFS_DIR, INTERNAL_PORT_ENABLED, WORKER_STORAGE_CREATE_FILE_POLICY}
@@ -112,13 +114,19 @@ abstract class WordCountTestBase extends AnyFunSuite with Logging with MiniClust
   }
 
   private def checkFlushingFileLength(): Unit = {
-    workers.map(worker => {
-      worker.storageManager.workingDirWriters.values().asScala.map(writers => {
-        writers.forEach((fileName, fileWriter) => {
-          assert(new File(fileName).length() == fileWriter.getDiskFileInfo.getFileLength)
-        })
-      })
-    })
+    // getDiskFileInfo.getFileLength is the logical byte count accounted as data is written, while
+    // the physical file is grown asynchronously by the LocalFlusher. Right after the job finishes
+    // the flusher may not have drained the last buffers yet, so the on-disk length can lag (briefly
+    // even 0). Wait for the flush to catch up before asserting equality instead of reading mid-flush.
+    eventually(timeout(30.seconds), interval(500.milliseconds)) {
+      workers.foreach { worker =>
+        worker.storageManager.workingDirWriters.values().asScala.foreach { writers =>
+          writers.forEach((fileName, fileWriter) => {
+            assert(new File(fileName).length() == fileWriter.getDiskFileInfo.getFileLength)
+          })
+        }
+      }
+    }
   }
 }