(celeborn) branch branch-0.6 updated: [CELEBORN-2063][FOLLOWUP] Fix timeout unit for parallel creation of partition writer

[ni...@apache.org]

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

nicholasjiang pushed a commit to branch branch-0.6
in repository https://gitbox.apache.org/repos/asf/celeborn.git


The following commit(s) were added to refs/heads/branch-0.6 by this push:
     new c78ef2b46 [CELEBORN-2063][FOLLOWUP] Fix timeout unit for parallel creation of partition writer
c78ef2b46 is described below

commit c78ef2b46619fa11a036044b342bfe718435f933
Author: SteNicholas <pr...@163.com>
AuthorDate: Thu Apr 9 20:35:35 2026 +0800

    [CELEBORN-2063][FOLLOWUP] Fix timeout unit for parallel creation of partition writer
    
    ### What changes were proposed in this pull request?
    
    Fix timeout unit for parallel creation of partition writer in `Utils#tryFuturesWithTimeout`.
    
    Follow up #3387, #3656.
    
    ### Why are the changes needed?
    
    `Utils#tryFuturesWithTimeout` uses wrong timeout unit which does not match the config option `celeborn.worker.writer.create.parallel.timeout` as follows:
    
    ```
    val WORKER_WRITER_CREATE_PARALLEL_TIMEOUT: ConfigEntry[Long] =
        buildConf("celeborn.worker.writer.create.parallel.timeout")
          .categories("worker")
          .version("0.6.3")
          .doc("Timeout for a worker to create a file writer in parallel.")
          .timeConf(TimeUnit.MILLISECONDS)
          .createWithDefaultString("120s")
    ```
    
    ### Does this PR resolve a correctness bug?
    
    No.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No.
    
    ### How was this patch tested?
    
    CI.
    
    Closes #3657 from SteNicholas/CELEBORN-2063.
    
    Authored-by: SteNicholas <pr...@163.com>
    Signed-off-by: SteNicholas <pr...@163.com>
    (cherry picked from commit c688c76d3c8de471afbf8c5fd1857b0a5926a721)
    Signed-off-by: SteNicholas <pr...@163.com>
---
 common/src/main/scala/org/apache/celeborn/common/util/Utils.scala | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala b/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
index 585ce4c56..109fdc326 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
@@ -1076,17 +1076,17 @@ object Utils extends Logging {
    * If other exception will be thrown directly.
    *
    * @param futures futures to try with timeout and callback
-   * @param timeoutInSeconds timeout limit value in seconds
+   * @param timeoutMs timeout limit value in milliseconds
    * @param errorMessage error message to log exception
    * @tparam T result type
    * @return results
    */
   def tryFuturesWithTimeout[T](
       futures: List[java.util.concurrent.Future[T]],
-      timeoutInSeconds: Long = 10,
+      timeoutMs: Long = 10000,
       errorMessage: String = "none"): List[T] = {
     try {
-      futures.map(_.get(timeoutInSeconds, TimeUnit.SECONDS))
+      futures.map(_.get(timeoutMs, TimeUnit.MILLISECONDS))
     } catch {
       case throwable: Throwable =>
         logError(