[incubator-celeborn] branch main updated: [CELEBORN-320][FLINK] fix handle wrong message type in FetchHandler. (#1254)

[ethanfeng]

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

ethanfeng 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 0df08fbd [CELEBORN-320][FLINK] fix handle wrong message type in 
FetchHandler. (#1254)
0df08fbd is described below

commit 0df08fbdf3dc769c6efbd299cb768f5614f6a245
Author: Ethan Feng <ethanf...@apache.org>
AuthorDate: Tue Feb 21 11:51:01 2023 +0800

    [CELEBORN-320][FLINK] fix handle wrong message type in FetchHandler. (#1254)
---
 .../plugin/flink/readclient/RssBufferStream.java   | 28 +++++++++++-----------
 .../service/deploy/worker/FetchHandler.scala       |  9 ++-----
 2 files changed, 16 insertions(+), 21 deletions(-)

diff --git 
a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
 
b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
index f5155aab..db6559ef 100644
--- 
a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
+++ 
b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
@@ -111,20 +111,20 @@ public class RssBufferStream {
   }
 
   public void addCredit(ReadAddCredit addCredit) {
-    this.client.sendRpc(
-        addCredit.toByteBuffer(),
-        new RpcResponseCallback() {
-          @Override
-          public void onSuccess(ByteBuffer response) {
-            // ignore response
-          }
-
-          @Override
-          public void onFailure(Throwable e) {
-            logger.warn(
-                "Send Add Credit {} failed to {}", addCredit, 
client.getSocketAddress().toString());
-          }
-        });
+    this.client
+        .getChannel()
+        .writeAndFlush(addCredit)
+        .addListener(
+            future -> {
+              if (future.isSuccess()) {
+                // Send ReadAddCredit do not expect response.
+              } else {
+                logger.warn(
+                    "Send ReadAddCredit to {} failed, detail {}",
+                    this.client.getSocketAddress().toString(),
+                    future.cause());
+              }
+            });
   }
 
   public static RssBufferStream empty() {
diff --git 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
index bd045be3..1504b400 100644
--- 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
+++ 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
@@ -81,20 +81,15 @@ class FetchHandler(val conf: TransportConf) extends 
BaseMessageHandler with Logg
         rpcSource.updateMessageMetrics(r, 0)
         handleChunkFetchRequest(client, r)
       case r: RpcRequest =>
-        handleRpcs(client, r)
+        handleOpenStream(client, r)
       case unknown: RequestMessage =>
         throw new IllegalArgumentException(s"Unknown message type id: 
${unknown.`type`.id}")
     }
   }
 
   // here are BackLogAnnouncement,OpenStream and OpenStreamWithCredit RPCs to 
handle
-  def handleRpcs(client: TransportClient, request: RpcRequest): Unit = {
+  def handleOpenStream(client: TransportClient, request: RpcRequest): Unit = {
     val msg = Message.decode(request.body().nioByteBuffer())
-    if (msg.`type`() == Type.BACKLOG_ANNOUNCEMENT) {
-      rpcSource.updateMessageMetrics(msg, 0)
-      handleReadAddCredit(client, msg.asInstanceOf[ReadAddCredit])
-      return
-    }
     val (shuffleKey, fileName) =
       if (msg.`type`() == Type.OPEN_STREAM) {
         val openStream = msg.asInstanceOf[OpenStream]