This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 3b0afc0eb [server] Fix tableCount metric after dropping partitioned
tables (#3482)
3b0afc0eb is described below
commit 3b0afc0eb0e839e6873a400fa3db49a0697eae1a
Author: yunhong <[email protected]>
AuthorDate: Sun Jun 14 09:03:14 2026 +0800
[server] Fix tableCount metric after dropping partitioned tables (#3482)
---
.../coordinator/CoordinatorEventProcessor.java | 11 ++++++++++
.../coordinator/event/CoordinatorEventManager.java | 15 ++++++++++++-
.../coordinator/CoordinatorEventProcessorTest.java | 25 ++++++++++++++++++++++
3 files changed, 50 insertions(+), 1 deletion(-)
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
index c2e297f19..0a93ed54b 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -959,6 +959,17 @@ public class CoordinatorEventProcessor implements
EventProcessor {
// 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) {
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java
index 33df6f484..3bf64f1ef 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java
@@ -32,6 +32,7 @@ import org.apache.fluss.utils.concurrent.ShutdownableThread;
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 @@ public final class CoordinatorEventManager implements
EventManager {
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();
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
index 8f27cdd10..001a90f74 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
@@ -891,6 +891,31 @@ class CoordinatorEventProcessorTest {
});
}
+ @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");
