[server] UpdateMetadata RPC should update tablePath and physicalTablePath cache (#890)

Author: swuferhong

fluss-common/src/main/java/com/alibaba/fluss/cluster/MetadataCache.java

@@ -45,8 +45,6 @@
 import com.alibaba.fluss.rpc.messages.MetadataResponse;
 import com.alibaba.fluss.rpc.messages.TableExistsRequest;
 import com.alibaba.fluss.rpc.messages.TableExistsResponse;
-import com.alibaba.fluss.rpc.messages.UpdateMetadataRequest;
-import com.alibaba.fluss.rpc.messages.UpdateMetadataResponse;
 import com.alibaba.fluss.rpc.protocol.ApiKeys;
 import com.alibaba.fluss.rpc.protocol.RPC;
 
@@ -129,15 +127,6 @@ public interface AdminReadOnlyGateway extends RpcGateway {
     @RPC(api = ApiKeys.GET_METADATA)
     CompletableFuture<MetadataResponse> metadata(MetadataRequest request);
 
-    /**
-     * request send to tablet server to update the metadata cache for every tablet server node,
-     * asynchronously.
-     *
-     * @return the update metadata response
-     */
-    @RPC(api = ApiKeys.UPDATE_METADATA)
-    CompletableFuture<UpdateMetadataResponse> updateMetadata(UpdateMetadataRequest request);
-
     /**
      * Get the latest kv snapshots of a primary key table. A kv snapshot is a snapshot of a kv
      * tablet, so a table can have multiple kv snapshots.

fluss-rpc/src/main/java/com/alibaba/fluss/rpc/gateway/AdminReadOnlyGateway.java

@@ -43,6 +43,8 @@
 import com.alibaba.fluss.rpc.messages.PutKvResponse;
 import com.alibaba.fluss.rpc.messages.StopReplicaRequest;
 import com.alibaba.fluss.rpc.messages.StopReplicaResponse;
+import com.alibaba.fluss.rpc.messages.UpdateMetadataRequest;
+import com.alibaba.fluss.rpc.messages.UpdateMetadataResponse;
 import com.alibaba.fluss.rpc.protocol.ApiKeys;
 import com.alibaba.fluss.rpc.protocol.RPC;
 
@@ -60,6 +62,15 @@ public interface TabletServerGateway extends RpcGateway, AdminReadOnlyGateway {
     CompletableFuture<NotifyLeaderAndIsrResponse> notifyLeaderAndIsr(
             NotifyLeaderAndIsrRequest notifyLeaderAndIsrRequest);
 
+    /**
+     * request send to tablet server to update the metadata cache for every tablet server node,
+     * asynchronously.
+     *
+     * @return the update metadata response
+     */
+    @RPC(api = ApiKeys.UPDATE_METADATA)
+    CompletableFuture<UpdateMetadataResponse> updateMetadata(UpdateMetadataRequest request);
+
     /**
      * Stop replica.
      *

fluss-rpc/src/main/java/com/alibaba/fluss/rpc/gateway/TabletServerGateway.java

@@ -170,6 +170,8 @@ message UpdateMetadataRequest {
   optional PbServerNode coordinator_server = 1;
   repeated PbServerNode tablet_servers = 2;
   repeated PbTableMetadata table_metadata = 3;
+  repeated PbPartitionMetadata partition_metadata = 4;
+  optional int32 coordinator_epoch = 5;
 }
 
 message UpdateMetadataResponse {
@@ -565,6 +567,9 @@ message PbTableMetadata {
   repeated PbBucketMetadata bucket_metadata = 5;
   required int64 created_time = 6;
   required int64 modified_time = 7;
+
+  // TODO add a new filed 'deleted_table' to indicate this table is deleted in UpdateMetadataRequest.
+  // trace by: https://github.com/alibaba/fluss/issues/981
 }
 
 message PbPartitionMetadata {
@@ -581,6 +586,7 @@ message PbBucketMetadata {
   optional int32 leader_id = 2;
   repeated int32 replica_id = 3 [packed = true];
   // TODO: Add isr here.
+  optional int32 leader_epoch = 4;
 }
 
 message PbProduceLogReqForBucket {