FMX commented on code in PR #2701:
URL: https://github.com/apache/celeborn/pull/2701#discussion_r1766073115
##########
common/src/main/proto/TransportMessages.proto:
##########
@@ -38,6 +38,8 @@ enum MessageType {
GET_REDUCER_FILE_GROUP_RESPONSE = 15;
UNREGISTER_SHUFFLE = 16;
UNREGISTER_SHUFFLE_RESPONSE = 17;
+ BATCH_UNREGISTER_SHUFFLES = 21;
Review Comment:
Use unused numbers to ensure these RPCs won't be treated as old RPCs.
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -229,6 +231,7 @@ class LifecycleManager(val appUniqueId: String, val conf:
CelebornConf) extends
new Runnable {
override def run(): Unit = Utils.tryLogNonFatalError {
self.send(RemoveExpiredShuffle)
+ // todo
Review Comment:
unnecessary change
##########
master/src/main/proto/Resource.proto:
##########
@@ -22,6 +22,7 @@ package deploy.master.meta;
enum Type {
UnRegisterShuffle = 12;
+ BatchUnRegisterShuffle = 28;
Review Comment:
Place this new type to the end of this section.
##########
master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java:
##########
@@ -65,6 +65,11 @@ public void handleUnRegisterShuffle(String shuffleKey,
String requestId) {
updateUnregisterShuffleMeta(shuffleKey);
}
+ @Override
+ public void batchHandleUnRegisterShuffles(List<String> shuffleKeys, String
requestId) {
+ shuffleKeys.forEach(shuffleKey -> updateUnregisterShuffleMeta(shuffleKey));
Review Comment:
```suggestion
registeredShuffle.removeAll(shuffleKeys);
```
##########
common/src/main/proto/TransportMessages.proto:
##########
@@ -402,10 +404,26 @@ message PbUnregisterShuffle {
string requestId = 3;
}
+message PbBatchUnregisterShuffles {
+ string appId = 1;
+ string requestId = 2;
+ repeated int32 shuffleIds = 3;
+}
+
message PbUnregisterShuffleResponse {
int32 status = 1;
}
+message PbBatchUnregisterShuffleResponses {
+ int32 status = 1;
+ repeated PbUnregisterShuffleResponsesInfo unregisterShuffleResponsesInfo = 2;
+}
+
+message PbUnregisterShuffleResponsesInfo {
Review Comment:
I think this proto is unnecessary. Maybe
```
message PbBatchUnregisterShuffleResponses {
int32 status = 1;
repeated int32 shuffleIds = 2;
}
```
##########
master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java:
##########
@@ -118,6 +118,13 @@ public ResourceResponse
handleWriteRequest(ResourceProtos.ResourceRequest reques
metaSystem.updateUnregisterShuffleMeta(shuffleKey);
break;
+ case BatchUnRegisterShuffle:
+ List<String> shuffleKeys =
+ request.getBatchUnregisterShuffleRequest().getShuffleKeysList();
+ shuffleKeys.forEach(key ->
metaSystem.updateUnregisterShuffleMeta(key));
Review Comment:
```suggestion
metaSystem.registeredShuffle.removeAll(shuffleKeys);
```
##########
common/src/main/proto/TransportMessages.proto:
##########
@@ -402,10 +404,26 @@ message PbUnregisterShuffle {
string requestId = 3;
}
+message PbBatchUnregisterShuffles {
+ string appId = 1;
+ string requestId = 2;
+ repeated int32 shuffleIds = 3;
+}
+
message PbUnregisterShuffleResponse {
int32 status = 1;
}
+message PbBatchUnregisterShuffleResponses {
+ int32 status = 1;
+ repeated PbUnregisterShuffleResponsesInfo unregisterShuffleResponsesInfo = 2;
+}
+
+message PbUnregisterShuffleResponsesInfo {
Review Comment:
If a batch unregistered shuffle RPC fails, retrying it can be enough.
##########
common/src/main/proto/TransportMessages.proto:
##########
@@ -402,10 +404,26 @@ message PbUnregisterShuffle {
string requestId = 3;
}
+message PbBatchUnregisterShuffles {
+ string appId = 1;
+ string requestId = 2;
+ repeated int32 shuffleIds = 3;
+}
+
message PbUnregisterShuffleResponse {
int32 status = 1;
}
+message PbBatchUnregisterShuffleResponses {
+ int32 status = 1;
+ repeated PbUnregisterShuffleResponsesInfo unregisterShuffleResponsesInfo = 2;
+}
+
+message PbUnregisterShuffleResponsesInfo {
Review Comment:
According to your implementation, the shuffleStatus can be ignored because
there is no logic to process the failure of errors that might occur when
removing the shuffle ID.
--
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]
