ChuckLin2025 commented on code in PR #57462:
URL: https://github.com/apache/spark/pull/57462#discussion_r3642456735
##########
common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java:
##########
@@ -354,6 +389,54 @@ public void operationComplete(final Future<Channel>
handshakeFuture) {
return client;
}
+ /**
+ * If the given connection-failure cause was a rejection by a dead netty
event loop (its worker
+ * thread terminated and netty rejects new registrations with a
+ * {@link RejectedExecutionException}), replace the worker group so
subsequent connections bind to
+ * fresh, live threads. A dead loop is never replaced within a fixed-size
group and keeps being
+ * selected by the round-robin chooser, so without this the degradation is
permanent. See
+ * SPARK-58292.
+ */
+ private void recreateWorkerGroupIfEventLoopDead(EventLoopGroup connectGroup,
Throwable cause) {
+ if (!recreateWorkerGroupOnDeadEventLoop) {
+ return;
+ }
+ boolean eventLoopDead = false;
+ for (Throwable t = cause; t != null; t = t.getCause()) {
+ // Match ONLY the terminated-loop rejection, not a transient
task-queue-full rejection.
+ // netty's SingleThreadEventExecutor.reject() throws exactly this
message when isShutdown();
+ // the queue-full handler path throws a RejectedExecutionException with
no message.
+ if (t instanceof RejectedExecutionException
+ && "event executor terminated".equals(t.getMessage())) {
+ eventLoopDead = true;
+ break;
+ }
+ }
+ if (eventLoopDead) {
+ recreateWorkerGroup(connectGroup);
+ }
+ }
+
+ /**
+ * Replace the worker group with a fresh one, if it is still the group the
failed connection used
+ * ({@code connectGroup}). The superseded group is not shut down here: its
still-live threads may
+ * be serving channels that are already open. We keep a weak reference and
shut it down
+ * best-effort at {@link #close()}; its threads are daemon, so a
not-yet-collected group cannot
+ * block JVM shutdown. Synchronized and identity-guarded so concurrent
callers that all hit the
+ * same dead group replace it exactly once rather than spawning many groups.
+ */
+ private synchronized void recreateWorkerGroup(EventLoopGroup connectGroup) {
+ // A concurrent caller that hit the same dead group already swapped it
out; nothing to do.
+ if (workerGroup != connectGroup) {
+ return;
+ }
+ workerGroup = NettyUtils.createEventLoop(
+ ioMode, conf.clientThreads(), conf.getModuleName() + "-client");
Review Comment:
Do we need a unique name here ?
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
