You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "HolyLow (via GitHub)" <gi...@apache.org> on 2025/09/19 07:31:18 UTC
[PR] [CELEBORN-2157][CIP-14] Support sending RegisterShuffle/Revive/MapperEnd messages for NettyRpcEndpointRef in cppClient [celeborn]
HolyLow opened a new pull request, #3484:
URL: https://github.com/apache/celeborn/pull/3484
### What changes were proposed in this pull request?
This PR supports sending RegisterShuffle/Revive/MapperEnd messages for NettyRpcEndpointRef in cppClient.
### Why are the changes needed?
These messages are used to communicate with CelebornMaster when writing data with cppClient.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Compilation and UTs.
--
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-2157][CIP-14] Support sending RegisterShuffle/Revive/MapperEnd messages for NettyRpcEndpointRef in cppClient [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3484:
URL: https://github.com/apache/celeborn/pull/3484#issuecomment-3388701512
Thanks. Merged to main(v0.7.0) and branch-0.6(v0.6.2).
--
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-2157][CIP-14] Support sending RegisterShuffle/Revive/MapperEnd messages for NettyRpcEndpointRef in cppClient [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas closed pull request #3484: [CELEBORN-2157][CIP-14] Support sending RegisterShuffle/Revive/MapperEnd messages for NettyRpcEndpointRef in cppClient
URL: https://github.com/apache/celeborn/pull/3484
--
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-2157][CIP-14] Support sending RegisterShuffle/Revive/MapperEnd messages for NettyRpcEndpointRef in cppClient [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3484:
URL: https://github.com/apache/celeborn/pull/3484#discussion_r2374819451
##########
cpp/celeborn/network/tests/NettyRpcEndpointRefTest.cpp:
##########
@@ -65,54 +77,150 @@ void readRpcAddress(
readUTF(buffer, host);
port = buffer.read<int32_t>();
}
+
+void verifyRequestForNettyRpcEndpointRef(
+ const RpcRequest& request,
+ const std::string& srcName,
+ const std::string& srcHost,
+ const int srcPort,
+ const std::string& dstHost,
+ const int dstPort) {
+ auto sentBody = request.body();
+ std::string host;
+ int port;
+ readRpcAddress(*sentBody, host, port);
+ EXPECT_EQ(host, srcHost);
+ EXPECT_EQ(port, srcPort);
+ readRpcAddress(*sentBody, host, port);
+ EXPECT_EQ(host, dstHost);
+ EXPECT_EQ(port, dstPort);
+ std::string name;
+ readUTF(*sentBody, name);
+ EXPECT_EQ(name, srcName);
+ EXPECT_EQ(
+ sentBody->read<uint8_t>(),
+ NettyRpcEndpointRef::kNativeTransportMessageFlag);
+}
+
} // namespace
-TEST(NettyRpcEndpointRefTest, askSyncGetReducerFileGroup) {
+TEST(NettyRpcEndpointRefTest, askSyncRegisterShuffle) {
+ auto mockedClient = std::make_shared<MockTransportClient>();
+ const std::string srcName = "test-name";
+ const std::string srcHost = "test-src-host";
+ const int srcPort = 100;
+ const std::string dstHost = "test-dst-host";
+ const int dstPort = 101;
+ const auto conf = std::make_shared<conf::CelebornConf>();
+ auto nettyRpcEndpointRef = NettyRpcEndpointRef(
+ srcName, srcHost, srcPort, dstHost, dstPort, mockedClient, *conf);
+
+ PbRegisterShuffleResponse pbRegisterShuffleResponse;
+ const int status = 5;
+ pbRegisterShuffleResponse.set_status(5);
+ protocol::TransportMessage transportMessage(
+ REGISTER_SHUFFLE_RESPONSE, pbRegisterShuffleResponse.SerializeAsString());
+ auto rpcResponse = makeResponseForNettyRpcEndpointRef(transportMessage, 1000);
+ mockedClient->setResponse(*rpcResponse);
+
+ protocol::RegisterShuffle request{1001};
+ auto response = nettyRpcEndpointRef.askSync<
+ protocol::RegisterShuffle,
+ protocol::RegisterShuffleResponse>(request, MS(100));
+ EXPECT_EQ(response->status, status);
+
+ auto sentRequest = mockedClient->getRequest();
+ verifyRequestForNettyRpcEndpointRef(
+ sentRequest, srcName, srcHost, srcPort, dstHost, dstPort);
+}
+
+TEST(NettyRpcEndpointRefTest, askSyncRevive) {
auto mockedClient = std::make_shared<MockTransportClient>();
const std::string srcName = "test-name";
const std::string srcHost = "test-src-host";
const int srcPort = 100;
const std::string dstHost = "test-dst-host";
const int dstPort = 101;
+ const auto conf = std::make_shared<conf::CelebornConf>();
auto nettyRpcEndpointRef = NettyRpcEndpointRef(
- srcName, srcHost, srcPort, dstHost, dstPort, mockedClient);
+ srcName, srcHost, srcPort, dstHost, dstPort, mockedClient, *conf);
+
+ PbRevive pbRevive;
+ pbRevive.set_shuffleid(5);
+ protocol::TransportMessage transportMessage(
+ CHANGE_LOCATION_RESPONSE, pbRevive.SerializeAsString());
+ auto rpcResponse = makeResponseForNettyRpcEndpointRef(transportMessage, 1000);
+ mockedClient->setResponse(*rpcResponse);
- const std::string responseBody = "test-response-body";
+ protocol::Revive request{1001};
+ auto response =
+ nettyRpcEndpointRef
+ .askSync<protocol::Revive, protocol::ChangeLocationResponse>(
+ request, MS(100));
+
+ auto sentRequest = mockedClient->getRequest();
+ verifyRequestForNettyRpcEndpointRef(
+ sentRequest, srcName, srcHost, srcPort, dstHost, dstPort);
+}
+
+TEST(NettyRpcEndpointRefTest, askSyncMapperEnd) {
+ auto mockedClient = std::make_shared<MockTransportClient>();
+ const std::string srcName = "test-name";
+ const std::string srcHost = "test-src-host";
+ const int srcPort = 100;
+ const std::string dstHost = "test-dst-host";
+ const int dstPort = 101;
+ const auto conf = std::make_shared<conf::CelebornConf>();
+ auto nettyRpcEndpointRef = NettyRpcEndpointRef(
+ srcName, srcHost, srcPort, dstHost, dstPort, mockedClient, *conf);
+
+ PbMapperEndResponse pbMapperEndResponse;
+ const int status = 5;
+ pbMapperEndResponse.set_status(5);
+ protocol::TransportMessage transportMessage(
+ MAPPER_END_RESPONSE, pbMapperEndResponse.SerializeAsString());
+ auto rpcResponse = makeResponseForNettyRpcEndpointRef(transportMessage, 1000);
+ mockedClient->setResponse(*rpcResponse);
+
+ protocol::MapperEnd request{1001};
+ auto response =
+ nettyRpcEndpointRef
+ .askSync<protocol::MapperEnd, protocol::MapperEndResponse>(
+ request, MS(100));
+ EXPECT_EQ(response->status, status);
+
+ auto sentRequest = mockedClient->getRequest();
+ verifyRequestForNettyRpcEndpointRef(
+ sentRequest, srcName, srcHost, srcPort, dstHost, dstPort);
+}
+
+TEST(NettyRpcEndpointRefTest, askSyncGetReducerFileGroup) {
+ auto mockedClient = std::make_shared<MockTransportClient>();
+ const std::string srcName = "test-name";
+ const std::string srcHost = "test-src-host";
+ const int srcPort = 100;
+ const std::string dstHost = "test-dst-host";
+ const int dstPort = 101;
+ const auto conf = std::make_shared<conf::CelebornConf>();
+ auto nettyRpcEndpointRef = NettyRpcEndpointRef(
+ srcName, srcHost, srcPort, dstHost, dstPort, mockedClient, *conf);
- // rpcResponse -> transportMessage -> getReducerFileGroupResponse
PbGetReducerFileGroupResponse pbGetReducerFileGroupResponse;
const int status = 5;
pbGetReducerFileGroupResponse.set_status(5);
protocol::TransportMessage transportMessage(
GET_REDUCER_FILE_GROUP_RESPONSE,
pbGetReducerFileGroupResponse.SerializeAsString());
- auto msgBody = transportMessage.toReadOnlyByteBuffer();
- auto writeBuffer = memory::ByteBuffer::createWriteOnly(sizeof(uint8_t));
- writeBuffer->write<uint8_t>(NettyRpcEndpointRef::kNativeTransportMessageFlag);
- auto flagBody = memory::ByteBuffer::toReadOnly(std::move(writeBuffer));
- auto concatBody = memory::ByteBuffer::concat(*flagBody, *msgBody);
-
- auto rpcResponse = std::make_unique<RpcResponse>(1000, std::move(concatBody));
+ auto rpcResponse = makeResponseForNettyRpcEndpointRef(transportMessage, 1000);
mockedClient->setResponse(*rpcResponse);
protocol::GetReducerFileGroup request{1001};
- auto response = nettyRpcEndpointRef.askSync(request, MS(100));
+ auto response = nettyRpcEndpointRef.askSync<
+ protocol::GetReducerFileGroup,
+ protocol::GetReducerFileGroupResponse>(request, MS(100));
EXPECT_EQ(response->status, status);
auto sentRequest = mockedClient->getRequest();
- auto sentBody = sentRequest.body();
- std::string host;
- int port;
- readRpcAddress(*sentBody, host, port);
- EXPECT_EQ(host, srcHost);
- EXPECT_EQ(port, srcPort);
- readRpcAddress(*sentBody, host, port);
- EXPECT_EQ(host, dstHost);
- EXPECT_EQ(port, dstPort);
- std::string name;
- readUTF(*sentBody, name);
- EXPECT_EQ(name, srcName);
- EXPECT_EQ(
- sentBody->read<uint8_t>(),
- NettyRpcEndpointRef::kNativeTransportMessageFlag);
+ verifyRequestForNettyRpcEndpointRef(
Review Comment:
Could you add some test cases for rpc timeout?
--
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-2157][CIP-14] Support sending RegisterShuffle/Revive/MapperEnd messages for NettyRpcEndpointRef in cppClient [celeborn]
Posted by "HolyLow (via GitHub)" <gi...@apache.org>.
HolyLow commented on PR #3484:
URL: https://github.com/apache/celeborn/pull/3484#issuecomment-3332809477
@SteNicholas Refactored as suggested. Any further suggestion is welcome, thanks a lot.
--
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-2157][CIP-14] Support sending RegisterShuffle/Revive/MapperEnd messages for NettyRpcEndpointRef in cppClient [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3484:
URL: https://github.com/apache/celeborn/pull/3484#issuecomment-3327084996
@HolyLow, thanks for contribution. Generally looks good, just left the minor comment.
--
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-2157][CIP-14] Support sending RegisterShuffle/Revive/MapperEnd messages for NettyRpcEndpointRef in cppClient [celeborn]
Posted by "HolyLow (via GitHub)" <gi...@apache.org>.
HolyLow commented on code in PR #3484:
URL: https://github.com/apache/celeborn/pull/3484#discussion_r2378219952
##########
cpp/celeborn/network/tests/NettyRpcEndpointRefTest.cpp:
##########
@@ -65,54 +77,150 @@ void readRpcAddress(
readUTF(buffer, host);
port = buffer.read<int32_t>();
}
+
+void verifyRequestForNettyRpcEndpointRef(
+ const RpcRequest& request,
+ const std::string& srcName,
+ const std::string& srcHost,
+ const int srcPort,
+ const std::string& dstHost,
+ const int dstPort) {
+ auto sentBody = request.body();
+ std::string host;
+ int port;
+ readRpcAddress(*sentBody, host, port);
+ EXPECT_EQ(host, srcHost);
+ EXPECT_EQ(port, srcPort);
+ readRpcAddress(*sentBody, host, port);
+ EXPECT_EQ(host, dstHost);
+ EXPECT_EQ(port, dstPort);
+ std::string name;
+ readUTF(*sentBody, name);
+ EXPECT_EQ(name, srcName);
+ EXPECT_EQ(
+ sentBody->read<uint8_t>(),
+ NettyRpcEndpointRef::kNativeTransportMessageFlag);
+}
+
} // namespace
-TEST(NettyRpcEndpointRefTest, askSyncGetReducerFileGroup) {
+TEST(NettyRpcEndpointRefTest, askSyncRegisterShuffle) {
+ auto mockedClient = std::make_shared<MockTransportClient>();
+ const std::string srcName = "test-name";
+ const std::string srcHost = "test-src-host";
+ const int srcPort = 100;
+ const std::string dstHost = "test-dst-host";
+ const int dstPort = 101;
+ const auto conf = std::make_shared<conf::CelebornConf>();
+ auto nettyRpcEndpointRef = NettyRpcEndpointRef(
+ srcName, srcHost, srcPort, dstHost, dstPort, mockedClient, *conf);
+
+ PbRegisterShuffleResponse pbRegisterShuffleResponse;
+ const int status = 5;
+ pbRegisterShuffleResponse.set_status(5);
+ protocol::TransportMessage transportMessage(
+ REGISTER_SHUFFLE_RESPONSE, pbRegisterShuffleResponse.SerializeAsString());
+ auto rpcResponse = makeResponseForNettyRpcEndpointRef(transportMessage, 1000);
+ mockedClient->setResponse(*rpcResponse);
+
+ protocol::RegisterShuffle request{1001};
+ auto response = nettyRpcEndpointRef.askSync<
+ protocol::RegisterShuffle,
+ protocol::RegisterShuffleResponse>(request, MS(100));
+ EXPECT_EQ(response->status, status);
+
+ auto sentRequest = mockedClient->getRequest();
+ verifyRequestForNettyRpcEndpointRef(
+ sentRequest, srcName, srcHost, srcPort, dstHost, dstPort);
+}
+
+TEST(NettyRpcEndpointRefTest, askSyncRevive) {
auto mockedClient = std::make_shared<MockTransportClient>();
const std::string srcName = "test-name";
const std::string srcHost = "test-src-host";
const int srcPort = 100;
const std::string dstHost = "test-dst-host";
const int dstPort = 101;
+ const auto conf = std::make_shared<conf::CelebornConf>();
auto nettyRpcEndpointRef = NettyRpcEndpointRef(
- srcName, srcHost, srcPort, dstHost, dstPort, mockedClient);
+ srcName, srcHost, srcPort, dstHost, dstPort, mockedClient, *conf);
+
+ PbRevive pbRevive;
+ pbRevive.set_shuffleid(5);
+ protocol::TransportMessage transportMessage(
+ CHANGE_LOCATION_RESPONSE, pbRevive.SerializeAsString());
+ auto rpcResponse = makeResponseForNettyRpcEndpointRef(transportMessage, 1000);
+ mockedClient->setResponse(*rpcResponse);
- const std::string responseBody = "test-response-body";
+ protocol::Revive request{1001};
+ auto response =
+ nettyRpcEndpointRef
+ .askSync<protocol::Revive, protocol::ChangeLocationResponse>(
+ request, MS(100));
+
+ auto sentRequest = mockedClient->getRequest();
+ verifyRequestForNettyRpcEndpointRef(
+ sentRequest, srcName, srcHost, srcPort, dstHost, dstPort);
+}
+
+TEST(NettyRpcEndpointRefTest, askSyncMapperEnd) {
+ auto mockedClient = std::make_shared<MockTransportClient>();
+ const std::string srcName = "test-name";
+ const std::string srcHost = "test-src-host";
+ const int srcPort = 100;
+ const std::string dstHost = "test-dst-host";
+ const int dstPort = 101;
+ const auto conf = std::make_shared<conf::CelebornConf>();
+ auto nettyRpcEndpointRef = NettyRpcEndpointRef(
+ srcName, srcHost, srcPort, dstHost, dstPort, mockedClient, *conf);
+
+ PbMapperEndResponse pbMapperEndResponse;
+ const int status = 5;
+ pbMapperEndResponse.set_status(5);
+ protocol::TransportMessage transportMessage(
+ MAPPER_END_RESPONSE, pbMapperEndResponse.SerializeAsString());
+ auto rpcResponse = makeResponseForNettyRpcEndpointRef(transportMessage, 1000);
+ mockedClient->setResponse(*rpcResponse);
+
+ protocol::MapperEnd request{1001};
+ auto response =
+ nettyRpcEndpointRef
+ .askSync<protocol::MapperEnd, protocol::MapperEndResponse>(
+ request, MS(100));
+ EXPECT_EQ(response->status, status);
+
+ auto sentRequest = mockedClient->getRequest();
+ verifyRequestForNettyRpcEndpointRef(
+ sentRequest, srcName, srcHost, srcPort, dstHost, dstPort);
+}
+
+TEST(NettyRpcEndpointRefTest, askSyncGetReducerFileGroup) {
+ auto mockedClient = std::make_shared<MockTransportClient>();
+ const std::string srcName = "test-name";
+ const std::string srcHost = "test-src-host";
+ const int srcPort = 100;
+ const std::string dstHost = "test-dst-host";
+ const int dstPort = 101;
+ const auto conf = std::make_shared<conf::CelebornConf>();
+ auto nettyRpcEndpointRef = NettyRpcEndpointRef(
+ srcName, srcHost, srcPort, dstHost, dstPort, mockedClient, *conf);
- // rpcResponse -> transportMessage -> getReducerFileGroupResponse
PbGetReducerFileGroupResponse pbGetReducerFileGroupResponse;
const int status = 5;
pbGetReducerFileGroupResponse.set_status(5);
protocol::TransportMessage transportMessage(
GET_REDUCER_FILE_GROUP_RESPONSE,
pbGetReducerFileGroupResponse.SerializeAsString());
- auto msgBody = transportMessage.toReadOnlyByteBuffer();
- auto writeBuffer = memory::ByteBuffer::createWriteOnly(sizeof(uint8_t));
- writeBuffer->write<uint8_t>(NettyRpcEndpointRef::kNativeTransportMessageFlag);
- auto flagBody = memory::ByteBuffer::toReadOnly(std::move(writeBuffer));
- auto concatBody = memory::ByteBuffer::concat(*flagBody, *msgBody);
-
- auto rpcResponse = std::make_unique<RpcResponse>(1000, std::move(concatBody));
+ auto rpcResponse = makeResponseForNettyRpcEndpointRef(transportMessage, 1000);
mockedClient->setResponse(*rpcResponse);
protocol::GetReducerFileGroup request{1001};
- auto response = nettyRpcEndpointRef.askSync(request, MS(100));
+ auto response = nettyRpcEndpointRef.askSync<
+ protocol::GetReducerFileGroup,
+ protocol::GetReducerFileGroupResponse>(request, MS(100));
EXPECT_EQ(response->status, status);
auto sentRequest = mockedClient->getRequest();
- auto sentBody = sentRequest.body();
- std::string host;
- int port;
- readRpcAddress(*sentBody, host, port);
- EXPECT_EQ(host, srcHost);
- EXPECT_EQ(port, srcPort);
- readRpcAddress(*sentBody, host, port);
- EXPECT_EQ(host, dstHost);
- EXPECT_EQ(port, dstPort);
- std::string name;
- readUTF(*sentBody, name);
- EXPECT_EQ(name, srcName);
- EXPECT_EQ(
- sentBody->read<uint8_t>(),
- NettyRpcEndpointRef::kNativeTransportMessageFlag);
+ verifyRequestForNettyRpcEndpointRef(
Review Comment:
done.
--
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-2157][CIP-14] Support sending RegisterShuffle/Revive/MapperEnd messages for NettyRpcEndpointRef in cppClient [celeborn]
Posted by "HolyLow (via GitHub)" <gi...@apache.org>.
HolyLow commented on PR #3484:
URL: https://github.com/apache/celeborn/pull/3484#issuecomment-3310992492
@SteNicholas @FMX @RexXiong Could you help review this PR? Thanks a lot.
Any suggestion is welcome.
--
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