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