[server] Zero-Copy for follower fetching logs from leader (apache#1186)

Author: swuferhong

fluss-client/src/main/java/com/alibaba/fluss/client/FlussConnection.java

@@ -79,7 +79,7 @@ public final class FlussConnection implements Connection {
         String clientId = conf.getString(ConfigOptions.CLIENT_ID);
         this.metricRegistry = metricRegistry;
         this.clientMetricGroup = new ClientMetricGroup(metricRegistry, clientId);
-        this.rpcClient = RpcClient.create(conf, clientMetricGroup);
+        this.rpcClient = RpcClient.create(conf, clientMetricGroup, false);
 
         // TODO this maybe remove after we introduce client metadata.
         this.metadataUpdater = new MetadataUpdater(conf, rpcClient);

fluss-client/src/test/java/com/alibaba/fluss/client/metadata/TestingMetadataUpdater.java

@@ -63,7 +63,8 @@ private TestingMetadataUpdater(
             List<ServerNode> tabletServers,
             Map<TablePath, TableInfo> tableInfos) {
         super(
-                RpcClient.create(new Configuration(), TestingClientMetricGroup.newInstance()),
+                RpcClient.create(
+                        new Configuration(), TestingClientMetricGroup.newInstance(), false),
                 Cluster.empty());
         initializeCluster(coordinatorServer, tabletServers, tableInfos);
         coordinatorGateway = new TestCoordinatorGateway();

fluss-client/src/test/java/com/alibaba/fluss/client/security/acl/FlussAuthorizationITCase.java

@@ -417,7 +417,7 @@ void testGetMetaInfo() throws Exception {
                         Collections.singleton(DATA1_TABLE_PATH_PK), null, null);
 
         try (RpcClient rpcClient =
-                RpcClient.create(guestConf, TestingClientMetricGroup.newInstance())) {
+                RpcClient.create(guestConf, TestingClientMetricGroup.newInstance(), false)) {
             AdminGateway guestGateway =
                     GatewayClientProxy.createGatewayProxy(
                             () -> FLUSS_CLUSTER_EXTENSION.getCoordinatorServerNode("CLIENT"),

fluss-client/src/test/java/com/alibaba/fluss/client/write/RecordAccumulatorTest.java

@@ -630,7 +630,8 @@ private RecordAccumulator createTestRecordAccumulator(
                         conf.getInt(ConfigOptions.CLIENT_WRITER_MAX_INFLIGHT_REQUESTS_PER_BUCKET),
                         GatewayClientProxy.createGatewayProxy(
                                 () -> cluster.getRandomTabletServer(),
-                                RpcClient.create(conf, TestingClientMetricGroup.newInstance()),
+                                RpcClient.create(
+                                        conf, TestingClientMetricGroup.newInstance(), false),
                                 TabletServerGateway.class)),
                 TestingWriterMetricGroup.newInstance(),
                 clock);

fluss-common/src/main/java/com/alibaba/fluss/rpc/messages/ApiMessage.java

@@ -63,6 +63,9 @@ default int sizeExcludingZeroCopy() {
      */
     boolean isLazilyParsed();
 
+    /** Get the parsed byte buffer of the message. */
+    ByteBuf getParsedByteBuf();
+
     /**
      * Deserialize the message from the given byte array.
      *

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitter.java

@@ -55,7 +55,8 @@ public void open() {
         String clientId = flussConf.getString(ConfigOptions.CLIENT_ID);
         MetricRegistry metricRegistry = MetricRegistry.create(flussConf, null);
         // don't care about metrics, but pass a ClientMetricGroup to make compiler happy
-        rpcClient = RpcClient.create(flussConf, new ClientMetricGroup(metricRegistry, clientId));
+        rpcClient =
+                RpcClient.create(flussConf, new ClientMetricGroup(metricRegistry, clientId), false);
         MetadataUpdater metadataUpdater = new MetadataUpdater(flussConf, rpcClient);
         this.coordinatorGateway =
                 GatewayClientProxy.createGatewayProxy(

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java

@@ -126,7 +126,7 @@ public void start() {
         FlinkMetricRegistry metricRegistry = new FlinkMetricRegistry(enumeratorMetricGroup);
         ClientMetricGroup clientMetricGroup =
                 new ClientMetricGroup(metricRegistry, "LakeTieringService");
-        this.rpcClient = RpcClient.create(flussConf, clientMetricGroup);
+        this.rpcClient = RpcClient.create(flussConf, clientMetricGroup, false);
         MetadataUpdater metadataUpdater = new MetadataUpdater(flussConf, rpcClient);
         this.coordinatorGateway =
                 GatewayClientProxy.createGatewayProxy(

fluss-protogen/fluss-protogen-generator/src/main/java/com/alibaba/fluss/protogen/generator/generator/ProtobufMessage.java

@@ -99,6 +99,7 @@ public void generate(PrintWriter w) {
         generateZeroCopySize(w);
         generateParseFrom(w);
         generateIsLazilyParsed(w);
+        generateGetParsedByteBuf(w);
         generateCheckRequiredFields(w);
         generateClear(w);
         generateCopyFrom(w);
@@ -175,6 +176,14 @@ private void generateIsLazilyParsed(PrintWriter w) {
         w.format("        }\n");
     }
 
+    private void generateGetParsedByteBuf(PrintWriter w) {
+        w.println();
+        w.println("       @Override");
+        w.format("        public ByteBuf getParsedByteBuf() {\n");
+        w.format("            return _parsedBuffer;\n");
+        w.format("        }\n");
+    }
+
     private void generateClear(PrintWriter w) {
         w.println();
         w.format("        public %s clear() {\n", message.getName());

fluss-rpc/src/main/java/com/alibaba/fluss/rpc/RpcClient.java

@@ -37,10 +37,13 @@ public interface RpcClient extends AutoCloseable {
      *
      * @param conf The configuration to use.
      * @param clientMetricGroup The client metric group
+     * @param isInnerClient Whether it is an inner client used for communicate from server to
+     *     server.
      * @return The RPC client.
      */
-    static RpcClient create(Configuration conf, ClientMetricGroup clientMetricGroup) {
-        return new NettyClient(conf, clientMetricGroup);
+    static RpcClient create(
+            Configuration conf, ClientMetricGroup clientMetricGroup, boolean isInnerClient) {
+        return new NettyClient(conf, clientMetricGroup, isInnerClient);
     }
 
     /**

fluss-rpc/src/main/java/com/alibaba/fluss/rpc/netty/client/NettyClient.java

@@ -76,9 +76,16 @@ public final class NettyClient implements RpcClient {
 
     private final Supplier<ClientAuthenticator> authenticatorSupplier;
 
+    /**
+     * Whether the NettyClient is used as inner network client (Communicating between Fluss's
+     * servers).
+     */
+    private final boolean isInnerClient;
+
     private volatile boolean isClosed = false;
 
-    public NettyClient(Configuration conf, ClientMetricGroup clientMetricGroup) {
+    public NettyClient(
+            Configuration conf, ClientMetricGroup clientMetricGroup, boolean isInnerClient) {
         this.connections = MapUtils.newConcurrentHashMap();
 
         // build bootstrap
@@ -99,6 +106,7 @@ public NettyClient(Configuration conf, ClientMetricGroup clientMetricGroup) {
                         .option(ChannelOption.TCP_NODELAY, true)
                         .option(ChannelOption.SO_KEEPALIVE, true)
                         .handler(new ClientChannelInitializer(connectionMaxIdle));
+        this.isInnerClient = isInnerClient;
         this.clientMetricGroup = clientMetricGroup;
         this.authenticatorSupplier = AuthenticationFactory.loadClientAuthenticatorSupplier(conf);
         NettyMetrics.registerNettyMetrics(clientMetricGroup, pooledAllocator);
@@ -189,7 +197,8 @@ private ServerConnection getOrCreateConnection(ServerNode node) {
                                     bootstrap,
                                     node,
                                     clientMetricGroup,
-                                    authenticatorSupplier.get());
+                                    authenticatorSupplier.get(),
+                                    isInnerClient);
                     connection.whenClose(ignore -> connections.remove(serverId, connection));
                     return connection;
                 });