SteNicholas commented on code in PR #3724:
URL: https://github.com/apache/celeborn/pull/3724#discussion_r3410791487
##########
cpp/celeborn/client/ShuffleClient.cpp:
##########
@@ -1145,6 +1241,116 @@ bool ShuffleClientImpl::revive(
return false;
}
+namespace {
+// Mirrors org.apache.celeborn.common.util.ExceptionUtils#connectFail.
+bool connectFail(const std::string& message) {
+ return (message.find("Connection from ") != std::string::npos &&
+ message.find(" closed") != std::string::npos) ||
+ message.find("Connection reset by peer") != std::string::npos ||
+ message.find("Failed to send request ") != std::string::npos;
+}
+} // namespace
+
+protocol::StatusCode ShuffleClientImpl::getPushDataFailCause(
+ const std::string& message) {
+ VLOG(1) << "Push data failed cause message: " << message;
+ if (message.empty()) {
+ LOG(ERROR) << "Push data throw unexpected exception";
+ return protocol::StatusCode::PUSH_DATA_FAIL_NON_CRITICAL_CAUSE;
+ }
+ // The transport wraps the worker's error, so match the StatusCode name as a
+ // substring (via protocol::toString) rather than a prefix. Order follows the
+ // Java client so more specific causes win.
+ static constexpr std::array<protocol::StatusCode, 13> kCandidateCauses = {
+ protocol::StatusCode::PUSH_DATA_FAIL_NON_CRITICAL_CAUSE_REPLICA,
+ protocol::StatusCode::PUSH_DATA_WRITE_FAIL_REPLICA,
+ protocol::StatusCode::PUSH_DATA_WRITE_FAIL_PRIMARY,
+ protocol::StatusCode::PUSH_DATA_CREATE_CONNECTION_FAIL_PRIMARY,
+ protocol::StatusCode::PUSH_DATA_CREATE_CONNECTION_FAIL_REPLICA,
+ protocol::StatusCode::PUSH_DATA_CONNECTION_EXCEPTION_PRIMARY,
+ protocol::StatusCode::PUSH_DATA_CONNECTION_EXCEPTION_REPLICA,
+ protocol::StatusCode::PUSH_DATA_TIMEOUT_PRIMARY,
+ protocol::StatusCode::PUSH_DATA_TIMEOUT_REPLICA,
+ protocol::StatusCode::REPLICATE_DATA_FAILED,
+ protocol::StatusCode::PUSH_DATA_PRIMARY_WORKER_EXCLUDED,
+ protocol::StatusCode::PUSH_DATA_REPLICA_WORKER_EXCLUDED,
+ protocol::StatusCode::PUSH_DATA_FAIL_PARTITION_NOT_FOUND,
+ };
+ for (auto cause : kCandidateCauses) {
+ if (message.find(protocol::toString(cause)) != std::string::npos) {
+ return cause;
+ }
+ }
+ if (message.find("Timed out") != std::string::npos) {
+ // A client-side push timeout surfaces as a folly FutureTimeout ("Timed
+ // out") with no StatusCode token; classify it as a push timeout so worker
+ // exclusion engages like the Java client.
+ return protocol::StatusCode::PUSH_DATA_TIMEOUT_PRIMARY;
+ }
+ if (connectFail(message)) {
+ // Thrown when push to primary worker connection fails.
+ return protocol::StatusCode::PUSH_DATA_CONNECTION_EXCEPTION_PRIMARY;
+ }
+ return protocol::StatusCode::PUSH_DATA_FAIL_NON_CRITICAL_CAUSE;
+}
+
+std::optional<protocol::StatusCode> ShuffleClientImpl::classifyPushFailure(
+ const std::string& errorMsg,
+ int remainingReviveTimes,
+ PushState& pushState) {
+ const protocol::StatusCode cause = getPushDataFailCause(errorMsg);
+ if (remainingReviveTimes <= 0) {
+ // Out of revive attempts: surface a terminal failure annotated with the
+ // cause, like Java's `new CelebornIOException(cause, e)`.
+ pushState.setException(std::make_unique<std::runtime_error>(
+ protocol::toString(cause) + ": " + errorMsg));
Review Comment:
`errorMsg` doesn't reliably begin with — or even contain — the StatusCode
name, so the prefix isn't redundant:
- `getPushDataFailCause` matches the name as a **substring, not a prefix**
(per the comment above the candidate loop), so even on a hit the name may sit
anywhere inside the wrapped transport message.
- The fallback branches classify a cause whose name is **absent** from
`errorMsg`: `"Timed out"` → `PUSH_DATA_TIMEOUT_PRIMARY`, `connectFail(...)` →
`PUSH_DATA_CONNECTION_EXCEPTION_PRIMARY`, and empty/no-match →
`PUSH_DATA_FAIL_NON_CRITICAL_CAUSE`. Dropping `toString(cause)` would lose the
classified cause exactly where the heuristics did the work.
It also mirrors Java: `new CelebornIOException(cause, e)` resolves to
`this(statusCode.name(), cause)` — the message *is* the StatusCode name, with
the original error kept in the chained cause. Since `std::runtime_error` has no
cause chaining, `toString(cause) + ": " + errorMsg` carries both. Keeping it
as-is.
--
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]
