[coordinator] Add rebalance progress and outcome metrics

[heyallencao]

Purpose

Closes #4055.

Expose coordinator-level metrics for monitoring rebalance progress, duration, and outcomes without polling the Admin API.

Brief change log

• Add seven gauges covering rebalance activity, pending/completed/failed/timed-out buckets, total elapsed time, and the current bucket's elapsed time.
• Add three counters for successfully completed, failed, and canceled rebalance rounds.
• Register the metrics once for the coordinator process lifetime through RebalanceMetrics, owned by CoordinatorMetricGroup. Retain the existing coordinator scope without additional table or bucket labels.
• Bind each RebalanceManager in startup() and unbind it on close. Counters survive leadership changes, while gauges return zero on standby.
• Reuse the existing task maps, state, and timestamps, and add regression tests and metric documentation.

Outcome semantics

A round with any failed or timed-out buckets increments the failure counter. The existing Admin API's COMPLETED status means that execution has finished, not necessarily that every bucket succeeded; this PR does not change that API behavior.

An empty plan counts as a successful round. The cancellation counter increments only when a running rebalance is canceled; repeated cancellation does not increment it again.

Leadership changes and recovery

Counters accumulate across all leader terms served by the same coordinator process and reset only when that process restarts. They remain unchanged on standby and are not transferred between coordinator processes.

Bucket-outcome gauges include only results observed in the current leader term. They remain available after completion or cancellation within that term and are cleared when the next round is registered. All gauges return zero on standby. Since per-bucket outcomes are not persisted, recovering an already finished round leaves the bucket-outcome gauges at zero and does not increment round counters. This avoids reporting historical failures as successes. These zeros mean that no outcomes were observed in the current term, not that the historical round had no failures.

Tests

• 84 targeted tests passed across 10 test classes, including all 19 RebalanceManagerTest tests and both RebalanceManagerITCase integration tests, plus coordinator lifecycle/election and metric-group tests.
• Regression coverage includes standby registration, counter continuity across leader terms, task recovery, metric registration/unregistration, and constructor-failure handling.
• Spotless, Checkstyle, Apache RAT, and git diff --check passed.

API and Format

No changes to public client APIs, RPC messages, persisted rebalance task format, or rebalance execution semantics.

This adds monitoring metrics. Their process-scoped lifetime and recovery behavior are documented.

Documentation

Add a Rebalance Metrics section to the monitoring documentation, describing metric names, types, meanings, and behavior during leadership changes and task recovery.

[morazow]

Thanks @heyallencao 🚀

Added minor nits

[heyallencao]

Thanks @heyallencao 🚀

Added minor nits

Thanks for the review, @morazow! All three comments are addressed. The 18 targeted rebalance tests and style checks pass.

[morazow]

Perfect!

Another suggestion:

Could we give the rebalance metrics the same lifetime as the coordinator process? We could introduce a RebalanceMetrics class owned by CoordinatorMetricGroup and register the counters and gauges once.

Then each RebalanceManager would bind/attach itself when created and unbind on close. The gauges would read from the current manager and return zero on standby, while counters would accumulate across leader terms within the same process.

This would avoid resetting counters and unregistering/re-registering metrics on leadership changes, and remove the need for a separate RebalanceMetricGroup:

• counters accumulate over every leader term the process serves and restart only when the process restarts
• gauges stay registered and report 0 on a standby, which is a meaningful value rather than an absent series
• getOrAddRebalanceMetricGroup, the RebalanceMetricGroup inner class and the CoordinatorMetricGroup.close() override all go away

Rough structure as below:

public class RebalanceMetrics {
    private final Counter rebalancesCompleted = new ThreadSafeSimpleCounter();
    // ...
    /** The manager of the current leader term, or null while this server is not the leader. */
    private volatile @Nullable RebalanceManager current;

