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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
