properly handle TableNotExistException in metadata operations

Author: beryllw

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

@@ -30,6 +30,7 @@
 import org.apache.fluss.exception.PartitionNotExistException;
 import org.apache.fluss.exception.RetriableException;
 import org.apache.fluss.exception.StaleMetadataException;
+import org.apache.fluss.exception.TableNotExistException;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TablePartition;
@@ -287,6 +288,9 @@ public void updateMetadata(
             } else if (t instanceof PartitionNotExistException) {
                 LOG.warn("Failed to update metadata because the partition does not exist", t);
                 throw (PartitionNotExistException) t;
+            } else if (t instanceof TableNotExistException) {
+                LOG.warn("Failed to update metadata because the table does not exist", t);
+                throw (TableNotExistException) t;
             } else {
                 throw new FlussRuntimeException("Failed to update metadata", t);
             }

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

@@ -22,6 +22,7 @@
 import org.apache.fluss.cluster.ServerType;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.StaleMetadataException;
+import org.apache.fluss.exception.TableNotExistException;
 import org.apache.fluss.rpc.RpcClient;
 import org.apache.fluss.rpc.gateway.AdminReadOnlyGateway;
 import org.apache.fluss.rpc.messages.MetadataRequest;
@@ -70,6 +71,24 @@ void testInitializeClusterWithRetries() throws Exception {
                 .hasMessageContaining("The metadata is stale.");
     }
 
+    @Test
+    void testTableNotExistExceptionIsNotWrapped() throws Exception {
+        Configuration configuration = new Configuration();
+        RpcClient rpcClient =
+                RpcClient.create(configuration, TestingClientMetricGroup.newInstance(), false);
+
+        // Gateway that throws TableNotExistException on metadata request
+        AdminReadOnlyGateway gateway = new TestingTableNotExistGateway();
+
+        // TableNotExistException should propagate directly without being wrapped
+        assertThatThrownBy(
+                        () ->
+                                MetadataUpdater.tryToInitializeClusterWithRetries(
+                                        rpcClient, CS_NODE, gateway, 3))
+                .isInstanceOf(TableNotExistException.class)
+                .hasMessageContaining("test_table");
+    }
+
     private static final class TestingAdminReadOnlyGateway extends TestCoordinatorGateway {
 
         private final int maxRetryCount;
@@ -95,4 +114,16 @@ public CompletableFuture<MetadataResponse> metadata(MetadataRequest request) {
             }
         }
     }
+
+    /**
+     * A testing gateway that throws TableNotExistException when metadata is requested. Used to
+     * verify that TableNotExistException is propagated without being wrapped.
+     */
+    private static final class TestingTableNotExistGateway extends TestCoordinatorGateway {
+
+        @Override
+        public CompletableFuture<MetadataResponse> metadata(MetadataRequest request) {
+            throw new TableNotExistException("Table 'test_table' does not exist.");
+        }
+    }
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java

@@ -23,6 +23,7 @@
 import org.apache.fluss.client.table.scanner.ScanRecord;
 import org.apache.fluss.client.table.scanner.log.LogScanner;
 import org.apache.fluss.client.table.scanner.log.ScanRecords;
+import org.apache.fluss.exception.TableNotExistException;
 import org.apache.fluss.flink.source.reader.BoundedSplitReader;
 import org.apache.fluss.flink.source.reader.RecordAndPos;
 import org.apache.fluss.flink.tiering.source.split.TieringLogSplit;
@@ -176,7 +177,7 @@ public RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> fetch() throws I
                 ScanRecords scanRecords;
                 try {
                     scanRecords = currentLogScanner.poll(pollTimeout);
-                } catch (Exception e) {
+                } catch (TableNotExistException e) {
                     // When a table is actually dropped, the log scanner's poll may fail
                     // because metadata update discovers the table no longer exists.
                     if (droppedTables.contains(currentTableId)) {

fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java

@@ -701,6 +701,8 @@ public Map<TablePath, TableInfo> getTables(Collection<TablePath> tablePaths)
                                         .getLakeCatalogContainer()
                                         .getDefaultTableLakeOptions()));
             }
+        } catch (TableNotExistException e) {
+            throw e;
         } catch (Exception e) {
             throw new FlussRuntimeException(
                     String.format("Failed to get tables '%s'.", tablePaths), e);