afterincomparableyum commented on code in PR #3737:
URL: https://github.com/apache/celeborn/pull/3737#discussion_r3489464660
##########
tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/HybridShuffleWordCountTest.scala:
##########
@@ -186,12 +188,18 @@ class HybridShuffleWordCountTest 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)) {
Review Comment:
I am going to leave this as is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
