diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java index e3d602d991..bdafe84617 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java @@ -122,14 +122,6 @@ private long physicalStorageRemoteKvSize() { .reduce(0L, Long::sum); } - private long getNumSnapshots(TableBucket tableBucket) { - return bucketCompletedSnapshotStores.get(tableBucket).getNumSnapshots(); - } - - private long getAllSnapshotSize(TableBucket tableBucket) { - return bucketCompletedSnapshotStores.get(tableBucket).getPhysicalStorageRemoteKvSize(); - } - public CompletedSnapshotStore getOrCreateCompletedSnapshotStore( TablePath tablePath, TableBucket tableBucket) { return bucketCompletedSnapshotStores.computeIfAbsent( @@ -152,10 +144,10 @@ public CompletedSnapshotStore getOrCreateCompletedSnapshotStore( if (bucketMetricGroup != null) { LOG.info("Add bucketMetricGroup for tableBucket {}.", bucket); bucketMetricGroup.gauge( - MetricNames.KV_NUM_SNAPSHOTS, () -> getNumSnapshots(bucket)); + MetricNames.KV_NUM_SNAPSHOTS, snapshotStore::getNumSnapshots); bucketMetricGroup.gauge( MetricNames.KV_ALL_SNAPSHOT_SIZE, - () -> getAllSnapshotSize(bucket)); + snapshotStore::getPhysicalStorageRemoteKvSize); } else { LOG.warn( "Failed to add bucketMetricGroup for tableBucket {} when creating completed snapshot.", 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 96220fee12..c54a2fb169 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 @@ -1068,9 +1068,13 @@ private void processCreatePartition(CreatePartitionEvent createPartitionEvent) { } private void processDropTable(DropTableEvent dropTableEvent) { - // If this is a primary key table, drop the kv snapshot store. long tableId = dropTableEvent.getTableId(); TableInfo dropTableInfo = coordinatorContext.getTableInfoById(tableId); + + // Remove table metrics before dropping their backing snapshot stores. + coordinatorMetricGroup.removeTableMetricGroup(dropTableInfo.getTablePath(), tableId); + + // If this is a primary key table, drop the kv snapshot store. if (dropTableInfo.hasPrimaryKey()) { Set deleteTableBuckets = coordinatorContext.getAllBucketsForTable(tableId); completedSnapshotStoreManager.removeCompletedSnapshotStoreByTableBuckets( @@ -1093,9 +1097,6 @@ private void processDropTable(DropTableEvent dropTableEvent) { null, Collections.emptySet()); - // 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) @@ -1115,6 +1116,11 @@ private void processDropPartition(DropPartitionEvent dropPartitionEvent) { // If this is a primary key table partition, drop the kv snapshot store. TableInfo dropTableInfo = coordinatorContext.getTableInfoById(tableId); + + // Remove partition metrics before dropping their backing snapshot stores. + coordinatorMetricGroup.removeTablePartitionMetricsGroup( + dropTableInfo.getTablePath(), tableId, tablePartition.getPartitionId()); + if (dropTableInfo.hasPrimaryKey()) { Set deleteTableBuckets = coordinatorContext.getAllBucketsForPartition( @@ -1133,10 +1139,6 @@ private void processDropPartition(DropPartitionEvent dropPartitionEvent) { tableId, tablePartition.getPartitionId(), Collections.emptySet()); - - // remove partition metrics. - coordinatorMetricGroup.removeTablePartitionMetricsGroup( - dropTableInfo.getTablePath(), tableId, tablePartition.getPartitionId()); } private void processDeleteReplicaResponseReceived( 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 0bbb40b5c8..2c42dc54c2 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 @@ -36,6 +36,9 @@ import org.apache.fluss.metadata.TableInfo; import org.apache.fluss.metadata.TablePartition; import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.metrics.Gauge; +import org.apache.fluss.metrics.MetricNames; +import org.apache.fluss.metrics.groups.AbstractMetricGroup; import org.apache.fluss.rpc.gateway.TabletServerGateway; import org.apache.fluss.rpc.messages.AdjustIsrResponse; import org.apache.fluss.rpc.messages.ApiMessage; @@ -46,6 +49,7 @@ import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrResponse; import org.apache.fluss.rpc.messages.NotifyRemoteLogOffsetsRequest; import org.apache.fluss.rpc.messages.UpdateMetadataRequest; +import org.apache.fluss.rpc.messages.UpdateMetadataResponse; import org.apache.fluss.rpc.protocol.ApiError; import org.apache.fluss.rpc.protocol.ApiKeys; import org.apache.fluss.rpc.protocol.Errors; @@ -106,6 +110,7 @@ import java.nio.file.Path; import java.time.Duration; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; @@ -115,9 +120,11 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; import java.util.function.Consumer; import java.util.function.Function; import java.util.stream.Collectors; @@ -138,6 +145,7 @@ import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.apache.fluss.testutils.common.CommonTestUtils.waitValue; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test for {@link CoordinatorEventProcessor}. */ @@ -785,6 +793,78 @@ void testAddBucketCompletedSnapshot(@TempDir Path tempDir) throws Exception { .isInstanceOf(InvalidCoordinatorException.class); } + @Test + void testMetricsRemainCollectableAfterDropPartitionFailure() throws Exception { + eventProcessor.shutdown(); + + FailingUpdateMetadataChannelManager failingChannelManager = + new FailingUpdateMetadataChannelManager(); + testCoordinatorChannelManager = failingChannelManager; + eventProcessor = buildCoordinatorEventProcessor(); + eventProcessor.startup(); + completedSnapshotStoreManager = eventProcessor.completedSnapshotStoreManager(); + + TablePath tablePath = TablePath.of(defaultDatabase, "test_metrics_during_drop_partition"); + initCoordinatorChannel(); + long tableId = + metadataManager.createTable( + tablePath, remoteDataDir, getPartitionedTable(), null, false); + Map assignments = + generateAssignment( + N_BUCKETS, + REPLICATION_FACTOR, + new TabletServerInfo[] { + new TabletServerInfo(0, "rack0"), + new TabletServerInfo(1, "rack1"), + new TabletServerInfo(2, "rack2") + }) + .getBucketAssignments(); + PartitionAssignment partitionAssignment = new PartitionAssignment(tableId, assignments); + PartitionIdName partition = + preparePartitionAssignment(tablePath, tableId, partitionAssignment).f0; + verifyPartitionCreated( + new TablePartition(tableId, partition.partitionId), + partitionAssignment, + N_BUCKETS, + REPLICATION_FACTOR); + + List tableBuckets = allTableBuckets(tableId, partition.partitionId, N_BUCKETS); + List bucketMetricGroups = new ArrayList<>(); + List> inFlightGauges = new ArrayList<>(); + for (TableBucket tableBucket : tableBuckets) { + completedSnapshotStoreManager.getOrCreateCompletedSnapshotStore(tablePath, tableBucket); + AbstractMetricGroup bucketMetricGroup = + (AbstractMetricGroup) + TestingMetricGroups.COORDINATOR_METRICS.getTableBucketMetricGroup( + tablePath, tableBucket); + assertThat(bucketMetricGroup).isNotNull(); + bucketMetricGroups.add(bucketMetricGroup); + inFlightGauges.add( + (Gauge) bucketMetricGroup.getMetrics().get(MetricNames.KV_NUM_SNAPSHOTS)); + inFlightGauges.add( + (Gauge) + bucketMetricGroup.getMetrics().get(MetricNames.KV_ALL_SNAPSHOT_SIZE)); + } + assertThat(inFlightGauges).doesNotContainNull(); + + failingChannelManager.failNextUpdateMetadata(); + zookeeperClient.deletePartition(tablePath, partition.partitionName); + failingChannelManager.awaitFailure(); + fromCtx(context -> null); + + assertThat(completedSnapshotStoreManager.getBucketCompletedSnapshotStores()).isEmpty(); + assertThat(bucketMetricGroups).allMatch(AbstractMetricGroup::isClosed); + assertThat(tableBuckets) + .allSatisfy( + tableBucket -> + assertThat( + TestingMetricGroups.COORDINATOR_METRICS + .getTableBucketMetricGroup( + tablePath, tableBucket)) + .isNull()); + assertThatCode(() -> inFlightGauges.forEach(Gauge::getValue)).doesNotThrowAnyException(); + } + @Test void testCreateAndDropPartition() throws Exception { TablePath tablePath = TablePath.of(defaultDatabase, "test_create_drop_partition"); @@ -2192,6 +2272,33 @@ private CoordinatorEventProcessor buildCoordinatorEventProcessor() { SystemClock.getInstance()); } + private static class FailingUpdateMetadataChannelManager extends TestCoordinatorChannelManager { + + private final CountDownLatch failureObserved = new CountDownLatch(1); + private volatile boolean failUpdateMetadata; + + private void failNextUpdateMetadata() { + failUpdateMetadata = true; + } + + @Override + public void sendUpdateMetadataRequest( + int serverId, + UpdateMetadataRequest request, + BiConsumer responseConsumer) { + if (failUpdateMetadata) { + failUpdateMetadata = false; + failureObserved.countDown(); + throw new RuntimeException("Injected update metadata failure"); + } + super.sendUpdateMetadataRequest(serverId, request, responseConsumer); + } + + private void awaitFailure() throws InterruptedException { + assertThat(failureObserved.await(30, TimeUnit.SECONDS)).isTrue(); + } + } + private static class RecordingAutoPartitionManager extends AutoPartitionManager { private final CoordinatorMetadataCache metadataCache;