ChuckLin2025 commented on code in PR #57462:
URL: https://github.com/apache/spark/pull/57462#discussion_r3662572615
##########
common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java:
##########
@@ -288,20 +316,30 @@ public void initChannel(SocketChannel ch) {
long preConnect = System.nanoTime();
ChannelFuture cf = bootstrap.connect(address);
- if (connCreateTimeout <= 0) {
- cf.await();
- assert cf.isDone();
- if (cf.isCancelled()) {
- throw new IOException(String.format("Connecting to %s cancelled",
address));
- } else if (!cf.isSuccess()) {
+ try {
+ if (connCreateTimeout <= 0) {
+ cf.await();
+ assert cf.isDone();
+ if (cf.isCancelled()) {
+ throw new IOException(String.format("Connecting to %s cancelled",
address));
+ } else if (!cf.isSuccess()) {
+ throw new IOException(String.format("Failed to connect to %s",
address), cf.cause());
+ }
+ } else if (!cf.await(connCreateTimeout)) {
+ throw new IOException(
+ String.format("Connecting to %s timed out (%s ms)",
+ address, connCreateTimeout));
+ } else if (cf.cause() != null) {
throw new IOException(String.format("Failed to connect to %s",
address), cf.cause());
}
- } else if (!cf.await(connCreateTimeout)) {
- throw new IOException(
- String.format("Connecting to %s timed out (%s ms)",
- address, connCreateTimeout));
- } else if (cf.cause() != null) {
- throw new IOException(String.format("Failed to connect to %s", address),
cf.cause());
+ } catch (IOException e) {
+ // If the connection failed because the channel could not be registered
on its netty event
+ // loop (the loop's thread has died and netty rejects new tasks with
"event executor
+ // terminated"), the worker group is permanently degraded: that dead
loop is never replaced
+ // and keeps being handed out by the round-robin chooser. Replace the
group so retries bind
+ // to fresh live threads, then rethrow so the caller (e.g.
RetryingBlockTransferor) retries.
+ recreateWorkerGroupIfEventLoopDead(connectGroup, cf.cause());
Review Comment:
Done. Add a closed signale to guard this.
--
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]
