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