SteNicholas commented on code in PR #3778:
URL: https://github.com/apache/celeborn/pull/3778#discussion_r3755900017
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java:
##########
@@ -317,28 +349,39 @@ public void initChannel(SocketChannel ch) {
// Connect to the remote server
long preConnect = System.nanoTime();
ChannelFuture cf = bootstrap.connect(address);
- if (connectTimeoutMs <= 0) {
- awaitWithChannelCleanup(
- () -> {
- cf.await();
- return true;
- },
- cf);
- assert cf.isDone();
- if (cf.isCancelled()) {
+ try {
+ if (connectTimeoutMs <= 0) {
+ awaitWithChannelCleanup(
+ () -> {
+ cf.await();
+ return true;
+ },
+ cf);
+ assert cf.isDone();
+ if (cf.isCancelled()) {
+ closeChannel(cf);
+ throw new IOException(String.format("Connecting to %s cancelled",
address));
+ } else if (!cf.isSuccess()) {
+ closeChannel(cf);
+ throw new IOException(String.format("Failed to connect to %s",
address), cf.cause());
+ }
+ } else if (!awaitWithChannelCleanup(() -> cf.await(connectTimeoutMs),
cf)) {
closeChannel(cf);
- throw new IOException(String.format("Connecting to %s cancelled",
address));
- } else if (!cf.isSuccess()) {
+ throw new CelebornIOException(
+ String.format("Connecting to %s timed out (%s ms)", address,
connectTimeoutMs));
+ } else if (cf.cause() != null) {
closeChannel(cf);
- throw new IOException(String.format("Failed to connect to %s",
address), cf.cause());
+ throw new CelebornIOException(
+ String.format("Failed to connect to %s", address), cf.cause());
}
- } else if (!awaitWithChannelCleanup(() -> cf.await(connectTimeoutMs), cf))
{
- closeChannel(cf);
- throw new CelebornIOException(
- String.format("Connecting to %s timed out (%s ms)", address,
connectTimeoutMs));
- } else if (cf.cause() != null) {
- closeChannel(cf);
- throw new CelebornIOException(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.
retryCreateClient) retries.
+ recreateWorkerGroupIfEventLoopDead(connectGroup, cf.cause());
Review Comment:
Agreed -- installing the replacement and then failing the request that paid
for it was backwards.
`internalCreateClient` now takes a `retryOnRecreatedWorkerGroup` flag and
reconnects **inline** once, immediately after a successful swap: no `retryWait`
sleep, and not counted against `celeborn.<module>.io.maxRetries`. The flag is
cleared on the second attempt, so a single `createClient` makes at most two
connects.
This covers both cases you named:
- `maxRetries=1`, where `retryCreateClient` would throw on the first failure
with no retry left to spend on the fresh group.
- `createUnmanagedClient`, which has no retry wrapper at all and could
therefore never have benefited from the recreation.
`recreateWorkerGroupIfEventLoopDead` now returns whether a retry is
worthwhile. It returns `true` both when this caller swapped the group and when
a concurrent caller already did (`workerGroup != connectGroup`) -- in the
latter case the current group is already fresh, so the caller should still
retry rather than fail. It returns `false` when the config is off or the
factory is closed, so neither can trigger a retry.
Reusing the `decoder` on the inline retry is safe: the dead loop rejected
the channel *registration*, so the `ChannelInitializer` never ran and the
decoder was never added to a pipeline.
Covered by `recreatedWorkerGroupIsUsedWithoutConsumingTheRetryBudget`
(`io.maxRetries=1`, `createClient` still succeeds) and
`createUnmanagedClientRecoversFromDeadEventLoop`. Both fail without the inline
reconnect.
--
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]
