You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "Kalvin2077 (via GitHub)" <gi...@apache.org> on 2026/07/08 14:05:56 UTC
[PR] [CELEBORN-2377] Support Heartbeat message decoding in C++ client [celeborn]
Kalvin2077 opened a new pull request, #3755:
URL: https://github.com/apache/celeborn/pull/3755
### What changes were proposed in this pull request?
This PR adds Heartbeat message decoding support to the C++ client network.
### Why are the changes needed?
The HEARTBEAT type (id 22) is already declared in the Message::Type enum and recognized by decodeType, but Message::decodeFrom had no case for it, so any heartbeat frame received by the C++ client hit the unsupported Message decode type failure path. As the protocol/server side begins to send heartbeat messages, the C++ client must be able to decode them instead of throwing.
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce any user-facing change?
- [ ] Yes
### How was this patch tested?
Added MessageTest.decodeHeartbeat, which constructs a heartbeat wire frame and asserts that Message::decodeFrom returns a non-null message with type HEARTBEAT. Built and ran the test with -- gtest_filter="MessageTest.decodeHeartbeat" inside the jraaaay/celeborn-cpp-dev:0.4 container; it 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-2377] Support Heartbeat message decoding in C++ client [celeborn]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3755:
URL: https://github.com/apache/celeborn/pull/3755#discussion_r3548547273
##########
cpp/celeborn/network/Message.cpp:
##########
@@ -218,5 +220,11 @@ void PushMergedData::internalEncodeTo(
protocol::encode(buffer, partitionUniqueIds_);
protocol::encode(buffer, batchOffsets_);
}
+
+std::unique_ptr<Heartbeat> Heartbeat::decodeFrom(
+ std::unique_ptr<memory::ReadOnlyByteBuffer>&& data) {
+ data->skip(1);
+ return std::make_unique<Heartbeat>();
+}
Review Comment:
Heartbeat::decodeFrom unconditionally skips 1 byte and returns a Heartbeat without validating the frame shape. If a malformed/forward-compatible frame includes extra bytes (e.g., non-zero bodyLength), this will silently ignore trailing bytes instead of failing fast.
##########
cpp/celeborn/network/Message.h:
##########
@@ -322,5 +322,14 @@ class PushMergedData : public Message {
std::vector<std::string> partitionUniqueIds_;
std::vector<int32_t> batchOffsets_;
};
+
+class Heartbeat : public Message {
+ public:
+ Heartbeat()
+ : Message(HEARTBEAT, memory::ReadOnlyByteBuffer::createEmptyBuffer()) {}
+
+ static std::unique_ptr<Heartbeat> decodeFrom(
+ std::unique_ptr<memory::ReadOnlyByteBuffer>&& data);
+};
Review Comment:
Heartbeat derives from Message but does not override internalEncodedLength()/internalEncodeTo(). If a Heartbeat instance is ever encoded (e.g., future client-side heartbeats), Message::encode will hit the CELEBORN_UNREACHABLE default and crash. Implementing the minimal encoder keeps the type symmetric with the Java protocol implementation (1 byte payload set to 0).
--
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-2377] Support Heartbeat message decoding in C++ client [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3755:
URL: https://github.com/apache/celeborn/pull/3755#issuecomment-4924826929
Thanks. Merged to main(v0.7.0).
--
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-2377] Support Heartbeat message decoding in C++ client [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas closed pull request #3755: [CELEBORN-2377] Support Heartbeat message decoding in C++ client
URL: https://github.com/apache/celeborn/pull/3755
--
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-2377] Support Heartbeat message decoding in C++ client [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3755:
URL: https://github.com/apache/celeborn/pull/3755#discussion_r3548609965
##########
cpp/celeborn/network/Message.h:
##########
@@ -322,5 +322,14 @@ class PushMergedData : public Message {
std::vector<std::string> partitionUniqueIds_;
std::vector<int32_t> batchOffsets_;
};
+
+class Heartbeat : public Message {
Review Comment:
Minor / optional (decode-only scope is otherwise fine):
1. `Heartbeat` overrides neither `internalEncodedLength()` nor `internalEncodeTo()`, so `Message::encode()` on a `Heartbeat` hits the base-class `CELEBORN_UNREACHABLE`. That's acceptable while the C++ client only ever *receives* heartbeats, but if a future change needs it to originate or respond with one (Java does `respond(new Heartbeat())` and periodic `writeAndFlush(new Heartbeat())`), encode support will be required.
2. Unlike every other `Message` subclass in this file, `Heartbeat` declares no copy constructor. Because the base `body_` is a `unique_ptr`, the implicit copy ctor is deleted, so `Heartbeat` is non-copyable — harmless today (the receive path only moves `unique_ptr<Message>`), but inconsistent with its siblings and would surface as a compile error the moment any code copies a `Heartbeat`.
##########
cpp/celeborn/network/Message.cpp:
##########
@@ -102,6 +102,8 @@ std::unique_ptr<Message> Message::decodeFrom(
return ChunkFetchSuccess::decodeFrom(std::move(data));
case CHUNK_FETCH_FAILURE:
return ChunkFetchFailure::decodeFrom(std::move(data));
+ case HEARTBEAT:
+ return Heartbeat::decodeFrom(std::move(data));
Review Comment:
Decoding the heartbeat here is correct, but note where the decoded message goes next: `MessageSerializeHandler::read` fires it to `MessageDispatcher::read` (the terminal handler attached via `setPipeline`), whose `switch` only handles `RPC_RESPONSE` / `RPC_FAILURE` / `CHUNK_FETCH_SUCCESS` / `CHUNK_FETCH_FAILURE`. A `HEARTBEAT` falls into the `default` branch, which does `LOG(ERROR) << "unsupported msg for dispatcher"`.
So once worker heartbeat is enabled (`celeborn.worker.push.heartbeat.enabled` / `celeborn.worker.fetch.heartbeat.enabled`, sent every `heartbeat.interval`, default 60s), every received heartbeat now produces a spurious ERROR log for the life of each connection. This PR correctly removes the hard `CELEBORN_FAIL` on decode, but to make heartbeats truly benign please also add a `case Message::HEARTBEAT: return;` (or a `VLOG`) to `MessageDispatcher::read` so they're consumed silently — the frame arriving is all that matters for read-idle; its payload is ignored, matching the Java `TransportChannelHandler.channelRead` path which just propagates a received `Heartbeat` without erroring.
##########
cpp/celeborn/network/tests/MessageTest.cpp:
##########
@@ -255,3 +255,19 @@ TEST(MessageTest, pushMergedDataCopyConstructor) {
EXPECT_EQ(copy.batchOffsets().size(), 2);
EXPECT_EQ(copy.batchOffsets()[1], 4);
}
+
+TEST(MessageTest, decodeHeartbeat) {
+ const int headerLength = sizeof(int32_t) + sizeof(uint8_t) + sizeof(int32_t);
+ const int encodedLength = 1;
+ const int bodyLength = 0;
+ size_t size = headerLength + encodedLength + bodyLength;
+ auto writeBuffer = memory::ByteBuffer::createWriteOnly(size);
+ writeBuffer->write<int32_t>(encodedLength);
+ writeBuffer->write<uint8_t>(Message::Type::HEARTBEAT);
+ writeBuffer->write<int32_t>(bodyLength);
+ writeBuffer->write<uint8_t>(0);
+ auto message = Message::decodeFrom(
+ memory::ByteBuffer::toReadOnly(std::move(writeBuffer)));
+ EXPECT_NE(message, nullptr);
+ EXPECT_EQ(message->type(), Message::Type::HEARTBEAT);
Review Comment:
Minor (test coverage): the test asserts only that the message is non-null and `type() == HEARTBEAT`; it never verifies that the one encoded byte was consumed. The single new behavior this PR adds — `data->skip(1)` and the empty body — is therefore unchecked: dropping the `skip(1)`, skipping the wrong count, or routing the leftover byte into the body would all still pass. Consider adding e.g. `EXPECT_EQ(message->body()->remainingSize(), 0);` to lock the 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