[incubator-celeborn] branch main updated: [CELEBORN-282] Add BacklogAnnouncement RPC. (#1217)

[zhouky]

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

zhouky 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 a364fb27 [CELEBORN-282] Add BacklogAnnouncement RPC. (#1217)
a364fb27 is described below

commit a364fb27b2f3865a984ffd0fe900f1a28a9fce2e
Author: Ethan Feng <ethanf...@apache.org>
AuthorDate: Thu Feb 16 14:58:39 2023 +0800

    [CELEBORN-282] Add BacklogAnnouncement RPC. (#1217)
---
 .../network/protocol/BacklogAnnouncement.java      | 63 ++++++++++++++++++++++
 .../celeborn/common/network/protocol/Message.java  |  8 ++-
 2 files changed, 70 insertions(+), 1 deletion(-)

diff --git 
a/common/src/main/java/org/apache/celeborn/common/network/protocol/BacklogAnnouncement.java
 
b/common/src/main/java/org/apache/celeborn/common/network/protocol/BacklogAnnouncement.java
new file mode 100644
index 00000000..45f02f5d
--- /dev/null
+++ 
b/common/src/main/java/org/apache/celeborn/common/network/protocol/BacklogAnnouncement.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.common.network.protocol;
+
+import static 
org.apache.celeborn.common.network.protocol.Message.Type.BACKLOG_ANNOUNCEMENT;
+
+import io.netty.buffer.ByteBuf;
+
+// This RPC is sent to flink plugin to tell flink client to be ready for 
buffers.
+public class BacklogAnnouncement extends RequestMessage {
+  private long streamId;
+  private int backlog;
+
+  public BacklogAnnouncement(long streamId, int backlog) {
+    this.streamId = streamId;
+    this.backlog = backlog;
+  }
+
+  @Override
+  public int encodedLength() {
+    return 12;
+  }
+
+  @Override
+  public void encode(ByteBuf buf) {
+    buf.writeLong(streamId);
+    buf.writeInt(backlog);
+  }
+
+  public static BacklogAnnouncement decode(ByteBuf in) {
+    long streamId = in.readLong();
+    int backlog = in.readInt();
+    return new BacklogAnnouncement(streamId, backlog);
+  }
+
+  @Override
+  public Type type() {
+    return BACKLOG_ANNOUNCEMENT;
+  }
+
+  public long getStreamId() {
+    return streamId;
+  }
+
+  public int getBacklog() {
+    return backlog;
+  }
+}
diff --git 
a/common/src/main/java/org/apache/celeborn/common/network/protocol/Message.java 
b/common/src/main/java/org/apache/celeborn/common/network/protocol/Message.java
index c82feb6d..a361c973 100644
--- 
a/common/src/main/java/org/apache/celeborn/common/network/protocol/Message.java
+++ 
b/common/src/main/java/org/apache/celeborn/common/network/protocol/Message.java
@@ -87,7 +87,8 @@ public abstract class Message implements Encodable {
     PUSH_DATA_HAND_SHAKE(15),
     READ_ADD_CREDIT(16),
     READ_DATA(17),
-    OPEN_STREAM_WITH_CREDIT(18);
+    OPEN_STREAM_WITH_CREDIT(18),
+    BACKLOG_ANNOUNCEMENT(19);
 
     private final byte id;
 
@@ -147,6 +148,8 @@ public abstract class Message implements Encodable {
           return READ_DATA;
         case 18:
           return OPEN_STREAM_WITH_CREDIT;
+        case 19:
+          return BACKLOG_ANNOUNCEMENT;
         case -1:
           throw new IllegalArgumentException("User type messages cannot be 
decoded.");
         default:
@@ -212,6 +215,9 @@ public abstract class Message implements Encodable {
       case OPEN_STREAM_WITH_CREDIT:
         return OpenStreamWithCredit.decode(in);
 
+      case BACKLOG_ANNOUNCEMENT:
+        return BacklogAnnouncement.decode(in);
+
       default:
         throw new IllegalArgumentException("Unexpected message type: " + 
msgType);
     }