[Client] Fix LogFetcher doesn't update bucket metadata when receive NotLeaderOrFollowerException in bucket level (#1885)

Author: swuferhong

fluss-client/src/main/java/org/apache/fluss/client/metadata/ClientSchemaGetter.java

@@ -42,7 +42,7 @@ public class ClientSchemaGetter implements SchemaGetter {
     private static final Logger LOG = LoggerFactory.getLogger(ClientSchemaGetter.class);
 
     private final TablePath tablePath;
-    private final Map<Integer, Schema> schemasById;
+    protected final Map<Integer, Schema> schemasById;
     private final Admin admin;
     private volatile SchemaInfo latestSchemaInfo;
 

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java

@@ -26,8 +26,10 @@
 import org.apache.fluss.cluster.BucketLocation;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.ApiException;
 import org.apache.fluss.exception.InvalidMetadataException;
 import org.apache.fluss.exception.LeaderNotAvailableException;
+import org.apache.fluss.exception.PartitionNotExistException;
 import org.apache.fluss.fs.FsPath;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.SchemaGetter;
@@ -48,6 +50,7 @@
 import org.apache.fluss.rpc.messages.PbFetchLogReqForTable;
 import org.apache.fluss.rpc.messages.PbFetchLogRespForBucket;
 import org.apache.fluss.rpc.messages.PbFetchLogRespForTable;
+import org.apache.fluss.rpc.protocol.ApiError;
 import org.apache.fluss.rpc.protocol.Errors;
 import org.apache.fluss.utils.IOUtils;
 import org.apache.fluss.utils.Projection;
@@ -217,14 +220,27 @@ private void checkAndUpdateMetadata(List<TableBucket> tableBuckets) {
             }
         }
 
-        if (isPartitioned && !partitionIds.isEmpty()) {
-            metadataUpdater.updateMetadata(Collections.singleton(tablePath), null, partitionIds);
-        } else if (needUpdate) {
-            metadataUpdater.updateTableOrPartitionMetadata(tablePath, null);
+        try {
+            if (isPartitioned && !partitionIds.isEmpty()) {
+                metadataUpdater.updateMetadata(
+                        Collections.singleton(tablePath), null, partitionIds);
+            } else if (needUpdate) {
+                metadataUpdater.updateTableOrPartitionMetadata(tablePath, null);
+            }
+        } catch (Exception e) {
+            if (e instanceof PartitionNotExistException) {
+                // ignore this exception, this is probably happen because the partition is deleted.
+                // The fetcher can also work fine. The caller like flink can remove the partition
+                // from fetch list when receive exception.
+                LOG.warn("Receive PartitionNotExistException when update metadata, ignore it", e);
+            } else {
+                throw e;
+            }
         }
     }
 