    public RebalanceMetrics(MetricGroup metricGroup) {
        metricGroup.counter(MetricNames.REBALANCES_COMPLETED_TOTAL, rebalancesCompleted);
        // ...
        metricGroup.gauge(
                MetricNames.REBALANCE_BUCKETS_PENDING,
                () -> readCurrent(RebalanceManager::pendingBucketCount));
        // ...
    }

    private long readCurrent(ToLongFunction<RebalanceManager> read) {
        RebalanceManager manager = current;
        return manager == null ? 0L : read.applyAsLong(manager);
    }

    void bind(RebalanceManager manager) {
        current = manager;
    }

    void unbind(RebalanceManager manager) {
        // a late close from an ended term must not detach the manager that replaced it
        if (current == manager) {
            current = null;
        }
    }
}

RebalanceManager then takes coordinatorMetricGroup.getRebalanceMetrics(), calls metrics.bind(this) in the constructor and metrics.unbind(this) in close(), and increments through metrics.incRebalancesCompleted() and friends.

What do you think about this? The initial ticket wanted per leader term, but that would be misleading: counters are not monotonic and readers could not tell a leader reset from a real drop.

[morazow]

Thanks @heyallencao 🚀

I have added couple improvements 🤝 Then should be good to go!

[morazow]

Suggested change

	/** Rebalance metrics registered for the lifetime of a coordinator server. */
	/**
	* Rebalance metrics registered for the lifetime of a coordinator server.
	*
	* <p>The coordinator event thread binds and unbinds the {@link RebalanceManager} of the current
	* leader term and increments the counters, while the metric reporter thread reads the gauges
	* through {@link #readCurrent}. Both sides are safe to call concurrently.
	*/
	@ThreadSafe

[morazow]

Suggested change

	this.metrics = metrics;
	this.metrics = checkNotNull(metrics, "metrics");

[morazow]

On shutdown of this class, we should ensure that rebalance manager is closed even if exceptions thrown from earlier closes.

In shutdown method:

try {
        clearOfflineLeaderRetryTask();
       // close the event manager
       coordinatorEventManager.close();
} finally {
      rebalanceManager.close();
}

[morazow]

For above suggestion:

Suggested change

	finishedRebalanceTasks.clear();
	finishedRebalanceTasks.clear();
	finishedBucketCounts.values().forEach(count -> count.set(0L));

[morazow]

Because of above suggestion:

Suggested change

	finishedBucketCounts.get(statusForBucket).incrementAndGet();
	// Clear gate (bucket) first, then data (startMs).

[morazow]

I wonder if we can still improve this, instead of iterating for each gauge. Also the finishedRebalanceTasks are never cleared.

Suggestion: Add a map to count the statuses as the buckets finish.

/**
 * How many buckets of the current rebalance ended in each status.
 */
private final Map<RebalanceStatus, AtomicLong> finishedBucketCounts =  newFinishedBucketCounts();

private static Map<RebalanceStatus, AtomicLong> newFinishedBucketCounts() {
   Map<RebalanceStatus, AtomicLong> counts = new EnumMap<>(RebalanceStatus.class);
   for (RebalanceStatus status : RebalanceStatus.values()) {
          counts.put(status, new AtomicLong());
    }
     return counts;
}

And on iteration counts replace:

return finishedBucketCounts.get(status).get();

What do you think?

[heyallencao]

Thanks @morazow! I have addressed all six suggestions:

• Documented the concurrency model of RebalanceMetrics and marked it @ThreadSafe.
• Added the null check for RebalanceMetrics.
• Ensured RebalanceManager is closed through finally during coordinator shutdown.
• Replaced repeated scans of finished results with per-status AtomicLong counters.
• Reset those counters when registering a new rebalance.
• Incremented the corresponding counter when a bucket finishes.

I also did another self-review. All 19 RebalanceManagerTest tests pass, along with Checkstyle, Spotless, and RAT checks. Thanks again for the thoughtful suggestions!