Ngone51 commented on code in PR #57462:
URL: https://github.com/apache/spark/pull/57462#discussion_r3642697119
##########
common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java:
##########
@@ -354,6 +392,60 @@ 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())) {
Review Comment:
Shall we always print a log for this error even if recreation is disabled? I
think it'd be good for visibility.
--
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]