-    private void sendFetchRequest(int destination, FetchLogRequest fetchLogRequest) {
+    @VisibleForTesting
+    void sendFetchRequest(int destination, FetchLogRequest fetchLogRequest) {
         TableOrPartitions tableOrPartitionsInFetchRequest =
                 getTableOrPartitionsInFetchRequest(fetchLogRequest);
         // TODO cache the tablet server gateway.
@@ -345,6 +361,14 @@ private synchronized void handleFetchLogResponse(
                                     respForBucket.getBucketId());
                     FetchLogResultForBucket fetchResultForBucket =
                             getFetchLogResultForBucket(tb, tablePath, respForBucket);
+
+                    // if error code is not NONE, it means the fetch log request failed, we need to
+                    // clear table bucket meta for InvalidMetadataException.
+                    if (fetchResultForBucket.getErrorCode() != Errors.NONE.code()) {
+                        ApiError error = ApiError.fromErrorMessage(respForBucket);
+                        handleFetchLogExceptionForBucket(tb, destination, error);
+                    }
+
                     Long fetchOffset = logScannerStatus.getBucketOffset(tb);
                     // if the offset is null, it means the bucket has been unsubscribed,
                     // we just set a Long.MAX_VALUE as the next fetch offset
@@ -387,6 +411,29 @@ private synchronized void handleFetchLogResponse(
         }
     }
 
+    private void handleFetchLogExceptionForBucket(TableBucket tb, int destination, ApiError error) {
+        ApiException exception = error.error().exception();
+        LOG.error("Failed to fetch log from node {} for bucket {}", destination, tb, exception);
+        if (exception instanceof InvalidMetadataException) {
+            LOG.warn(
+                    "Invalid metadata error in fetch log request. "
+                            + "Going to request metadata update.",
+                    exception);
+            long tableId = tb.getTableId();
+            TableOrPartitions tableOrPartitions;
+            if (tb.getPartitionId() == null) {
+                tableOrPartitions = new TableOrPartitions(Collections.singleton(tableId), null);
+            } else {
+                tableOrPartitions =
+                        new TableOrPartitions(
+                                null,
+                                Collections.singleton(
+                                        new TablePartition(tableId, tb.getPartitionId())));
+            }
+            invalidTableOrPartitions(tableOrPartitions);
+        }
+    }
+
     private void pendRemoteFetches(
             RemoteLogFetchInfo remoteLogFetchInfo, long firstFetchOffset, long highWatermark) {
         checkNotNull(remoteLogFetchInfo);
@@ -417,7 +464,8 @@ private void pendRemoteFetches(
         }
     }
 
-    private Map<Integer, FetchLogRequest> prepareFetchLogRequests() {
+    @VisibleForTesting
+    Map<Integer, FetchLogRequest> prepareFetchLogRequests() {
         Map<Integer, List<PbFetchLogReqForBucket>> fetchLogReqForBuckets = new HashMap<>();
         int readyForFetchCount = 0;
         Long tableId = null;

fluss-client/src/test/java/org/apache/fluss/client/metadata/TestingClientSchemaGetter.java

@@ -0,0 +1,55 @@
+/*
+ * 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.fluss.client.metadata;
+
+import org.apache.fluss.client.admin.FlussAdmin;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.SchemaInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.rpc.RpcClient;
+import org.apache.fluss.rpc.metrics.TestingClientMetricGroup;
+
+import java.util.concurrent.CompletableFuture;
+
+/** Testing class for {@link ClientSchemaGetter}. */
+public class TestingClientSchemaGetter extends ClientSchemaGetter {
+    public TestingClientSchemaGetter(
+            TablePath tablePath,
+            SchemaInfo latestSchemaInfo,
+            TestingMetadataUpdater metadataUpdater) {
+        super(
+                tablePath,
+                latestSchemaInfo,
+                new FlussAdmin(
+                        RpcClient.create(
+                                new Configuration(), TestingClientMetricGroup.newInstance(), false),
+                        metadataUpdater));
+    }
+
+    @Override
+    public Schema getSchema(int schemaId) {
+        return schemasById.get(schemaId);
+    }
+
+    @Override
+    public CompletableFuture<SchemaInfo> getSchemaInfoAsync(int schemaId) {
+        return CompletableFuture.completedFuture(
+                new SchemaInfo(schemasById.get(schemaId), schemaId));
+    }
+}

fluss-client/src/test/java/org/apache/fluss/client/metadata/TestingMetadataUpdater.java

@@ -42,13 +42,13 @@
 
 /** Testing class for metadata updater. */
 public class TestingMetadataUpdater extends MetadataUpdater {
-    private static final ServerNode COORDINATOR =
+    public static final ServerNode COORDINATOR =
             new ServerNode(0, "localhost", 90, ServerType.COORDINATOR);
-    private static final ServerNode NODE1 =
+    public static final ServerNode NODE1 =
             new ServerNode(1, "localhost", 90, ServerType.TABLET_SERVER, "rack1");
-    private static final ServerNode NODE2 =
+    public static final ServerNode NODE2 =
             new ServerNode(2, "localhost", 91, ServerType.TABLET_SERVER, "rack2");
-    private static final ServerNode NODE3 =
+    public static final ServerNode NODE3 =
             new ServerNode(3, "localhost", 92, ServerType.TABLET_SERVER, "rack3");
 
     private final TestCoordinatorGateway coordinatorGateway;
@@ -63,7 +63,7 @@ public TestingMetadataUpdater(Map<TablePath, TableInfo> tableInfos) {
                 new Configuration());
     }
 
-    private TestingMetadataUpdater(
+    public TestingMetadataUpdater(
             ServerNode coordinatorServer,
             List<ServerNode> tabletServers,
             Map<TablePath, TableInfo> tableInfos,
@@ -137,10 +137,6 @@ public void updateCluster(Cluster cluster) {
         this.cluster = cluster;
     }
 
-    public void setResponseLogicId(int serverId, int responseLogicId) {
-        tabletServerGatewayMap.get(serverId).setResponseLogicId(responseLogicId);
-    }
-
     @Override
     public void checkAndUpdateTableMetadata(Set<TablePath> tablePaths) {
         Set<TablePath> needUpdateTablePaths =