(uniffle) branch master updated: [#2780] fix(client): Fix prefetch executor thread leaks (#2781)

[zu...@apache.org]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new b1ac32ffe [#2780] fix(client): Fix prefetch executor thread leaks (#2781)
b1ac32ffe is described below

commit b1ac32ffee1015ab45277821b4015f21eb5cddca
Author: Zhen Wang <64...@qq.com>
AuthorDate: Mon Aug 17 16:49:17 2026 +0800

    [#2780] fix(client): Fix prefetch executor thread leaks (#2781)
    
    ### What changes were proposed in this pull request?
    
    This PR fixes the prefetch executor thread leak in client read handlers.
    
    The following changes are included:
    
    - Invoke `super.close()` in `HadoopShuffleReadHandler.close()` to shut down the executor owned by `PrefetchableClientReadHandler`.
    - Implement `MultiReplicaClientReadHandler.close()` to close all underlying replica read handlers.
    - Initialize `prefetchExecutors` with a named daemon thread factory:
    
      ```java
      ThreadUtils.getThreadFactory("PrefetchableClientReadHandler")
      ```
    
      The generated thread names follow the `PrefetchableClientReadHandler-%d` format.
    
    ### Why are the changes needed?
    
    When prefetch is enabled, each `PrefetchableClientReadHandler` creates a dedicated executor.
    
    `HadoopShuffleReadHandler` overrides `close()` without invoking the parent implementation. In addition, `MultiReplicaClientReadHandler` does not propagate `close()` to its underlying handlers. Consequently, the prefetch executors may remain active after shuffle reading finishes.
    
    Under sustained workloads, the leaked threads can exhaust the process's native-thread resources and cause the following error:
    
    ```text
    java.lang.OutOfMemoryError: unable to create native thread: possibly out of memory or process/resource limits reached
        at java.base/java.lang.Thread.start0(Native Method)
        at java.base/java.lang.Thread.start(Thread.java:809)
        at java.base/java.util.concurrent.ThreadPoolExecutor.addWorker(ThreadPoolExecutor.java:945)
        at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1353)
        at java.base/java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:123)
        at org.apache.uniffle.storage.handler.impl.PrefetchableClientReadHandler.readShuffleData(PrefetchableClientReadHandler.java:96)
        at org.apache.uniffle.storage.handler.impl.HadoopClientReadHandler.readShuffleData(HadoopClientReadHandler.java:228)
        at org.apache.uniffle.storage.handler.impl.ComposedClientReadHandler.readShuffleData(ComposedClientReadHandler.java:122)
        at org.apache.uniffle.storage.handler.impl.MultiReplicaClientReadHandler.readShuffleData(MultiReplicaClientReadHandler.java:66)
    ```
    
    Closing the complete handler hierarchy ensures that the executors are shut down correctly. Using named daemon threads also improves thread identification and prevents prefetch threads from keeping the JVM alive.
    
    ### Does this PR introduce any user-facing changes?
    
    No.
    
    ### How was this patch tested?
---
 .../uniffle/storage/handler/impl/HadoopShuffleReadHandler.java       | 2 ++
 .../uniffle/storage/handler/impl/MultiReplicaClientReadHandler.java  | 5 +++++
 .../uniffle/storage/handler/impl/PrefetchableClientReadHandler.java  | 5 ++++-
 3 files changed, 11 insertions(+), 1 deletion(-)

diff --git a/storage/src/main/java/org/apache/uniffle/storage/handler/impl/HadoopShuffleReadHandler.java b/storage/src/main/java/org/apache/uniffle/storage/handler/impl/HadoopShuffleReadHandler.java
index 09372c32f..c49ecdcf2 100644
--- a/storage/src/main/java/org/apache/uniffle/storage/handler/impl/HadoopShuffleReadHandler.java
+++ b/storage/src/main/java/org/apache/uniffle/storage/handler/impl/HadoopShuffleReadHandler.java
@@ -217,6 +217,8 @@ public class HadoopShuffleReadHandler extends DataSkippableReadHandler {
   }
 
   public synchronized void close() {
+    super.close();
+
     try {
       dataReader.close();
     } catch (IOException ioe) {
diff --git a/storage/src/main/java/org/apache/uniffle/storage/handler/impl/MultiReplicaClientReadHandler.java b/storage/src/main/java/org/apache/uniffle/storage/handler/impl/MultiReplicaClientReadHandler.java
index 7b2774f6e..41501615e 100644
--- a/storage/src/main/java/org/apache/uniffle/storage/handler/impl/MultiReplicaClientReadHandler.java
+++ b/storage/src/main/java/org/apache/uniffle/storage/handler/impl/MultiReplicaClientReadHandler.java
@@ -100,6 +100,11 @@ public class MultiReplicaClientReadHandler extends AbstractClientReadHandler {
     } while (true);
   }
 
+  @Override
+  public void close() {
+    handlers.forEach(ClientReadHandler::close);
+  }
+
   @Override
   public void updateConsumedBlockInfo(BufferSegment bs, boolean isSkippedMetrics) {
     super.updateConsumedBlockInfo(bs, isSkippedMetrics);
diff --git a/storage/src/main/java/org/apache/uniffle/storage/handler/impl/PrefetchableClientReadHandler.java b/storage/src/main/java/org/apache/uniffle/storage/handler/impl/PrefetchableClientReadHandler.java
index 2082911ac..90aedc18a 100644
--- a/storage/src/main/java/org/apache/uniffle/storage/handler/impl/PrefetchableClientReadHandler.java
+++ b/storage/src/main/java/org/apache/uniffle/storage/handler/impl/PrefetchableClientReadHandler.java
@@ -31,6 +31,7 @@ import org.slf4j.LoggerFactory;
 
 import org.apache.uniffle.common.ShuffleDataResult;
 import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.ThreadUtils;
 import org.apache.uniffle.storage.handler.ClientReadHandlerMetric;
 
 public abstract class PrefetchableClientReadHandler extends AbstractClientReadHandler {
@@ -58,7 +59,9 @@ public abstract class PrefetchableClientReadHandler extends AbstractClientReadHa
       this.prefetchTimeoutSec = option.timeoutSec;
       this.prefetchResultQueue = new LinkedBlockingQueue<>(option.capacity);
       // todo: support multi threads to prefetch
-      this.prefetchExecutors = Executors.newFixedThreadPool(1);
+      this.prefetchExecutors =
+          Executors.newFixedThreadPool(
+              1, ThreadUtils.getThreadFactory("PrefetchableClientReadHandler"));
       this.abnormalFetchTag = new AtomicBoolean(false);
       this.finishedTag = new AtomicBoolean(false);
       this.queueingNumber = new AtomicInteger(0);