cloud-fan commented on code in PR #57462:
URL: https://github.com/apache/spark/pull/57462#discussion_r3658803883
##########
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:
Keep the factory closed on this path. After `close()` shuts down
`workerGroup`, a later `createClient()` fails with this same
terminated-executor cause and recreates a fresh live group, so the closed
factory is resurrected and can leak threads. Please guard replacement with
explicit closed state and extend `closeFactoryBeforeCreateClient` to assert the
group is never replaced.
##########
common/network-common/src/test/java/org/apache/spark/network/client/TransportClientFactorySuite.java:
##########
@@ -265,4 +265,49 @@ public void unlimitedConnectionAndCreationTimeouts()
throws IOException, Interru
assertNotEquals(exception.getCause(), null);
}
}
+
+ @Test
+ public void recreatesWorkerGroupWhenEventLoopIsDead() throws Exception {
+ // SPARK-58292: a dead netty worker event loop is never replaced within a
fixed-size group and
+ // permanently poisons connections. Simulate it by shutting down the
factory's worker group:
+ // the next createClient's channel register is rejected with "event
executor terminated", so
Review Comment:
```suggestion
// the next createClient's channel registration is rejected with "event
executor terminated", so
```
--
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]
