Merge branch 'refs/heads/main' into flink-support-patition-pushdown

Author: Alibaba-HZY

.github/ISSUE_TEMPLATE/bug.yml

@@ -39,7 +39,8 @@ body:
       description: What Fluss version are you using?
       multiple: false
       options:
-        - "0.6.0 (latest release)"
+        - "0.7.0 (latest release)"
+        - "0.6.0"
         - "0.5.0"
         - "main (development)"
     validations:

.github/ISSUE_TEMPLATE/feature.yml

@@ -17,7 +17,7 @@
 ################################################################################
 
 name: Feature Request 🚀
-description: Add new feature, improvement, and more
+description: User-facing functionality or improvement you’d like to see added
 type: "feature"
 body:
   - type: markdown

.github/ISSUE_TEMPLATE/task.yml

@@ -17,7 +17,7 @@
 ################################################################################
 
 name: Task 📌
-description: A specific piece of work
+description: Specific work item - either part of a larger feature or independent project maintenance
 type: "task"
 body:
   - type: markdown

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(