You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "sunchao (via GitHub)" <gi...@apache.org> on 2026/06/24 18:07:28 UTC
[PR] [CELEBORN-2369] Retry stopped master Outbox failures [celeborn]
sunchao opened a new pull request, #3744:
URL: https://github.com/apache/celeborn/pull/3744
## Why are the changes needed?
When a Celeborn client loses its cached master connection, the associated Outbox can be stopped before queued or late master RPC messages are sent. Those requests fail with `CelebornException: Message is dropped because Outbox is stopped`.
`MasterClient` currently retries I/O and RPC timeout failures, but it does not recognize this stopped-Outbox failure when it is wrapped by the normal `awaitResult` path. In HA mode, a request can therefore fail instead of clearing the stale `RpcEndpointRef` and reconnecting to another available master.
The same legacy message is also used when the local `RpcEnv` shuts down permanently. Treating every occurrence as retryable without distinguishing shutdown would cause futile reconnect attempts through an already-stopped environment.
JIRA: https://issues.apache.org/jira/browse/CELEBORN-2369
## What changes were proposed in this PR?
- Recognize the exact legacy stopped-Outbox failure through its cause chain and reset the cached master endpoint so the existing HA retry path can reconnect.
- Use `RpcEnvStoppedException` for terminal local shutdown so shutdown failures remain non-retryable.
- Preserve each Outbox's original stop cause for queued and late messages.
- Fail messages submitted after `RpcEnv` shutdown immediately with the terminal cause.
- Add focused coverage for HA reconnection, terminal no-reconnect behavior, and transient versus terminal Outbox causes.
### Does this PR resolve a correctness bug?
- [x] Yes
### Does this PR introduce _any_ user-facing change?
- [x] Yes. Master RPCs can now fail over after a transient stopped-Outbox failure instead of failing immediately. No public API or configuration changes are introduced.
## How was this PR tested?
Using JDK 17:
- `build/mvn --no-transfer-progress -DskipTests spotless:apply`
- `build/mvn --no-transfer-progress -pl common -DskipTests test-compile`
- `build/mvn --no-transfer-progress -pl common -DargLine= -Dtest=org.apache.celeborn.common.client.MasterClientSuiteJ surefire:test` — 17 tests passed.
- `build/mvn --no-transfer-progress -pl common -DargLine= -DwildcardSuites=org.apache.celeborn.common.rpc.netty.OutboxSuite scalatest:test` — 3 tests passed.
- `build/mvn --no-transfer-progress -pl common -DargLine= -DwildcardSuites=org.apache.celeborn.common.rpc.netty.NettyRpcEnvSuite scalatest:test` — 33 tests passed.
- `build/mvn --no-transfer-progress -DskipTests spotless:check` — all 10 reactor modules passed.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2369] Retry stopped master Outbox failures [celeborn]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #3744:
URL: https://github.com/apache/celeborn/pull/3744#issuecomment-4848330723
Thanks for review and merge!!
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2369] Retry stopped master Outbox failures [celeborn]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #3744:
URL: https://github.com/apache/celeborn/pull/3744#discussion_r3476313093
##########
common/src/main/java/org/apache/celeborn/common/client/MasterClient.java:
##########
@@ -186,13 +189,28 @@ private boolean shouldRetry(@Nullable RpcEndpointRef oldRef, Throwable e) {
resetRpcEndpointRef(oldRef);
}
return true;
- } else if (e.getCause() instanceof IOException || e instanceof RpcTimeoutException) {
+ } else if (isRetryableRpcFailure(e)) {
resetRpcEndpointRef(oldRef);
return true;
}
return false;
}
+ private boolean isRetryableRpcFailure(Throwable throwable) {
+ Throwable current = throwable;
+ while (current != null) {
+ if (current instanceof IOException || current instanceof RpcTimeoutException) {
Review Comment:
Fixed in eedab7ef7. `isRetryableRpcFailure` now preserves the original `throwable.getCause() instanceof IOException || throwable instanceof RpcTimeoutException` checks, and only walks the cause chain for `OutboxStoppedException`. I also added nested IOException/RpcTimeout regression tests.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2369] Retry stopped master Outbox failures [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3744:
URL: https://github.com/apache/celeborn/pull/3744#discussion_r3472022435
##########
common/src/main/java/org/apache/celeborn/common/client/MasterClient.java:
##########
@@ -186,13 +189,28 @@ private boolean shouldRetry(@Nullable RpcEndpointRef oldRef, Throwable e) {
resetRpcEndpointRef(oldRef);
}
return true;
- } else if (e.getCause() instanceof IOException || e instanceof RpcTimeoutException) {
+ } else if (isRetryableRpcFailure(e)) {
resetRpcEndpointRef(oldRef);
return true;
}
return false;
}
+ private boolean isRetryableRpcFailure(Throwable throwable) {
+ Throwable current = throwable;
+ while (current != null) {
+ if (current instanceof IOException || current instanceof RpcTimeoutException) {
Review Comment:
This rewrite also changes the pre-existing `IOException` / `RpcTimeoutException` classification, which I don't think is the PR's intent.
Previously the check was `e.getCause() instanceof IOException || e instanceof RpcTimeoutException` — `IOException` was matched **only one level down** (at `e.getCause()`) and `RpcTimeoutException` **only at the top**. Walking the full cause chain now makes either type retryable **at any depth, including the top level**. So a top-level `IOException`, or one wrapped 2+ levels deep inside an otherwise-permanent failure, now triggers `resetRpcEndpointRef` + full HA failover/`maxRetries` where it previously surfaced immediately.
If broadening I/O retry is intended, could you note it in the PR description? If not, consider keeping the original `IOException`/`RpcTimeoutException` matching and only adding the new stopped-Outbox `CelebornException` arm to the walk.
##########
common/src/main/java/org/apache/celeborn/common/client/MasterClient.java:
##########
@@ -74,6 +75,8 @@ public MasterClient(RpcEnv rpcEnv, CelebornConf conf, boolean isWorker) {
}
private static final String SPLITTER = "#";
+ private static final String OUTBOX_STOPPED_MESSAGE =
Review Comment:
The retry contract now hinges on this exact string, which is independently re-declared in `Outbox.scala` (`stop()` -> `new CelebornException("Message is dropped because Outbox is stopped")`, ~line 260). The two copies are coupled only by byte-for-byte equality across a module/language boundary — no shared constant, no test asserting they match.
A reword, typo-fix, prefix, or i18n on either side would silently make this matcher stop recognizing the failure, regressing HA failover back to the exact bug this PR fixes, with no compile-time or test signal.
Suggest a dedicated marker exception type (e.g. `OutboxStoppedException extends CelebornException`) matched via `instanceof`, so producer and consumer share one source of truth instead of a string literal.
##########
common/src/main/scala/org/apache/celeborn/common/rpc/netty/Outbox.scala:
##########
@@ -248,16 +253,20 @@ private[celeborn] class Outbox(nettyEnv: NettyRpcEnv, val address: RpcAddress) {
client = null
}
- /**
- * Stop [[Outbox]]. The remaining messages in the [[Outbox]] will be notified with a
- * [[CelebornException]].
- */
- def stop(): Unit = {
+ /** Stop [[Outbox]] using a terminal cause when its owning RPC environment is shutting down. */
+ def stop(): Unit =
Review Comment:
The no-arg `stop()` derives opposite-retryability causes (`RpcEnvStoppedException` = terminal vs. the `CelebornException` = retryable) from a live read of `nettyEnv.isStopped`, outside the `synchronized` block in `stop(cause)`. The only remaining no-arg caller is `removeOutbox` (from `channelInactive` on a remote disconnect), so the retryability of those drained messages depends on a global mutable flag rather than the caller's own intent.
It's safe today because the env-shutdown path that observes `isStopped == true` is the intended-terminal case, but routing the decision through a global flag is fragile — consider having `removeOutbox` pass an explicit cause instead. (A marker exception type, per the `MasterClient` comment, would also remove the magic string from this branch.)
##########
common/src/test/scala/org/apache/celeborn/common/rpc/netty/OutboxSuite.scala:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.common.rpc.netty
+
+import java.nio.ByteBuffer
+import java.util.concurrent.{CountDownLatch, TimeUnit}
+import java.util.concurrent.atomic.AtomicReference
+
+import org.mockito.Mockito.{mock, when}
+
+import org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.common.exception.CelebornException
+import org.apache.celeborn.common.rpc.{RpcAddress, RpcEnvStoppedException}
+
+class OutboxSuite extends CelebornFunSuite {
+
+ private def failureMessage(
+ failure: AtomicReference[Throwable],
+ failed: CountDownLatch): RpcOutboxMessage =
+ RpcOutboxMessage(
+ ByteBuffer.allocate(0),
+ e => {
+ failure.set(e)
+ failed.countDown()
+ },
+ (_, _) => ())
+
+ test("send after terminal stop uses the original cause") {
+ val outbox = new Outbox(mock(classOf[NettyRpcEnv]), RpcAddress("localhost", 12345))
+ val cause = new RpcEnvStoppedException()
+ val failure = new AtomicReference[Throwable]()
+ val failed = new CountDownLatch(1)
+
+ outbox.stop(cause)
+ outbox.send(failureMessage(failure, failed))
+
+ assert(failed.await(10, TimeUnit.SECONDS))
+ assert(failure.get() eq cause)
+ }
+
+ test("send after transient stop remains retryable") {
+ val outbox = new Outbox(mock(classOf[NettyRpcEnv]), RpcAddress("localhost", 12345))
Review Comment:
`isStopped` is never stubbed on this mock, so the assertion that the failure is the retryable `CelebornException` relies on Mockito's default `false` return. If `isStopped` is later refactored or the mock made strict, `stop()` would silently take the `RpcEnvStoppedException` branch and this test would pass against the wrong (non-retryable) cause. Recommend stubbing `when(...isStopped).thenReturn(false)` explicitly to pin the intended path.
##########
common/src/test/scala/org/apache/celeborn/common/rpc/netty/NettyRpcEnvSuite.scala:
##########
@@ -65,6 +65,30 @@ class NettyRpcEnvSuite extends RpcEnvSuite with TimeLimits {
assert(e.getCause.getMessage.contains(uri))
}
+ test("ask through a stopped RPC environment fails immediately") {
+ val endpointName = "stopped-rpc-env"
+ env.setupEndpoint(
+ endpointName,
+ new RpcEndpoint {
+ override val rpcEnv: RpcEnv = env
+ override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
+ case message => context.reply(message)
+ }
+ })
+ val clientEnv = createRpcEnv(createCelebornConf(), "stopped-client", 0, clientMode = true)
Review Comment:
Unlike the other env-creating tests in this suite (e.g. lines 106-109, 195-202), `clientEnv` here isn't wrapped in `try { ... } finally { clientEnv.shutdown() }`. If `setupEndpointRef` throws (e.g. a bind/connect race), the env is never shut down and its Netty event-loop threads + timeout scheduler leak into subsequent tests (flaky-CI risk). Suggest the same try/finally pattern used elsewhere in the suite.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2369] Retry stopped master Outbox failures [celeborn]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #3744:
URL: https://github.com/apache/celeborn/pull/3744#discussion_r3476315344
##########
common/src/main/scala/org/apache/celeborn/common/rpc/netty/Outbox.scala:
##########
@@ -248,16 +253,20 @@ private[celeborn] class Outbox(nettyEnv: NettyRpcEnv, val address: RpcAddress) {
client = null
}
- /**
- * Stop [[Outbox]]. The remaining messages in the [[Outbox]] will be notified with a
- * [[CelebornException]].
- */
- def stop(): Unit = {
+ /** Stop [[Outbox]] using a terminal cause when its owning RPC environment is shutting down. */
+ def stop(): Unit =
Review Comment:
Good point. I left this unchanged intentionally: the shutdown path already passes `new RpcEnvStoppedException()` explicitly from `cleanup()`. The remaining no-arg path is `removeOutbox`; reading `isStopped` there distinguishes an active-env disconnect (retryable marker) from a disconnect racing with or after owner shutdown (terminal). Passing one fixed cause from `removeOutbox` would misclassify one of those cases.
##########
common/src/test/scala/org/apache/celeborn/common/rpc/netty/OutboxSuite.scala:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.common.rpc.netty
+
+import java.nio.ByteBuffer
+import java.util.concurrent.{CountDownLatch, TimeUnit}
+import java.util.concurrent.atomic.AtomicReference
+
+import org.mockito.Mockito.{mock, when}
+
+import org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.common.exception.CelebornException
+import org.apache.celeborn.common.rpc.{RpcAddress, RpcEnvStoppedException}
+
+class OutboxSuite extends CelebornFunSuite {
+
+ private def failureMessage(
+ failure: AtomicReference[Throwable],
+ failed: CountDownLatch): RpcOutboxMessage =
+ RpcOutboxMessage(
+ ByteBuffer.allocate(0),
+ e => {
+ failure.set(e)
+ failed.countDown()
+ },
+ (_, _) => ())
+
+ test("send after terminal stop uses the original cause") {
+ val outbox = new Outbox(mock(classOf[NettyRpcEnv]), RpcAddress("localhost", 12345))
+ val cause = new RpcEnvStoppedException()
+ val failure = new AtomicReference[Throwable]()
+ val failed = new CountDownLatch(1)
+
+ outbox.stop(cause)
+ outbox.send(failureMessage(failure, failed))
+
+ assert(failed.await(10, TimeUnit.SECONDS))
+ assert(failure.get() eq cause)
+ }
+
+ test("send after transient stop remains retryable") {
+ val outbox = new Outbox(mock(classOf[NettyRpcEnv]), RpcAddress("localhost", 12345))
Review Comment:
I left this unchanged intentionally. The test now asserts `OutboxStoppedException` specifically, so taking the `RpcEnvStoppedException` branch cannot silently pass; a strict mock would fail loudly rather than pass the wrong branch. Explicit stubbing would only duplicate Mockito's boolean default without increasing coverage.
##########
common/src/test/scala/org/apache/celeborn/common/rpc/netty/NettyRpcEnvSuite.scala:
##########
@@ -65,6 +65,30 @@ class NettyRpcEnvSuite extends RpcEnvSuite with TimeLimits {
assert(e.getCause.getMessage.contains(uri))
}
+ test("ask through a stopped RPC environment fails immediately") {
+ val endpointName = "stopped-rpc-env"
+ env.setupEndpoint(
+ endpointName,
+ new RpcEndpoint {
+ override val rpcEnv: RpcEnv = env
+ override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
+ case message => context.reply(message)
+ }
+ })
+ val clientEnv = createRpcEnv(createCelebornConf(), "stopped-client", 0, clientMode = true)
Review Comment:
Fixed in eedab7ef7. I wrapped endpoint setup, shutdown, and the assertion in `try/finally`; the finally block always calls `clientEnv.shutdown()` and `awaitTermination()`.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2369] Retry stopped master Outbox failures [celeborn]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #3744:
URL: https://github.com/apache/celeborn/pull/3744#discussion_r3476314611
##########
common/src/main/java/org/apache/celeborn/common/client/MasterClient.java:
##########
@@ -74,6 +75,8 @@ public MasterClient(RpcEnv rpcEnv, CelebornConf conf, boolean isWorker) {
}
private static final String SPLITTER = "#";
+ private static final String OUTBOX_STOPPED_MESSAGE =
Review Comment:
Fixed in eedab7ef7. I added `OutboxStoppedException extends CelebornException`; Outbox emits it and MasterClient matches it with `instanceof`, so there is no duplicated string contract.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org