(celeborn) branch main updated: [CELEBORN-2462][WORKER] Reduce misleading WARN/ERROR logs during decommission shutdown

[zaynt]

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

zaynt4606 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 1fe959a322 [CELEBORN-2462][WORKER] Reduce misleading WARN/ERROR logs 
during decommission shutdown
1fe959a322 is described below

commit 1fe959a3226e53602bebf05a39124875fc0d864f
Author: yew1eb <yew...@gmail.com>
AuthorDate: Tue Oct 13 12:34:00 2026 +0800

    [CELEBORN-2462][WORKER] Reduce misleading WARN/ERROR logs during 
decommission shutdown
    
    ### What changes were proposed in this pull request?
    
    1. `Worker.decommissionWorker()`: skip the state transition when the worker 
is already in `InDecommission` state.
    2. `ReadBufferDispatcher`: catch `InterruptedException` separately in the 
dispatcher loop — log as INFO when shutting down (`stopFlag=true`), WARN 
otherwise.
    
    ### Why are the changes needed?
    
    When a worker exits after decommission is triggered, the shutdown hook 
prints misleading logs that look like real errors:
    
    <img width="1836" height="1626" alt="image" 
src="https://github.com/user-attachments/assets/ec6801ed-dbb1-4b72-b729-0b84fe3af941";
 />
    
    Both are harmless shutdown noise:
    
    1. The worker is already in `InDecommission` state (set by the decommission 
event path via heartbeat response or REST `/exit?type=DECOMMISSION`) when the 
shutdown hook calls `decommissionWorker()`, and the state machine has no 
self-loop, so the repeated transition is rejected with a WARN.
    2. `ReadBufferDispatcher.close()` sets `stopFlag=true` and interrupts the 
dispatcher thread blocked in `requests.poll()`; the `InterruptedException` 
falls into the generic `catch (Throwable)` and is logged as ERROR with a `null` 
message (`InterruptedException.getMessage()` returns null).
    
    ### Does this PR resolve a correctness bug?
    
    - [ ] Yes
    
    ### Does this PR introduce _any_ user-facing change?
    
    - [ ] Yes
    
    ### How was this patch tested?
    
    Compiled the worker module: `build/mvn -pl worker -am -DskipTests compile`. 
Logging-only change, no behavior change.
    
    Closes #3847 from yew1eb/CELEBORN-2462.
    
    Authored-by: yew1eb <yew...@gmail.com>
    Signed-off-by: zhengtao <shuaizhentao....@alibaba-inc.com>
    Co-Authored-By: Codex <nore...@openai.com>
    AI-Model: gpt-6
    AI-Contributed/Feature: 0/10
    AI-Contributed/UT: 0/0
---
 .../celeborn/service/deploy/worker/memory/ReadBufferDispatcher.java | 6 ++++++
 .../scala/org/apache/celeborn/service/deploy/worker/Worker.scala    | 4 +++-
 2 files changed, 9 insertions(+), 1 deletion(-)

diff --git 
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/ReadBufferDispatcher.java
 
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/ReadBufferDispatcher.java
index c729b8b6be..e7a5eb9673 100644
--- 
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/ReadBufferDispatcher.java
+++ 
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/ReadBufferDispatcher.java
@@ -151,6 +151,12 @@ public class ReadBufferDispatcher {
             }
             request.getBufferListener().notifyBuffers(null, e);
           }
+        } catch (InterruptedException e) {
+          if (stopFlag) {
+            logger.info("Read buffer dispatcher is shutting down.");
+          } else {
+            logger.warn("Read buffer dispatcher interrupted unexpectedly.", e);
+          }
         } catch (Throwable e) {
           logger.error("Read buffer dispatcher encountered error: {}", 
e.getMessage(), e);
         }
diff --git 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
index 79f3f20b32..8d713f2700 100644
--- 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
+++ 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
@@ -1060,7 +1060,9 @@ private[celeborn] class Worker(
 
   def decommissionWorker(): Unit = {
     logInfo("Worker start to decommission")
-    workerStatusManager.transitionState(State.InDecommission)
+    if (workerStatusManager.currentWorkerStatus.getState != 
State.InDecommission) {
+      workerStatusManager.transitionState(State.InDecommission)
+    }
     sendWorkerDecommissionToMaster()
     shutdown.set(true)
     val interval = conf.workerDecommissionCheckInterval