[server] Fix tableCount metric after dropping partitioned tables (#3482)

Author: swuferhong

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

@@ -959,6 +959,17 @@ private void processDropTable(DropTableEvent dropTableEvent) {
 
         // remove table metrics.
         coordinatorMetricGroup.removeTableMetricGroup(dropTableInfo.getTablePath(), tableId);
+
+        // For partitioned tables, the dropped table has no table-level replicas
+        // (all buckets live under partitionAssignments), so getAllReplicasForTable
+        // returns empty and areAllReplicasInState(.., ReplicaDeletionSuccessful)
+        // is vacuously true. Without this explicit trigger, the table would linger
+        // in tablesToBeDeleted (and the tableCount metric) until some unrelated
+        // replica-deletion response happens to invoke resumeDeletions(), which
+        // can be never if no other table is being dropped.
+        if (dropTableInfo.isPartitioned()) {
+            tableManager.resumeDeletions();
+        }
     }
 
     private void processDropPartition(DropPartitionEvent dropPartitionEvent) {

fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java

@@ -32,6 +32,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Set;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
@@ -119,7 +120,19 @@ private void updateMetricsViaAccessContext() {
                         context -> {
                             int coordinatorServerCount = context.getLiveCoordinatorServers().size();
                             int tabletServerCount = context.getLiveTabletServers().size();
-                            int tableCount = context.allTables().size();
+                            // Exclude tables that have been queued for deletion (DropTable RPC
+                            // already acked, but completeDeleteTable has not yet removed them
+                            // from tablePathById). We dedup by tableId rather than subtracting
+                            // sizes so the result is robust even if tablesToBeDeleted ever
+                            // drifts out of sync with tablePathById -- it can never go negative
+                            // and only counts ids that are truly still in tablePathById.
+                            Set<Long> allTables = context.allTables().keySet();
+                            int tableCount = allTables.size();
+                            for (Long toDelete : context.getTablesToBeDeleted()) {
+                                if (allTables.contains(toDelete)) {
+                                    tableCount--;
+                                }
+                            }
                             int lakeTableCount = context.getLakeTableCount();
                             int bucketCount = context.bucketLeaderAndIsr().size();
                             int partitionCount = context.getTotalPartitionCount();

fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java

@@ -891,6 +891,31 @@ void testDropPartitionedTableRoutedThroughCleanupManager() throws Exception {
                 });
     }
 
+    @Test
+    void testDropPartitionedTableWithNoPartitionsClearsTableState() throws Exception {
+        TablePath tablePath =
+                TablePath.of(defaultDatabase, "test_drop_partitioned_table_no_partitions");
+        initCoordinatorChannel();
+
+        long tableId =
+                metadataManager.createTable(
+                        tablePath, remoteDataDir, getPartitionedTable(), null, false);
+        retryVerifyContext(ctx -> assertThat(ctx.getTablePathById(tableId)).isNotNull());
+
+        metadataManager.dropTable(tablePath, false);
+
+        // The fix at the tail of processDropTable explicitly calls resumeDeletions(), which
+        // sees vacuous-true (getAllReplicasForTable returns empty) and runs
+        // completeDeleteTable -> removeTable, evicting the table from both tablesToBeDeleted
+        // and tablePathById.
+        retryVerifyContext(
+                ctx -> {
+                    assertThat(ctx.getTablePathById(tableId)).isNull();
+                    assertThat(ctx.getTablesToBeDeleted()).doesNotContain(tableId);
+                    assertThat(ctx.allTables()).doesNotContainKey(tableId);
+                });
+    }
+
     @Test
     void testStartupResumesDropPartitionThroughCleanupManager() throws Exception {
         TablePath tablePath = TablePath.of(defaultDatabase, "test_startup_resume_via_cleanup");