Re: [PR] [CELEBORN-1147] Added a dedicated API for RPC messages which also accepts an RpcResponseCallback instance [incubator-celeborn]

["rmcyang (via GitHub)" <gi...@apache.org>]

rmcyang commented on code in PR #2123:
URL: https://github.com/apache/incubator-celeborn/pull/2123#discussion_r1412667354


##########
common/src/main/java/org/apache/celeborn/common/network/server/TransportRequestHandler.java:
##########
@@ -73,8 +76,64 @@ public void channelInactive() {
 
   @Override
   public void handle(RequestMessage request) {
+    logger.trace("Received request {} from {}", request.getClass().getName(), reverseClient);
     if (checkRegistered(request)) {
-      msgHandler.receive(reverseClient, request);
+      if (request instanceof RpcRequest) {
+        processRpcRequest((RpcRequest) request);
+      } else if (request instanceof OneWayMessage) {
+        processOneWayMessage((OneWayMessage) request);
+      } else {
+        processOtherMessages(request);
+      }
+    }
+  }
+
+  private void processRpcRequest(final RpcRequest req) {
+    try {
+      logger.trace("Process rpc request {}", req.requestId);
+      msgHandler.receive(
+          reverseClient,
+          req,
+          new RpcResponseCallback() {
+            @Override
+            public void onSuccess(ByteBuffer response) {
+              respond(new RpcResponse(req.requestId, new NioManagedBuffer(response)));
+            }
+
+            @Override
+            public void onFailure(Throwable e) {
+              respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
+            }
+          });
+    } catch (Exception e) {
+      logger.error("Error while invoking handler#receive() on RPC id " + req.requestId, e);
+      respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
+    } finally {
+      req.body().release();
+    }
+  }
+
+  private void processOneWayMessage(OneWayMessage req) {
+    try {
+      logger.trace("Process one way request");
+      msgHandler.receive(reverseClient, req);
+    } catch (Exception e) {
+      logger.error("Error while invoking handler#receive() for one-way message.", e);
+    } finally {
+      req.body().release();
+    }
+  }
+
+  private void processOtherMessages(RequestMessage req) {
+    try {
+      logger.trace("delegating to handler to process other request");
+      msgHandler.receive(reverseClient, req);
+    } catch (Exception e) {
+      logger.error("Error while invoking handler#receive() for other message.", e);
+    } finally {
+      if (req.body() != null) {
+        req.body().release();
+      }

Review Comment:
   The logics in these two methods are almost identical but just the message type they process are different. Can we optimize to have one method to reduce some duplicating code here? I think we can just have one method for both, passing `RequestMessage` as a parameter and than check if the instance is OneWayMessage or not. wdyt?



##########
common/src/main/java/org/apache/celeborn/common/network/server/TransportRequestHandler.java:
##########
@@ -73,8 +76,64 @@ public void channelInactive() {
 
   @Override
   public void handle(RequestMessage request) {
+    logger.trace("Received request {} from {}", request.getClass().getName(), reverseClient);
     if (checkRegistered(request)) {
-      msgHandler.receive(reverseClient, request);
+      if (request instanceof RpcRequest) {
+        processRpcRequest((RpcRequest) request);
+      } else if (request instanceof OneWayMessage) {
+        processOneWayMessage((OneWayMessage) request);
+      } else {
+        processOtherMessages(request);
+      }
+    }
+  }
+
+  private void processRpcRequest(final RpcRequest req) {

Review Comment:
   what's the rationale to declare final for `req` in `processRpcRequest` but not for the same in other two processing methods(`processOneWayMessage` and `processOtherMessages`)?



-- 
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