This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new cfd01ef342 [CELEBORN-2412] Propagate the worker error message through
cpp MessageDispatcher failures
cfd01ef342 is described below
commit cfd01ef3426e77c5dc60414504839c41559cace4
Author: Yu Gan <[email protected]>
AuthorDate: Wed Aug 12 15:19:14 2026 +0800
[CELEBORN-2412] Propagate the worker error message through cpp
MessageDispatcher failures
### What changes were proposed in this pull request?
`MessageDispatcher::read` decodes the error string a worker sends back on
`RPC_FAILURE` / `CHUNK_FETCH_FAILURE` and then throws it away — both paths
fulfil the pending promise with a blank `std::exception()`:
```cpp
holder.msgPromise.setException(folly::exception_wrapper(std::exception()));
```
This PR builds the exception from the message instead:
- `RPC_FAILURE` →
`folly::make_exception_wrapper<std::runtime_error>(failure->errorMsg())`
- `CHUNK_FETCH_FAILURE` → same, using the already-formatted `errorMsg` that
includes the `streamChunkSlice` context.
`MessageDispatcherTest` is strengthened accordingly: a shared
`takeExceptionMessage()` helper is added, the two failure tests now assert on
the message content rather than merely `hasException()`, and
`sendPushDataAndReceiveFailure` uses a real StatusCode name
(`PUSH_DATA_FAIL_PARTITION_NOT_FOUND`) so it exercises the classification path
it is meant to protect.
### Why are the changes needed?
`ShuffleClientImpl::getPushDataFailCause` classifies a push failure by
matching StatusCode names as substrings of the error message, and
`PushDataCallback::onFailure` / `PushMergedDataCallback` feed it
`exception->what()`. With a blank `std::exception`, `what()` is a fixed runtime
string that matches none of the 13 candidate causes, so classification always
falls through to `PUSH_DATA_FAIL_NON_CRITICAL_CAUSE`. Two consequences:
1. `excludeWorkerByCause` only acts on the connection-fail / timeout causes
and hits its `default:` branch every time, so
`celeborn.client.excludeWorker.enabled` never actually excludes a worker on the
C++ push path.
2. When revive attempts are exhausted, `classifyPushFailure` raises
`"PUSH_DATA_FAIL_NON_CRITICAL_CAUSE: <blank>"` — the worker's real reason
(partition not found, worker excluded, replica write failure, …) is gone from
the task-failure message, which makes these failures very hard to diagnose from
client logs alone.
Chunk fetch failures lose the worker's diagnostics the same way.
This restores parity with the Java client, where `TransportResponseHandler`
surfaces the worker's error text and `ShuffleClientImpl#getPushDataFailCause`
can act on it.
### Does this PR resolve a correctness bug?
- [ ] Yes
Shuffle output is unaffected, so this is not a result-correctness bug. It
is a failure-classification bug: worker exclusion never engages on the C++ push
path, and the worker's reason is missing from the task-failure message.
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
No config, API or behaviour change. Push and fetch failure messages now
carry the worker's reason instead of a placeholder.
### How was this patch tested?
Unit tests in `cpp/celeborn/network/tests/MessageDispatcherTest.cpp`:
- `sendRpcRequestAndReceiveFailure` — asserts the RPC failure message
reaches the caller.
- `sendPushDataAndReceiveFailure` — asserts a StatusCode name
(`PUSH_DATA_FAIL_PARTITION_NOT_FOUND`) survives on the exception, which is
exactly what `getPushDataFailCause` needs to see.
Covered by the `Celeborn Cpp Integration Test` workflow (`Run Unittests of
Celeborn Cpp`).
Closes #3791 from yugan95/CELEBORN-2412.
Authored-by: Yu Gan <[email protected]>
Signed-off-by: 子懿 <[email protected]>
---
cpp/celeborn/network/MessageDispatcher.cpp | 13 ++++++++---
.../network/tests/MessageDispatcherTest.cpp | 27 ++++++++++++++++++----
2 files changed, 33 insertions(+), 7 deletions(-)
diff --git a/cpp/celeborn/network/MessageDispatcher.cpp
b/cpp/celeborn/network/MessageDispatcher.cpp
index f70a396f86..c9d8735141 100644
--- a/cpp/celeborn/network/MessageDispatcher.cpp
+++ b/cpp/celeborn/network/MessageDispatcher.cpp
@@ -60,11 +60,16 @@ void MessageDispatcher::read(Context*,
std::unique_ptr<Message> toRecvMsg) {
registry.erase(failure->requestId());
return std::move(result);
});
+ const std::string errorMsg = failure->errorMsg();
LOG(ERROR) << "Rpc failed, requestId: " << failure->requestId()
- << " errorMsg: " << failure->errorMsg() << std::endl;
+ << " errorMsg: " << errorMsg << std::endl;
if (found) {
+ // Carry the worker's error message on the exception so the push/fetch
+ // callbacks can recover the precise cause via
+ // ShuffleClientImpl::getPushDataFailCause. A blank std::exception
+ // would collapse every failure into the non-critical default.
holder.msgPromise.setException(
- folly::exception_wrapper(std::exception()));
+ folly::make_exception_wrapper<std::runtime_error>(errorMsg));
}
return;
}
@@ -115,8 +120,10 @@ void MessageDispatcher::read(Context*,
std::unique_ptr<Message> toRecvMsg) {
failure->errorMsg());
LOG(ERROR) << errorMsg;
if (found) {
+ // Carry the streamChunkSlice context and the worker's error message so
+ // the reader's fetch-failure path sees the real cause.
holder.msgPromise.setException(
- folly::exception_wrapper(std::exception()));
+ folly::make_exception_wrapper<std::runtime_error>(errorMsg));
}
return;
}
diff --git a/cpp/celeborn/network/tests/MessageDispatcherTest.cpp
b/cpp/celeborn/network/tests/MessageDispatcherTest.cpp
index 45d2d4e82b..959aca6e32 100644
--- a/cpp/celeborn/network/tests/MessageDispatcherTest.cpp
+++ b/cpp/celeborn/network/tests/MessageDispatcherTest.cpp
@@ -61,6 +61,11 @@ std::unique_ptr<memory::ReadOnlyByteBuffer>
toReadOnlyByteBuffer(
return memory::ByteBuffer::toReadOnly(std::move(buffer));
}
+std::string takeExceptionMessage(
+ folly::Future<std::unique_ptr<Message>>&& future) {
+ return std::move(future).result().exception().what().toStdString();
+}
+
} // namespace
TEST(MessageDispatcherTest, sendRpcRequestAndReceiveResponse) {
@@ -124,7 +129,10 @@ TEST(MessageDispatcherTest,
sendRpcRequestAndReceiveFailure) {
std::make_unique<RpcFailure>(requestId, std::move(copiedErrorMsg));
dispatcher->read(nullptr, std::move(rpcFailure));
- EXPECT_TRUE(future.hasException());
+ ASSERT_TRUE(future.hasException());
+ EXPECT_NE(
+ takeExceptionMessage(std::move(future)).find(errorMsg),
+ std::string::npos);
}
TEST(MessageDispatcherTest, sendPushDataAndReceiveSuccess) {
@@ -204,13 +212,19 @@ TEST(MessageDispatcherTest,
sendPushDataAndReceiveFailure) {
EXPECT_EQ(
sentPushData->body()->readToString(requestBody.size()), requestBody);
- const std::string errorMsg = "test-error-msg";
+ // A push failure carries the worker's StatusCode name. It must survive on
the
+ // exception, otherwise ShuffleClientImpl::getPushDataFailCause cannot tell
+ // this apart from a generic failure and worker exclusion never engages.
+ const std::string errorMsg = "PUSH_DATA_FAIL_PARTITION_NOT_FOUND";
auto copiedErrorMsg = errorMsg;
auto rpcFailure =
std::make_unique<RpcFailure>(requestId, std::move(copiedErrorMsg));
dispatcher->read(nullptr, std::move(rpcFailure));
- EXPECT_TRUE(future.hasException());
+ ASSERT_TRUE(future.hasException());
+ EXPECT_NE(
+ takeExceptionMessage(std::move(future)).find(errorMsg),
+ std::string::npos);
}
TEST(MessageDispatcherTest, sendFetchChunkRequestAndReceiveSuccess) {
@@ -282,7 +296,12 @@ TEST(MessageDispatcherTest,
sendFetchChunkRequestAndReceiveFailure) {
streamChunkSlice, std::move(copiedErrorMsg));
dispatcher->read(nullptr, std::move(chunkFetchFailure));
- EXPECT_TRUE(future.hasException());
+ ASSERT_TRUE(future.hasException());
+ // The fetch-failure message must keep both the worker's error text and the
+ // streamChunkSlice it belongs to, so the reader can attribute the failure.
+ const auto exceptionMsg = takeExceptionMessage(std::move(future));
+ EXPECT_NE(exceptionMsg.find(errorMsg), std::string::npos);
+ EXPECT_NE(exceptionMsg.find(streamChunkSlice.toString()), std::string::npos);
}
TEST(MessageDispatcherTest, heartbeatIsSilentlyConsumed) {
