SteNicholas commented on code in PR #3778:
URL: https://github.com/apache/celeborn/pull/3778#discussion_r3755896841
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java:
##########
@@ -92,7 +92,15 @@ public Channel getChannel() {
}
public boolean isActive() {
- return !timedOut && (channel.isOpen() || channel.isActive());
+ // A channel is pinned to one netty event-loop thread for its lifetime. If
that event loop has
+ // terminated (e.g. an uncaught error killed the thread; netty does not
replace it in a
+ // fixed-size EventLoopGroup), the channel can no longer send or complete
anything: writes and
+ // listener notifications route to the dead loop and are silently dropped.
Such a client must
+ // not be treated as active/reused, otherwise a request on it can orphan
and hang. See
+ // SPARK-58292.
+ return !timedOut
+ && !channel.eventLoop().isShuttingDown()
Review Comment:
You're right, and thanks for tracing it to `CelebornBufferStream`
specifically -- that is exactly the case that eviction alone does not cover,
since it holds the client for the lifetime of the partition and never
reacquires it.
Two things now fail the work instead of letting it orphan:
- `TransportClientFactory.failClientsOnDeadEventLoops()` -- once a
connection failure has revealed a dead loop, it synchronously fails the
outstanding requests of every pooled client still pinned to one. It
deliberately runs **outside** the connection-pool lock (from a `finally` in
`createClient`), because failing a request invokes its callback on the calling
thread and callbacks re-enter the factory; doing it under `clientPool.locks[i]`
would invert the pool-then-factory lock order.
- `TransportClient.sendRpc()` fails the callback up front rather than
writing into a dead loop, which covers credits issued *after* the sweep. This
matters most for RPCs specifically:
`failExpiredPushRequest`/`failExpiredFetchRequest` run on their own checker
threads and cover `outstandingPushes`/`outstandingFetches`, but
`outstandingRpcs` has no such backstop -- which is why the Flink credit path
was the one that hung indefinitely.
On force-closing: I could not make that work, and I think your own repro
says why -- `channel.close()` is itself submitted to the dead loop, so it never
takes effect. Failing the callbacks is the strongest available substitute, and
it is what lets the stream owner observe the failure: `onFailure` ->
`messageConsumer.accept(new TransportableError(...))` -> the reader tears down
and reopens.
Two related notes:
- `closeStream()` skipping `BUFFER_STREAM_END` is not a regression from the
`isActive()` change. Before it, the `sendRpc` was enqueued to the dead loop and
never executed, so the server did not receive it either way.
- While tracing this I found `CelebornBufferStream` logged `e.getCause()`
(null for these failures) in `addCredit`/`notifyRequiredSegment` and dropped
the throwable entirely in `openStreamInternal`, so the recovery was
undiagnosable in exactly this scenario. Fixed in the same commit.
Covered by
`TransportClientFactorySuiteJ.failsOutstandingRequestsOfPooledClientsOnDeadEventLoops`
and
`TransportClientSuiteJ.{sendRpcFailsFastWhenEventLoopIsDead,invalidatesClientWhenEventLoopIsDead}`.
--
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]
