diff --git a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java index ac60ee5f101..3649f7e23c0 100644 --- a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java +++ b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java @@ -47,6 +47,16 @@ public class MetricNames { public static final String KV_LEADER_REPLICA_CAPACITY = "kvLeaderReplicaCapacity"; public static final String REPLICAS_TO_DELETE_COUNT = "replicasToDeleteCount"; public static final String PENDING_LEADER_ACTIVATION_COUNT = "pendingLeaderActivationCount"; + public static final String REBALANCE_IN_PROGRESS = "rebalanceInProgress"; + public static final String REBALANCE_BUCKETS_PENDING = "rebalanceBucketsPending"; + public static final String REBALANCE_BUCKETS_COMPLETED = "rebalanceBucketsCompleted"; + public static final String REBALANCE_BUCKETS_FAILED = "rebalanceBucketsFailed"; + public static final String REBALANCE_BUCKETS_TIMED_OUT = "rebalanceBucketsTimedOut"; + public static final String REBALANCE_DURATION_MS = "rebalanceDurationMs"; + public static final String INFLIGHT_BUCKET_DURATION_MS = "inflightBucketDurationMs"; + public static final String REBALANCES_COMPLETED_TOTAL = "rebalancesCompletedTotal"; + public static final String REBALANCES_FAILED_TOTAL = "rebalancesFailedTotal"; + public static final String REBALANCES_CANCELED_TOTAL = "rebalancesCanceledTotal"; // for coordinator sender (per-tablet-server control request sender threads) public static final String SENDER_QUEUE_SIZE = "senderQueueSize"; public static final String SENDER_QUEUE_TIME_MS = "senderQueueTimeMs"; 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 aa04f8bec9e..fb0d3ca2124 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 @@ -277,7 +277,11 @@ public CoordinatorEventProcessor( this.internalListenerName = conf.getString(ConfigOptions.INTERNAL_LISTENER_NAME); this.rebalanceManager = new RebalanceManager( - this, zooKeeperClient, coordinatorEventManager, SystemClock.getInstance()); + this, + zooKeeperClient, + coordinatorEventManager, + SystemClock.getInstance(), + coordinatorMetricGroup.getRebalanceMetrics()); this.offlineLeaderRetryDelayMs = conf.get(ConfigOptions.COORDINATOR_OFFLINE_LEADER_RETRY_DELAY).toMillis(); if (offlineLeaderRetryDelayMs <= 0) { @@ -342,10 +346,13 @@ public void startup() { } public void shutdown() { - clearOfflineLeaderRetryTask(); - // close the event manager - coordinatorEventManager.close(); - rebalanceManager.close(); + try { + clearOfflineLeaderRetryTask(); + // close the event manager + coordinatorEventManager.close(); + } finally { + rebalanceManager.close(); + } onShutdown(); coordinatorContext.resetContext(); updateObservedKvLeaderReplicaCount(); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java index ce95dd862a0..4c818de1d1f 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java @@ -49,6 +49,7 @@ import javax.annotation.Nullable; import java.util.ArrayDeque; +import java.util.EnumMap; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -62,12 +63,15 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import static org.apache.fluss.cluster.rebalance.RebalanceStatus.CANCELED; import static org.apache.fluss.cluster.rebalance.RebalanceStatus.COMPLETED; +import static org.apache.fluss.cluster.rebalance.RebalanceStatus.FAILED; import static org.apache.fluss.cluster.rebalance.RebalanceStatus.FINAL_STATUSES; import static org.apache.fluss.cluster.rebalance.RebalanceStatus.NOT_STARTED; import static org.apache.fluss.cluster.rebalance.RebalanceStatus.REBALANCING; +import static org.apache.fluss.cluster.rebalance.RebalanceStatus.TIMEOUT; import static org.apache.fluss.utils.Preconditions.checkArgument; import static org.apache.fluss.utils.Preconditions.checkNotNull; @@ -90,6 +94,7 @@ public class RebalanceManager { private final EventManager eventManager; private final Clock clock; private final ScheduledExecutorService timeoutChecker; + private final RebalanceMetrics metrics; /** A queue of in progress table bucket to rebalance. */ private final Queue inProgressRebalanceTasksQueue = new ArrayDeque<>(); @@ -102,8 +107,11 @@ public class RebalanceManager { private final Map finishedRebalanceTasks = new ConcurrentHashMap<>(); + private final Map finishedBucketCounts = newFinishedBucketCounts(); + private final GoalOptimizer goalOptimizer; private volatile long registerTime; + private volatile boolean bucketResultsAvailable; private volatile @Nullable RebalanceStatus rebalanceStatus; private volatile @Nullable String currentRebalanceId; private volatile boolean isClosed = false; @@ -126,12 +134,14 @@ public RebalanceManager( CoordinatorEventProcessor eventProcessor, ZooKeeperClient zkClient, EventManager eventManager, - Clock clock) { + Clock clock, + RebalanceMetrics metrics) { this( eventProcessor, zkClient, eventManager, clock, + metrics, // TODO: Reuse the CoordinatorServer shared scheduler for this lightweight // coordinator timeout checker instead of creating a component-owned scheduler. Executors.newScheduledThreadPool( @@ -144,6 +154,7 @@ public RebalanceManager( ZooKeeperClient zkClient, EventManager eventManager, Clock clock, + RebalanceMetrics metrics, ScheduledExecutorService timeoutChecker) { this.eventProcessor = eventProcessor; this.zkClient = zkClient; @@ -151,10 +162,43 @@ public RebalanceManager( this.clock = clock == null ? SystemClock.getInstance() : clock; this.timeoutChecker = timeoutChecker; this.goalOptimizer = new GoalOptimizer(); + this.metrics = checkNotNull(metrics, "metrics"); + } + + long rebalanceInProgress() { + return rebalanceStatus == REBALANCING ? 1L : 0L; + } + + long pendingBucketCount() { + return inProgressRebalanceTasks.size(); + } + + long rebalanceDurationMs() { + return rebalanceStatus == REBALANCING + ? Math.max(0L, clock.milliseconds() - registerTime) + : 0L; + } + + long finishedBucketCount(RebalanceStatus status) { + if (!bucketResultsAvailable) { + return 0L; + } + return finishedBucketCounts.get(status).get(); + } + + private boolean hasFinishedBucket(RebalanceStatus status) { + return finishedBucketCounts.get(status).get() > 0; + } + + long inflightBucketDurationMs() { + TableBucket bucket = inflightTaskBucket; + long startMs = inflightTaskStartMs; + return bucket == null || startMs < 0 ? 0L : Math.max(0L, clock.milliseconds() - startMs); } public void startup() { LOG.info("Start up rebalance manager."); + metrics.bind(this); initialize(); } @@ -194,11 +238,13 @@ public void registerRebalance( Map rebalancePlan, RebalanceStatus newStatus) { checkNotClosed(); - registerTime = System.currentTimeMillis(); + registerTime = clock.milliseconds(); // first clear all exists tasks. inProgressRebalanceTasks.clear(); inProgressRebalanceTasksQueue.clear(); finishedRebalanceTasks.clear(); + finishedBucketCounts.values().forEach(count -> count.set(0L)); + bucketResultsAvailable = !FINAL_STATUSES.contains(newStatus); // Clear gate (bucket) first, then data (startMs). inflightTaskBucket = null; inflightTaskStartMs = -1; @@ -240,6 +286,7 @@ public void finishRebalanceTask(TableBucket tableBucket, RebalanceStatus statusF finishedRebalanceTasks.put( tableBucket, RebalanceResultForBucket.of(resultForBucket.plan(), statusForBucket)); + finishedBucketCounts.get(statusForBucket).incrementAndGet(); // Clear gate (bucket) first, then data (startMs). inflightTaskBucket = null; inflightTaskStartMs = -1; @@ -320,6 +367,9 @@ public void cancelRebalance(@Nullable String rebalanceId) { LOG.error("Error when delete rebalance plan from zookeeper.", e); } + if (rebalanceStatus == REBALANCING) { + metrics.incRebalancesCanceled(); + } rebalanceStatus = CANCELED; inProgressRebalanceTasksQueue.clear(); inProgressRebalanceTasks.clear(); @@ -343,20 +393,20 @@ public RebalanceTask generateRebalanceTask(List goalsByPriority) { String rebalanceId = UUID.randomUUID().toString(); try { // Generate the latest cluster model. - long startTime = System.currentTimeMillis(); + long startTime = clock.milliseconds(); ClusterModel clusterModel = buildClusterModel(eventProcessor.getCoordinatorContext()); LOG.info( "Build cluster model for rebalance id {} with {} ms.", rebalanceId, - System.currentTimeMillis() - startTime); + clock.milliseconds() - startTime); // do optimize. - startTime = System.currentTimeMillis(); + startTime = clock.milliseconds(); rebalancePlanForBuckets = goalOptimizer.doOptimizeOnce(clusterModel, goalsByPriority); LOG.info( "Do optimize for rebalance id {} with {} ms.", rebalanceId, - System.currentTimeMillis() - startTime); + clock.milliseconds() - startTime); } catch (Exception e) { LOG.error("Failed to generate rebalance plan.", e); throw e; @@ -407,6 +457,13 @@ private void completeRebalance() { LOG.error("Error when update rebalance plan from zookeeper.", e); } + if (bucketResultsAvailable) { + if (hasFinishedBucket(FAILED) || hasFinishedBucket(TIMEOUT)) { + metrics.incRebalancesFailed(); + } else { + metrics.incRebalancesCompleted(); + } + } rebalanceStatus = COMPLETED; inProgressRebalanceTasks.clear(); inProgressRebalanceTasksQueue.clear(); @@ -414,7 +471,7 @@ private void completeRebalance() { // Here, it will not clear finishedRebalanceTasks, because it will be used by // listRebalanceProgress. It will be cleared when next register. - LOG.info("Rebalance complete with {} ms.", System.currentTimeMillis() - registerTime); + LOG.info("Rebalance complete with {} ms.", clock.milliseconds() - registerTime); } private ClusterModel buildClusterModel(CoordinatorContext coordinatorContext) { @@ -470,6 +527,14 @@ private RebalanceTask buildRebalanceTask( return new RebalanceTask(rebalanceId, NOT_STARTED, bucketPlan); } + private static Map newFinishedBucketCounts() { + Map counts = new EnumMap<>(RebalanceStatus.class); + for (RebalanceStatus status : RebalanceStatus.values()) { + counts.put(status, new AtomicLong()); + } + return counts; + } + private boolean isOfflineTagged(ServerTag serverTag) { return serverTag == ServerTag.PERMANENT_OFFLINE || serverTag == ServerTag.TEMPORARY_OFFLINE; } @@ -518,6 +583,7 @@ private void checkNotClosed() { public void close() { isClosed = true; + metrics.unbind(this); timeoutChecker.shutdownNow(); } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceMetrics.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceMetrics.java new file mode 100644 index 00000000000..97b1bed7799 --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceMetrics.java @@ -0,0 +1,104 @@ +/* + * 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.server.coordinator.rebalance; + +import org.apache.fluss.metrics.Counter; +import org.apache.fluss.metrics.MetricNames; +import org.apache.fluss.metrics.ThreadSafeSimpleCounter; +import org.apache.fluss.metrics.groups.MetricGroup; + +import javax.annotation.Nullable; +import javax.annotation.concurrent.ThreadSafe; + +import java.util.function.ToLongFunction; + +import static org.apache.fluss.cluster.rebalance.RebalanceStatus.COMPLETED; +import static org.apache.fluss.cluster.rebalance.RebalanceStatus.FAILED; +import static org.apache.fluss.cluster.rebalance.RebalanceStatus.TIMEOUT; + +/** + * Rebalance metrics registered for the lifetime of a coordinator server. + * + *

The coordinator event thread binds and unbinds the current {@link RebalanceManager} and + * updates counters, while metric reporter threads may concurrently read gauges via {@link + * #readCurrent(ToLongFunction)}. + */ +@ThreadSafe +public class RebalanceMetrics { + + private final Counter rebalancesCompleted = new ThreadSafeSimpleCounter(); + private final Counter rebalancesFailed = new ThreadSafeSimpleCounter(); + private final Counter rebalancesCanceled = new ThreadSafeSimpleCounter(); + + private volatile @Nullable RebalanceManager current; + + /** Registers the rebalance metrics on the coordinator metric group. */ + public RebalanceMetrics(MetricGroup metricGroup) { + metricGroup.counter(MetricNames.REBALANCES_COMPLETED_TOTAL, rebalancesCompleted); + metricGroup.counter(MetricNames.REBALANCES_FAILED_TOTAL, rebalancesFailed); + metricGroup.counter(MetricNames.REBALANCES_CANCELED_TOTAL, rebalancesCanceled); + metricGroup.gauge( + MetricNames.REBALANCE_IN_PROGRESS, + () -> readCurrent(RebalanceManager::rebalanceInProgress)); + metricGroup.gauge( + MetricNames.REBALANCE_BUCKETS_PENDING, + () -> readCurrent(RebalanceManager::pendingBucketCount)); + metricGroup.gauge( + MetricNames.REBALANCE_BUCKETS_COMPLETED, + () -> readCurrent(manager -> manager.finishedBucketCount(COMPLETED))); + metricGroup.gauge( + MetricNames.REBALANCE_BUCKETS_FAILED, + () -> readCurrent(manager -> manager.finishedBucketCount(FAILED))); + metricGroup.gauge( + MetricNames.REBALANCE_BUCKETS_TIMED_OUT, + () -> readCurrent(manager -> manager.finishedBucketCount(TIMEOUT))); + metricGroup.gauge( + MetricNames.REBALANCE_DURATION_MS, + () -> readCurrent(RebalanceManager::rebalanceDurationMs)); + metricGroup.gauge( + MetricNames.INFLIGHT_BUCKET_DURATION_MS, + () -> readCurrent(RebalanceManager::inflightBucketDurationMs)); + } + + void bind(RebalanceManager manager) { + current = manager; + } + + void unbind(RebalanceManager manager) { + if (current == manager) { + current = null; + } + } + + void incRebalancesCompleted() { + rebalancesCompleted.inc(); + } + + void incRebalancesFailed() { + rebalancesFailed.inc(); + } + + void incRebalancesCanceled() { + rebalancesCanceled.inc(); + } + + private long readCurrent(ToLongFunction read) { + RebalanceManager manager = current; + return manager == null ? 0L : read.applyAsLong(manager); + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/CoordinatorMetricGroup.java b/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/CoordinatorMetricGroup.java index 31120962f6d..9ade6cc31c4 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/CoordinatorMetricGroup.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/CoordinatorMetricGroup.java @@ -26,6 +26,7 @@ import org.apache.fluss.metrics.groups.MetricGroup; import org.apache.fluss.metrics.registry.MetricRegistry; import org.apache.fluss.server.coordinator.event.CoordinatorEvent; +import org.apache.fluss.server.coordinator.rebalance.RebalanceMetrics; import javax.annotation.Nullable; @@ -54,12 +55,15 @@ public class CoordinatorMetricGroup extends AbstractMetricGroup { private final Map, CoordinatorEventMetricGroup> eventMetricGroups = new ConcurrentHashMap<>(); + private final RebalanceMetrics rebalanceMetrics; + public CoordinatorMetricGroup( MetricRegistry registry, String clusterId, String hostname, String serverId) { super(registry, new String[] {clusterId, hostname, NAME}, null); this.clusterId = clusterId; this.hostname = hostname; this.serverId = serverId; + this.rebalanceMetrics = new RebalanceMetrics(this); } @Override @@ -80,6 +84,11 @@ public CoordinatorEventMetricGroup getOrAddEventTypeMetricGroup( eventClass, e -> new CoordinatorEventMetricGroup(registry, eventClass, this)); } + /** Returns the rebalance metrics for this coordinator server. */ + public RebalanceMetrics getRebalanceMetrics() { + return rebalanceMetrics; + } + // ------------------------------------------------------------------------ // table buckets groups // ------------------------------------------------------------------------ diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java index 3bf3680fda6..2b5ebe49f1d 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java @@ -23,6 +23,12 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metrics.Counter; +import org.apache.fluss.metrics.Gauge; +import org.apache.fluss.metrics.Metric; +import org.apache.fluss.metrics.MetricNames; +import org.apache.fluss.metrics.registry.MetricRegistry; +import org.apache.fluss.metrics.registry.NOPMetricRegistry; import org.apache.fluss.server.coordinator.AutoPartitionManager; import org.apache.fluss.server.coordinator.CoordinatorContext; import org.apache.fluss.server.coordinator.CoordinatorEventProcessor; @@ -38,6 +44,7 @@ import org.apache.fluss.server.coordinator.lease.KvSnapshotLeaseManager; import org.apache.fluss.server.coordinator.remote.RemoteDirDynamicLoader; import org.apache.fluss.server.metadata.CoordinatorMetadataCache; +import org.apache.fluss.server.metrics.group.CoordinatorMetricGroup; import org.apache.fluss.server.metrics.group.TestingMetricGroups; import org.apache.fluss.server.zk.NOPErrorHandler; import org.apache.fluss.server.zk.ZkEpoch; @@ -56,10 +63,13 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -67,9 +77,18 @@ import java.util.concurrent.ScheduledThreadPoolExecutor; import static org.apache.fluss.cluster.rebalance.RebalanceStatus.COMPLETED; +import static org.apache.fluss.cluster.rebalance.RebalanceStatus.FAILED; import static org.apache.fluss.cluster.rebalance.RebalanceStatus.NOT_STARTED; import static org.apache.fluss.cluster.rebalance.RebalanceStatus.TIMEOUT; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; /** Test for {@link RebalanceManager}. */ public class RebalanceManagerTest { @@ -88,6 +107,9 @@ public class RebalanceManagerTest { private ReplicaCapacityController replicaCapacityController; private LakeTableTieringManager lakeTableTieringManager; private RebalanceManager rebalanceManager; + private CoordinatorMetricGroup coordinatorMetricGroup; + private MetricRegistry metricRegistry; + private ManualClock metricClock; private KvSnapshotLeaseManager kvSnapshotLeaseManager; private Scheduler scheduler; @@ -134,18 +156,24 @@ void beforeEach() { new LakeTableTieringManager(TestingMetricGroups.LAKE_TIERING_METRICS); CoordinatorEventProcessor eventProcessor = buildCoordinatorEventProcessor(conf); RecordingEventManager recordingEventManager = new RecordingEventManager(); + metricRegistry = mock(MetricRegistry.class); + coordinatorMetricGroup = + new CoordinatorMetricGroup(metricRegistry, "cluster", "host", "coordinator"); + metricClock = new ManualClock(0L); rebalanceManager = new RebalanceManager( eventProcessor, zookeeperClient, recordingEventManager, - SystemClock.getInstance()); + metricClock, + coordinatorMetricGroup.getRebalanceMetrics()); rebalanceManager.startup(); } @AfterEach void afterEach() throws Exception { rebalanceManager.close(); + coordinatorMetricGroup.close(); if (scheduler != null) { scheduler.shutdown(); } @@ -192,7 +220,12 @@ void testStartupQueuesRecoverRebalanceEvent() throws Exception { RebalanceManager manager = new RebalanceManager( - eventProcessor, zookeeperClient, eventManager, clock, executor); + eventProcessor, + zookeeperClient, + eventManager, + clock, + createCoordinatorMetricGroup().getRebalanceMetrics(), + executor); // If startup() finds a pending rebalance task in ZooKeeper, it should enqueue a // RecoverRebalanceEvent to be processed by the coordinator event thread, instead of // calling registerRebalance() directly on the startup thread. @@ -229,7 +262,12 @@ void testTimeoutEnqueuesEvent() throws Exception { RebalanceManager manager = new RebalanceManager( - eventProcessor, zookeeperClient, eventManager, clock, executor); + eventProcessor, + zookeeperClient, + eventManager, + clock, + createCoordinatorMetricGroup().getRebalanceMetrics(), + executor); manager.startup(); TableBucket tb1 = new TableBucket(1L, 0); @@ -281,7 +319,12 @@ void testTimeoutAfterCompletionIsNoOp() throws Exception { RebalanceManager manager = new RebalanceManager( - eventProcessor, zookeeperClient, eventManager, clock, executor); + eventProcessor, + zookeeperClient, + eventManager, + clock, + createCoordinatorMetricGroup().getRebalanceMetrics(), + executor); manager.startup(); TableBucket tb1 = new TableBucket(1L, 0); @@ -318,7 +361,12 @@ void testTimeoutTreatsTaskAsCompleted() throws Exception { RebalanceManager manager = new RebalanceManager( - eventProcessor, zookeeperClient, eventManager, clock, executor); + eventProcessor, + zookeeperClient, + eventManager, + clock, + createCoordinatorMetricGroup().getRebalanceMetrics(), + executor); manager.startup(); TableBucket tb1 = new TableBucket(1L, 0); @@ -356,7 +404,379 @@ void testTimeoutTreatsTaskAsCompleted() throws Exception { manager.close(); } + @Test + void testRebalanceMetricsLifecycle() { + assertThat(coordinatorMetricGroup.getLogicalScope(value -> value, '_')) + .isEqualTo("coordinator"); + assertThat(coordinatorMetricGroup.getScopeComponents()) + .containsExactly("cluster", "host", "coordinator"); + assertThat(coordinatorMetricGroup.getAllVariables()) + .containsOnlyKeys("cluster_id", "host", "server_id") + .containsEntry("cluster_id", "cluster") + .containsEntry("host", "host") + .containsEntry("server_id", "coordinator"); + assertThat(coordinatorMetricGroup.getMetrics()).hasSize(10); + coordinatorMetricGroup + .getMetrics() + .values() + .forEach( + metric -> { + if (metric instanceof Gauge) { + assertThat(((Number) ((Gauge) metric).getValue()).longValue()) + .isZero(); + } else { + assertThat(((Counter) metric).getCount()).isZero(); + } + }); + + Map plan = createRebalancePlan(2); + List buckets = new ArrayList<>(plan.keySet()); + rebalanceManager.registerRebalance("success", plan, NOT_STARTED); + assertThat(gaugeValue(MetricNames.REBALANCE_IN_PROGRESS)).isEqualTo(1); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_PENDING)).isEqualTo(2); + + metricClock.advanceTime(Duration.ofMillis(500)); + assertThat(gaugeValue(MetricNames.REBALANCE_DURATION_MS)).isEqualTo(500); + assertThat(gaugeValue(MetricNames.INFLIGHT_BUCKET_DURATION_MS)).isEqualTo(500); + + rebalanceManager.finishRebalanceTask(buckets.get(0), COMPLETED); + metricClock.advanceTime(Duration.ofMillis(200)); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_PENDING)).isEqualTo(1); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_COMPLETED)).isEqualTo(1); + assertThat(gaugeValue(MetricNames.REBALANCE_DURATION_MS)).isEqualTo(700); + assertThat(gaugeValue(MetricNames.INFLIGHT_BUCKET_DURATION_MS)).isEqualTo(200); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isZero(); + + rebalanceManager.finishRebalanceTask(buckets.get(1), COMPLETED); + rebalanceManager.finishRebalanceTask(buckets.get(1), COMPLETED); + assertThat(gaugeValue(MetricNames.REBALANCE_IN_PROGRESS)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_PENDING)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_COMPLETED)).isEqualTo(2); + assertThat(gaugeValue(MetricNames.REBALANCE_DURATION_MS)).isZero(); + assertThat(gaugeValue(MetricNames.INFLIGHT_BUCKET_DURATION_MS)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isEqualTo(1); + + rebalanceManager.registerRebalance("empty", Collections.emptyMap(), NOT_STARTED); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_COMPLETED)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isEqualTo(2); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_CANCELED_TOTAL)).isZero(); + } + + @Test + void testRebalanceFailureMetrics() { + Map plan = createRebalancePlan(3); + List buckets = new ArrayList<>(plan.keySet()); + rebalanceManager.registerRebalance("mixed-results", plan, NOT_STARTED); + rebalanceManager.finishRebalanceTask(buckets.get(0), FAILED); + rebalanceManager.finishRebalanceTask(buckets.get(1), TIMEOUT); + + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_PENDING)).isEqualTo(1); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_FAILED)).isEqualTo(1); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_TIMED_OUT)).isEqualTo(1); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isZero(); + + rebalanceManager.finishRebalanceTask(buckets.get(2), COMPLETED); + rebalanceManager.finishRebalanceTask(buckets.get(0), FAILED); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_COMPLETED)).isEqualTo(1); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_FAILED)).isEqualTo(1); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_TIMED_OUT)).isEqualTo(1); + assertThat(gaugeValue(MetricNames.REBALANCE_IN_PROGRESS)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isEqualTo(1); + assertThat(rebalanceManager.getRebalanceStatus()).isEqualTo(COMPLETED); + + rebalanceManager.registerRebalance("next", createRebalancePlan(1), NOT_STARTED); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_COMPLETED)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_FAILED)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_TIMED_OUT)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isEqualTo(1); + } + + @Test + void testRebalanceTimeoutMetrics() { + Map plan = createRebalancePlan(1); + TableBucket bucket = plan.keySet().iterator().next(); + rebalanceManager.registerRebalance("timeout", plan, NOT_STARTED); + metricClock.advanceTime(Duration.ofMinutes(3)); + rebalanceManager.checkTimeout(); + + assertThat(gaugeValue(MetricNames.INFLIGHT_BUCKET_DURATION_MS)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_PENDING)).isEqualTo(1); + assertThat(gaugeValue(MetricNames.REBALANCE_DURATION_MS)).isEqualTo(180_000); + rebalanceManager.finishRebalanceTask(bucket, TIMEOUT); + + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_TIMED_OUT)).isEqualTo(1); + assertThat(gaugeValue(MetricNames.REBALANCE_DURATION_MS)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isEqualTo(1); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isZero(); + } + + @Test + void testRebalanceCancellationMetrics() { + rebalanceManager.cancelRebalance(null); + assertThat(counterValue(MetricNames.REBALANCES_CANCELED_TOTAL)).isZero(); + + Map plan = createRebalancePlan(2); + rebalanceManager.registerRebalance("cancel", plan, NOT_STARTED); + rebalanceManager.finishRebalanceTask(plan.keySet().iterator().next(), COMPLETED); + rebalanceManager.cancelRebalance("cancel"); + rebalanceManager.cancelRebalance("cancel"); + assertThat(counterValue(MetricNames.REBALANCES_CANCELED_TOTAL)).isEqualTo(1); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_IN_PROGRESS)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_PENDING)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_COMPLETED)).isEqualTo(1); + assertThat(gaugeValue(MetricNames.REBALANCE_DURATION_MS)).isZero(); + assertThat(gaugeValue(MetricNames.INFLIGHT_BUCKET_DURATION_MS)).isZero(); + } + + @ParameterizedTest + @EnumSource( + value = RebalanceStatus.class, + names = {"COMPLETED", "CANCELED", "FAILED", "TIMEOUT"}) + void testRecoveringFinishedRebalanceDoesNotRestoreOutcomeMetrics(RebalanceStatus status) { + rebalanceManager.registerRebalance("recovered", createRebalancePlan(1), status); + assertThat(gaugeValue(MetricNames.REBALANCE_IN_PROGRESS)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_PENDING)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_COMPLETED)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_FAILED)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_TIMED_OUT)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_DURATION_MS)).isZero(); + assertThat(gaugeValue(MetricNames.INFLIGHT_BUCKET_DURATION_MS)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_CANCELED_TOTAL)).isZero(); + + rebalanceManager.registerRebalance("recovered-empty", Collections.emptyMap(), status); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_COMPLETED)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_FAILED)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_TIMED_OUT)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_CANCELED_TOTAL)).isZero(); + } + + @ParameterizedTest + @EnumSource( + value = RebalanceStatus.class, + names = {"FAILED", "TIMEOUT"}) + void testRecoveringMixedResultsDoesNotReportSuccessfulBuckets(RebalanceStatus failedStatus) + throws Exception { + Map plan = createRebalancePlan(2); + List buckets = new ArrayList<>(plan.keySet()); + String failureMetric = + failedStatus == FAILED + ? MetricNames.REBALANCE_BUCKETS_FAILED + : MetricNames.REBALANCE_BUCKETS_TIMED_OUT; + zookeeperClient.registerRebalanceTask( + new RebalanceTask("mixed-results", NOT_STARTED, plan)); + rebalanceManager.registerRebalance("mixed-results", plan, NOT_STARTED); + rebalanceManager.finishRebalanceTask(buckets.get(0), failedStatus); + rebalanceManager.finishRebalanceTask(buckets.get(1), COMPLETED); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_COMPLETED)).isEqualTo(1); + assertThat(gaugeValue(failureMetric)).isEqualTo(1); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isEqualTo(1); + + rebalanceManager.close(); + RecordingEventManager eventManager = new RecordingEventManager(); + rebalanceManager = + new RebalanceManager( + mock(CoordinatorEventProcessor.class), + zookeeperClient, + eventManager, + metricClock, + coordinatorMetricGroup.getRebalanceMetrics(), + new NoOpScheduledExecutor()); + rebalanceManager.startup(); + assertThat(eventManager.events).hasSize(1); + assertThat(eventManager.events.get(0)).isInstanceOf(RecoverRebalanceEvent.class); + RebalanceTask recoveredTask = + ((RecoverRebalanceEvent) eventManager.events.get(0)).getRebalanceTask(); + assertThat(recoveredTask).isEqualTo(new RebalanceTask("mixed-results", COMPLETED, plan)); + rebalanceManager.registerRebalance( + recoveredTask.getRebalanceId(), + recoveredTask.getExecutePlan(), + recoveredTask.getRebalanceStatus()); + + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_COMPLETED)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_FAILED)).isZero(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_TIMED_OUT)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isEqualTo(1); + assertThat(counterValue(MetricNames.REBALANCES_CANCELED_TOTAL)).isZero(); + assertThat(rebalanceManager.listRebalanceProgress(null).progressForBucketMap().values()) + .hasSize(2) + .extracting(RebalanceResultForBucket::status) + .containsOnly(COMPLETED); + + zookeeperClient.registerRebalanceTask(new RebalanceTask("next", NOT_STARTED, plan)); + rebalanceManager.registerRebalance("next", plan, NOT_STARTED); + rebalanceManager.finishRebalanceTask(buckets.get(0), failedStatus); + rebalanceManager.finishRebalanceTask(buckets.get(1), COMPLETED); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_COMPLETED)).isEqualTo(1); + assertThat(gaugeValue(failureMetric)).isEqualTo(1); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isEqualTo(2); + } + + @Test + void testRebalanceMetricsSurviveLeadershipChange() throws Exception { + Map registeredMetrics = new HashMap<>(coordinatorMetricGroup.getMetrics()); + rebalanceManager.registerRebalance("first-term", Collections.emptyMap(), NOT_STARTED); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isEqualTo(1); + + Map plan = createRebalancePlan(1); + rebalanceManager.registerRebalance("failed", plan, NOT_STARTED); + rebalanceManager.finishRebalanceTask(plan.keySet().iterator().next(), FAILED); + rebalanceManager.registerRebalance("canceled", plan, NOT_STARTED); + rebalanceManager.cancelRebalance("canceled"); + rebalanceManager.registerRebalance("unfinished", plan, NOT_STARTED); + zookeeperClient.registerRebalanceTask(new RebalanceTask("unfinished", NOT_STARTED, plan)); + metricClock.advanceTime(Duration.ofMillis(100)); + + RebalanceManager previousManager = rebalanceManager; + rebalanceManager.close(); + assertThat(coordinatorMetricGroup.isClosed()).isFalse(); + assertGaugesAreZero(); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isEqualTo(1); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isEqualTo(1); + assertThat(counterValue(MetricNames.REBALANCES_CANCELED_TOTAL)).isEqualTo(1); + + RecordingEventManager eventManager = new RecordingEventManager(); + rebalanceManager = + new RebalanceManager( + mock(CoordinatorEventProcessor.class), + zookeeperClient, + eventManager, + metricClock, + coordinatorMetricGroup.getRebalanceMetrics(), + new NoOpScheduledExecutor()); + rebalanceManager.startup(); + + assertThat(eventManager.events).hasSize(1); + RebalanceTask recoveredTask = + ((RecoverRebalanceEvent) eventManager.events.get(0)).getRebalanceTask(); + rebalanceManager.registerRebalance( + recoveredTask.getRebalanceId(), + recoveredTask.getExecutePlan(), + recoveredTask.getRebalanceStatus()); + previousManager.close(); + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_PENDING)).isEqualTo(1); + rebalanceManager.finishRebalanceTask(plan.keySet().iterator().next(), COMPLETED); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isEqualTo(2); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isEqualTo(1); + assertThat(counterValue(MetricNames.REBALANCES_CANCELED_TOTAL)).isEqualTo(1); + assertThat(coordinatorMetricGroup.getMetrics()).hasSize(10); + registeredMetrics.forEach( + (name, metric) -> + assertThat(coordinatorMetricGroup.getMetrics().get(name)).isSameAs(metric)); + verify(metricRegistry, times(10)).register(any(), anyString(), eq(coordinatorMetricGroup)); + verify(metricRegistry, never()).unregister(any(), anyString(), eq(coordinatorMetricGroup)); + + coordinatorMetricGroup.close(); + assertThat(coordinatorMetricGroup.getMetrics()).isEmpty(); + verify(metricRegistry, times(10)) + .unregister(any(), anyString(), eq(coordinatorMetricGroup)); + + rebalanceManager.close(); + coordinatorMetricGroup = createCoordinatorMetricGroup(); + assertGaugesAreZero(); + assertThat(counterValue(MetricNames.REBALANCES_COMPLETED_TOTAL)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_FAILED_TOTAL)).isZero(); + assertThat(counterValue(MetricNames.REBALANCES_CANCELED_TOTAL)).isZero(); + } + + @Test + void testRebalanceMetricsRegisteredOnStandby() { + CoordinatorMetricGroup standbyGroup = + new CoordinatorMetricGroup(metricRegistry, "cluster", "standby", "standby"); + try { + assertThat(standbyGroup.getMetrics().values()) + .hasSize(10) + .allSatisfy( + metric -> { + if (metric instanceof Gauge) { + assertThat( + ((Number) ((Gauge) metric).getValue()) + .longValue()) + .isZero(); + } else { + assertThat(((Counter) metric).getCount()).isZero(); + } + }); + verify(metricRegistry, times(10)).register(any(), anyString(), eq(standbyGroup)); + } finally { + standbyGroup.close(); + } + } + + @Test + void testUnstartedManagerDoesNotReplaceCurrentManager() { + rebalanceManager.registerRebalance("current", createRebalancePlan(1), NOT_STARTED); + RebalanceManager unstartedManager = + new RebalanceManager( + mock(CoordinatorEventProcessor.class), + zookeeperClient, + new RecordingEventManager(), + metricClock, + coordinatorMetricGroup.getRebalanceMetrics(), + new NoOpScheduledExecutor()); + try { + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_PENDING)).isEqualTo(1); + } finally { + unstartedManager.close(); + } + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_PENDING)).isEqualTo(1); + } + + @Test + void testFailedProcessorConstructionDoesNotReplaceCurrentManager() { + rebalanceManager.registerRebalance("current", createRebalancePlan(1), NOT_STARTED); + Configuration conf = new Configuration(); + conf.set(ConfigOptions.COORDINATOR_OFFLINE_LEADER_RETRY_DELAY, Duration.ZERO); + + assertThatThrownBy(() -> buildCoordinatorEventProcessor(conf, coordinatorMetricGroup)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(ConfigOptions.COORDINATOR_OFFLINE_LEADER_RETRY_DELAY.key()); + + assertThat(gaugeValue(MetricNames.REBALANCE_BUCKETS_PENDING)).isEqualTo(1); + } + + private void assertGaugesAreZero() { + coordinatorMetricGroup + .getMetrics() + .values() + .forEach( + metric -> { + if (metric instanceof Gauge) { + assertThat(((Number) ((Gauge) metric).getValue()).longValue()) + .isZero(); + } + }); + } + + private long gaugeValue(String metricName) { + return ((Number) + ((Gauge) coordinatorMetricGroup.getMetrics().get(metricName)).getValue()) + .longValue(); + } + + private long counterValue(String metricName) { + return ((Counter) coordinatorMetricGroup.getMetrics().get(metricName)).getCount(); + } + + private static CoordinatorMetricGroup createCoordinatorMetricGroup() { + return new CoordinatorMetricGroup(NOPMetricRegistry.INSTANCE, "cluster", "host", "0"); + } + private CoordinatorEventProcessor buildCoordinatorEventProcessor(Configuration conf) { + return buildCoordinatorEventProcessor(conf, createCoordinatorMetricGroup()); + } + + private CoordinatorEventProcessor buildCoordinatorEventProcessor( + Configuration conf, CoordinatorMetricGroup metricGroup) { return new CoordinatorEventProcessor( zookeeperClient, serverMetadataCache, @@ -365,7 +785,7 @@ private CoordinatorEventProcessor buildCoordinatorEventProcessor(Configuration c replicaCapacityController, autoPartitionManager, lakeTableTieringManager, - TestingMetricGroups.COORDINATOR_METRICS, + metricGroup, conf, Executors.newFixedThreadPool(1, new ExecutorThreadFactory("test-coordinator-io")), metadataManager, diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md index 0d31c09dd3b..a8a610745f4 100644 --- a/website/docs/maintenance/observability/monitor-metrics.md +++ b/website/docs/maintenance/observability/monitor-metrics.md @@ -449,6 +449,84 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM +#### Rebalance Metrics + +Rebalance metrics use the `coordinator` scope with no additional infix or per-table/per-bucket labels. +They are registered when the coordinator server starts and remain registered across leadership changes. +Gauges report 0 on standby. On the active coordinator, they describe the most recently registered +rebalance. Finished bucket counts include only outcomes observed in the current leader term, remain +available after completion or cancellation within that term, and are cleared when the next rebalance +is registered. Recovering a previously finished rebalance leaves these counts at 0 because per-bucket +outcomes are not persisted. These zeros indicate no +outcomes observed by this leader, not that the historical rebalance had no failures or timeouts. +Counters accumulate across rebalances and leader terms within the same coordinator process. They +retain their values on standby and reset when the coordinator server restarts, not on leadership +changes. These are per-process counts, not a cluster-wide total transferred between coordinators. +Recovering a previously finished rebalance does not increment the counters; a recovered unfinished +rebalance is counted by the coordinator that observes its completion. + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
MetricsDescriptionType
rebalanceInProgress1 while a rebalance is running; otherwise 0.Gauge
rebalanceBucketsPendingNumber of unfinished buckets, including the currently running bucket.Gauge
rebalanceBucketsCompletedNumber of buckets observed to complete successfully by this leader in the most recently registered rebalance.Gauge
rebalanceBucketsFailedNumber of buckets observed to fail by this leader in the most recently registered rebalance, excluding timeouts.Gauge
rebalanceBucketsTimedOutNumber of buckets observed to time out by this leader in the most recently registered rebalance.Gauge
rebalanceDurationMsElapsed milliseconds since the running rebalance was registered on this leader; 0 when idle.Gauge
inflightBucketDurationMsElapsed milliseconds for the currently running bucket; 0 when no bucket is running.Gauge
rebalancesCompletedTotalNumber of rebalances observed by this coordinator process to finish without failed or timed-out buckets, including empty plans, across all its leader terms.Counter
rebalancesFailedTotalNumber of rebalances observed by this coordinator process to finish with at least one failed or timed-out bucket, across all its leader terms. This is an outcome metric; the existing Admin API can still report the rebalance status as COMPLETED because execution has finished.Counter
rebalancesCanceledTotalNumber of running rebalances canceled by this coordinator process across all its leader terms. Repeated cancellation and cancellation when idle do not increment it.Counter
+ ### Tablet Server