[incubator-celeborn] branch main updated: [CELEBORN-781] Refactor RPC message type name

[an...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

angerszhuuuu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git


The following commit(s) were added to refs/heads/main by this push:
     new 1642090f9 [CELEBORN-781] Refactor RPC message type name
1642090f9 is described below

commit 1642090f9f06e566e51d39c91ae6ce0d443b5fff
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Wed Jul 12 14:16:19 2023 +0800

    [CELEBORN-781] Refactor RPC message type name
    
    ### What changes were proposed in this pull request?
    After https://github.com/apache/incubator-celeborn/pull/1658 merged, we can format the message type now.
    
    ### Why are the changes needed?
    
    ### Does this PR introduce _any_ user-facing change?
    
    ### How was this patch tested?
    
    Closes #1696 from AngersZhuuuu/CELEBORN-731.
    
    Authored-by: Angerszhuuuu <an...@gmail.com>
    Signed-off-by: Angerszhuuuu <an...@gmail.com>
---
 common/src/main/proto/TransportMessages.proto            |  8 ++++----
 .../common/protocol/message/ControlMessages.scala        | 16 ++++++++--------
 2 files changed, 12 insertions(+), 12 deletions(-)

diff --git a/common/src/main/proto/TransportMessages.proto b/common/src/main/proto/TransportMessages.proto
index b7f26d268..0d9b67bcb 100644
--- a/common/src/main/proto/TransportMessages.proto
+++ b/common/src/main/proto/TransportMessages.proto
@@ -21,14 +21,14 @@ enum MessageType {
   UNKNOWN_MESSAGE = 0;
   REGISTER_WORKER = 1;
   HEARTBEAT_FROM_WORKER = 2;
-  HEARTBEAT_RESPONSE = 3;
+  HEARTBEAT_FROM_WORKER_RESPONSE = 3;
   REGISTER_SHUFFLE = 4;
   REGISTER_SHUFFLE_RESPONSE = 5;
   REQUEST_SLOTS = 6;
   RELEASE_SLOTS = 7;
   RELEASE_SLOTS_RESPONSE = 8;
   REQUEST_SLOTS_RESPONSE = 9;
-  REVIVE = 10;
+  CHANGE_LOCATION = 10;
   CHANGE_LOCATION_RESPONSE = 11;
   MAPPER_END = 12;
   MAPPER_END_RESPONSE = 13;
@@ -59,8 +59,8 @@ enum MessageType {
   // THREAD_DUMP_RESPONSE = 38;
   REMOVE_EXPIRED_SHUFFLE = 39;
   ONE_WAY_MESSAGE_RESPONSE = 40;
-  CHECK_FOR_WORKER_TIMEOUT = 41;
-  CHECK_FOR_APPLICATION_TIMEOUT = 42;
+  CHECK_WORKER_TIMEOUT = 41;
+  CHECK_APPLICATION_TIMEOUT = 42;
   WORKER_LOST = 43;
   WORKER_LOST_RESPONSE = 44;
   STAGE_END = 45;
diff --git a/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala b/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
index 690e785b5..1be450005 100644
--- a/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
@@ -416,10 +416,10 @@ object ControlMessages extends Logging {
   // TODO change message type to GeneratedMessageV3
   def toTransportMessage(message: Any): TransportMessage = message match {
     case _: PbCheckForWorkerTimeoutOrBuilder =>
-      new TransportMessage(MessageType.CHECK_FOR_WORKER_TIMEOUT, null)
+      new TransportMessage(MessageType.CHECK_WORKER_TIMEOUT, null)
 
     case CheckForApplicationTimeOut =>
-      new TransportMessage(MessageType.CHECK_FOR_APPLICATION_TIMEOUT, null)
+      new TransportMessage(MessageType.CHECK_APPLICATION_TIMEOUT, null)
 
     case CheckForHDFSExpiredDirsTimeout =>
       new TransportMessage(MessageType.CHECK_FOR_HDFS_EXPIRED_DIRS_TIMEOUT, null)
@@ -463,7 +463,7 @@ object ControlMessages extends Logging {
         .addAllExpiredShuffleKeys(expiredShuffleKeys)
         .setRegistered(registered)
         .build().toByteArray
-      new TransportMessage(MessageType.HEARTBEAT_RESPONSE, payload)
+      new TransportMessage(MessageType.HEARTBEAT_FROM_WORKER_RESPONSE, payload)
 
     case pb: PbRegisterShuffle =>
       new TransportMessage(MessageType.REGISTER_SHUFFLE, pb.toByteArray)
@@ -523,7 +523,7 @@ object ControlMessages extends Logging {
       new TransportMessage(MessageType.REQUEST_SLOTS_RESPONSE, payload)
 
     case pb: PbRevive =>
-      new TransportMessage(MessageType.REVIVE, pb.toByteArray)
+      new TransportMessage(MessageType.CHANGE_LOCATION, pb.toByteArray)
 
     case pb: PbChangeLocationResponse =>
       new TransportMessage(MessageType.CHANGE_LOCATION_RESPONSE, pb.toByteArray)
@@ -799,7 +799,7 @@ object ControlMessages extends Logging {
           estimatedAppDiskUsage,
           pbHeartbeatFromWorker.getRequestId)
 
-      case HEARTBEAT_RESPONSE_VALUE =>
+      case HEARTBEAT_FROM_WORKER_RESPONSE_VALUE =>
         val pbHeartbeatFromWorkerResponse =
           PbHeartbeatFromWorkerResponse.parseFrom(message.getPayload)
         val expiredShuffleKeys = new util.HashSet[String]()
@@ -852,7 +852,7 @@ object ControlMessages extends Logging {
           PbSerDeUtils.fromPbWorkerResource(
             pbRequestSlotsResponse.getWorkerResourceMap))
 
-      case REVIVE_VALUE =>
+      case CHANGE_LOCATION_VALUE =>
         PbRevive.parseFrom(message.getPayload)
 
       case CHANGE_LOCATION_RESPONSE_VALUE =>
@@ -1031,10 +1031,10 @@ object ControlMessages extends Logging {
       case ONE_WAY_MESSAGE_RESPONSE_VALUE =>
         OneWayMessageResponse
 
-      case CHECK_FOR_WORKER_TIMEOUT_VALUE =>
+      case CHECK_WORKER_TIMEOUT_VALUE =>
         pbCheckForWorkerTimeout
 
-      case CHECK_FOR_APPLICATION_TIMEOUT_VALUE =>
+      case CHECK_APPLICATION_TIMEOUT_VALUE =>
         CheckForApplicationTimeOut
 
       case CHECK_FOR_HDFS_EXPIRED_DIRS_TIMEOUT_VALUE =>