Status

Current state: Under Discussion

Discussion thread: here

JIRA: here

Motivation

When a Kafka broker is restarted or has its log directories rebuilt — for example after a process restart, OS reboot, host or hardware swap, storage (disk / volume) replacement, VM migration, or container reschedule — it rejoins the cluster and must re-replicate a large amount of data from its peers. Catch-up happens per partition: each partition's log fetches independently and re-enters the ISR at its own pace.

In KRaft today, the controller hands leadership back to the recovering broker the moment a partition re-enters the ISR — regardless of how much catch-up work the broker still has outstanding on its other partitions. The result:

• The recovering broker becomes leader for the already-caught-up partitions while its disk / network / page cache is still saturated by replica fetches for the lagging partitions.

• Producers to those newly-led partitions see high P99 latency (and, in the worst case, timeouts and retries).

• On large clusters, the leadership comes back in a single burst, producing a spike in incoming connections and produce/fetch requests on the broker.

• Internal topics (__consumer_offsets, __transaction_state, __share_group_state) are particularly affected: their preferred leader can return to a still-loading broker while the old leader is still unloading, prolonging consumer-group recovery.

Current KRaft behavior

The periodic electPreferred task runs every leader.imbalance.check.interval.seconds (default 300s) and calls ReplicationControlManager.maybeBalancePartitionLeaders():

// ReplicationControlManager.java
static final int MAX_ELECTIONS_PER_IMBALANCE = 1_000;   // hardcoded

ControllerResult<Boolean> maybeBalancePartitionLeaders() {
    List<ApiMessageAndVersion> records = new ArrayList<>();
    maybeTriggerLeaderChangeForPartitionsWithoutPreferredLeader(records, maxElectionsPerImbalance);
    return ControllerResult.of(records, records.size() >= maxElectionsPerImbalance);
}

Two problems for the broker-recovery scenario:

1. No gate. Preferred-leader election fires as soon as the preferred

   replica is in the ISR for a given partition. There is no notion of "this broker is still catching up globally, don't give it leadership yet."

2. No effective throttle. maxElectionsPerImbalance is hardcoded to

   1000 and is not configurable in production (only overridden in tests via QuorumController.Builder.setMaxElectionsPerImbalance). Worse, when a run hits the 1000 cap it returns true, and PeriodicTaskControlManager reschedules the task immediately (nextDelayTimeNs(immediate=true) → task.immediatePeriodNs()), so elections drain in back-to-back 1000-partition batches — effectively all at once.

imbalancedPartitions is a flat, per-partition TimelineHashSet; there is no per-broker grouping or per-broker rate limiting anywhere in the KRaft path.

The immediate-reschedule batching behavior predates KRaft (it is visible in 3.9.1 logs as well); what 4.1+ changed is the timing of the first election relative to the broker event (now decoupled from broker unfencing, fired on the next periodic tick 0–300s later). This KIP addresses both the batching and the timing.

Related work

This KIP adds a leadership-pacing/gating layer (the controller decides when and how fast a recovering broker regains leadership). It is distinct from and complementary to replication throttling (KIP-73 / KIP-1009 / KIP-1051), which governs how fast a broker catches up. The two can coexist.

Public Interfaces

Five new controller configurations are added to the leader.imbalance.* family. All are cluster-level dynamic configs, settable at runtime via kafka-configs.sh --entity-type brokers --entity-default --alter. The active controller's value is the one used by the next periodic electPreferred run; values on standby controllers are read only when they become active. Topic-level overrides are not supported — the periodic balancer is a controller-wide decision, not per-topic.

Defaults preserve current behavior exactly. The feature is opt-in.

Four new controller-side JMX metrics are added under kafka.controller:type=KafkaController:

No existing public APIs (AdminClient, request/response schemas, ACLs, CLI tools, on-disk metadata) are changed.

Proposed Changes

Throttle

Make the per-run election cap configurable and pace successive runs when the cap is hit.

• Read leader.imbalance.election.max.per.run from ReplicationConfigs and pass it into ReplicationControlManager.Builder.setMaxElectionsPerImbalance (today this is pinned to the hardcoded MAX_ELECTIONS_PER_IMBALANCE constant in QuorumController.Builder).

• Extend PeriodicTask with an optional throttled reschedule delay, used only when the task signals "more work remains" (return value true). Sketch of the API change:

public class PeriodicTask {
    // existing
    private final long periodNs;
    private final long immediatePeriodNs;

    // new, optional; empty preserves current immediate-reschedule behavior
    private final OptionalLong throttledRescheduleDelayNs;

    public static class Builder {
        public Builder setThrottledRescheduleDelay(OptionalLong d) { ... }
    }
}

// PeriodicTaskControlManager.nextDelayTimeNs:
if (immediate) {
    return task.throttledRescheduleDelayNs().orElse(task.immediatePeriodNs());
}

`QuorumController.registerElectPreferred(...)` populates `throttledRescheduleDelayNs` from `leader.imbalance.election.throttle.interval.ms`. All other periodic tasks (`electUnclean`, `maybeFenceStaleBroker`, `generatePeriodicPerformanceMessage`, `expireDelegationTokens`) leave it empty and retain their current immediate-reschedule behavior.

Interaction with leader.imbalance.check.interval.seconds. The check interval defines the baseline periodic schedule, used when a run did not cap out. The throttle interval defines the minimum gap between consecutive capped runs. When both apply the throttle interval is used (so the operator is in full control of pacing during a recovery). If throttle.interval.ms > check.interval.seconds * 1000, the throttle wins when capped; when not capped, normal periodic scheduling resumes.

Gate

The gate is engaged only when leader.imbalance.election.algorithm = wait-for-sync. The decision is computed once per balancer run, from data already available in ReplicationControlManager:

1. Group all entries of imbalancedPartitions by their preferred replica

   (partition.replicas[0]).

2. For each candidate broker b, compute

   outOfSyncCount(b) = number of partitions in the group where b is not in partition.isr, and totalCount(b) = number of partitions in the group.

3. Mark b as gated if either:

   • outOfSyncCount(b) / totalCount(b) > threshold.percent / 100, and

   • b has been gated for less than max.wait.ms.

4. Skip preferred election for every partition whose preferred replica is a

   gated broker for this run. Those partitions are reconsidered on subsequent runs.

Hysteresis (flap protection). A broker is released from the gate only after 2 consecutive runs in which it would not be gated. This prevents a transient ISR drop from revoking eligibility mid-rebalance. The constant 2 is internal and not configurable in v1; it can be promoted to a config if needed.

Max-wait escape hatch. max.wait.ms is a safety net: if a broker remains gated for that long (continuously), the gate releases it. This protects against latent bugs and against pathological situations where the broker can never catch up due to factors outside this code path. Per-broker gated-start time is kept in a transient Map<Integer, Long> on the active controller (reset on controller failover; that is acceptable because failover itself is a recovery event that re-evaluates the gate state).

Why threshold-based, not strict. A strict "block on any out-of-sync" gate (threshold.percent = 0) creates a real failure mode: a single straggler partition keeps the broker from leading thousands of other already-caught-up partitions. The threshold lets operators tune the trade-off between "lead nothing until perfect" (0) and "lead unless substantially behind" (5 is a reasonable starting point). The default 0 preserves the strictest, most conservative behavior — operators relax it only if they hit the straggler scenario.

Multi-broker recovery: round-robin scheduling

The current implementation iterates imbalancedPartitions in hash order and takes the first N. Under a rolling restart this is unfair: the first broker's backlog can monopolize many consecutive runs and starve later brokers of leadership for many minutes.

This KIP changes the candidate selection inside maybeTriggerLeaderChangeForPartitionsWithoutPreferredLeader to round-robin across eligible preferred replicas (i.e., those that pass the gate):

candidates = group(imbalancedPartitions, by = preferredReplica)
candidates = filter(candidates, broker not gated)

chosen = []
while size(chosen) < maxPerRun and any candidate bucket non-empty:
    for broker in cycle(candidates.keys()):
        if candidates[broker] non-empty:
            chosen.append(candidates[broker].pop())
            if size(chosen) == maxPerRun: break

Under immediate (default) algorithm, this changes nothing observable: all brokers are eligible, and over time each gets its fair share. Under wait-for-sync with multiple recovering brokers, both brokers' leadership ramps up in parallel rather than one waiting for the other to finish.

The round-robin layer adds one pass over imbalancedPartitions to construct the grouping; this is O(N) and is negligible compared to the existing per-partition election work.

Combined behavior

The gate decides which brokers receive leadership return (a broker that is substantially behind across many preferred partitions gets none until it catches up). The throttle decides how fast eligible elections drain (one batch of max.per.run partitions every throttle.interval.ms). Round-robin scheduling decides fairness across multiple eligible brokers (parallel ramp, not serial). Together they turn the broker-recovery period from a "catch-up + leadership flood" event into a "catch-up, then smooth parallel ramp" event.

Worked example

A broker is replaced and rejoins; it is the preferred replica for ~5,000 partitions and is catching up on all of them.

Configuration: algorithm = wait-for-sync, threshold.percent = 5, max.wait.ms = 1800000, max.per.run = 200, throttle.interval.ms = 15000.

• Today: as partitions trickle into ISR, each electPreferred run (and its immediate reschedules) hands leadership back in 1000-partition bursts while the broker is still replicating the rest, producing produce-latency spikes on the returned leaders.

• With this KIP: the broker receives no leadership while more than 250 (5% of 5,000) of its preferred partitions are still out of ISR. Once it drops below the threshold (and stays below for at least 2 consecutive periodic runs), leadership returns 200 partitions at a time every 15 seconds. Estimated total leadership-ramp window for 5,000 partitions: ~6.25 min after the gate releases (5000 ÷ 200 × 15s). Measured numbers will be reported in the load-test results.

• With two brokers recovering simultaneously (rolling restart): each run alternates ~100 elections for broker A and ~100 for broker B until both drain. Neither broker sits idle while the other completes.

Compatibility, Deprecation, and Migration Plan

• Backward compatible. All five new configurations default to today's behavior (immediate, 0, 1800000, 1000, 0). A broker/controller upgraded to a release containing this KIP behaves identically to the prior release until an operator opts in by changing leader.imbalance.election.algorithm and/or leader.imbalance.election.throttle.interval.ms.

• No deprecations. No existing configuration, API, metric, or CLI is removed or changed. The internal constant MAX_ELECTIONS_PER_IMBALANCE remains as the default value of the new config.

• Rolling upgrade. The new configurations are read by the active KRaft controller on each balancer run. A standard rolling upgrade of the controller quorum is sufficient; no broker-side or client-side change is required. During the upgrade window, the active controller's version is the one that matters: if the active controller is the upgraded one, the new configs apply; if it is a not-yet-upgraded one, behavior is as before. Operators wanting deterministic behavior should complete the controller quorum upgrade before setting the configs.

• Downgrade. Safe. The new configurations are unknown to older versions and are ignored. No on-disk metadata format changes. A controller downgrade after the configs were set will silently revert to the prior behavior.

• Dynamic config changes mid-recovery. All five configs are dynamic. Setting algorithm = wait-for-sync during an in-progress recovery immediately gates further preferred-leader elections from the next periodic run; already-elected leaders are not rolled back (this is by design — the goal is to stop the bleeding, not to revoke decisions already made).

• Interaction with __consumer_offsets, __transaction_state, and other internal topics. Internal topics participate in the gate and throttle on the same terms as user topics. They are not exempted. This is intentional: internal-topic leadership return on a still-loading broker is one of the primary symptoms reported in KAFKA-20075. Operators who need stricter behavior for internal topics can use a more aggressive threshold.percent = 0 (strict gate) — the gate already covers them.

• Interaction with controlled shutdown. Controlled shutdown performs its own leadership handoff before the broker stops; this KIP only affects the periodic preferred-leader rebalance that runs after the broker comes back. The two do not compose oddly.

• Interaction with KIP-966 (Eligible Leader Replicas). ELR is consulted only when the ISR is empty (no clean election possible). The gate is computed against ISR membership, so a preferred replica that is in ELR but not ISR is treated as out-of-sync by the gate, which is the correct behavior (an ELR-only replica is behind on HWM by definition and should not be elected as preferred leader for a still-recovering broker).

• Interaction with reassignments. Unchanged. The existing partitionsBeingReassigned and topic-deletion guards in the candidate filter still apply before the gate is evaluated.

Rejected Alternatives

1. Replication throttling only (leader.replication.throttled.rate /

   follower.replication.throttled.rate). Available today and attacks the resource-contention root cause, but does not change leadership timing (caught-up partitions still lead while others lag) and it slows recovery. Complementary to this KIP, not a substitute.

2. External leader-drain tooling (reassign replicas so the recovering

   broker is last, auto-rollback when offset lag reaches zero). Works without controller changes but reorders replica assignments for tens of thousands of partitions (heavy metadata churn), is reactive/external rather than the controller making the correct decision, and rollback/races are fragile. Useful only as an operational bridge before this KIP lands.

3. Strict (binary) gate without a threshold. Simpler to specify and

   implement, but creates a real failure mode: a single straggler partition holds the recovering broker out of all leadership for as long as the straggler is stuck. The threshold-based gate captures the same intent for the common case (threshold = 0 preserves the strict semantics) while giving operators a knob to escape the straggler-pin case.

4. Per-broker fully-fair throttle queues. Closer to a per-broker

   model with separate per-broker rate limits. Round-robin selection (this KIP) achieves the same parallel-ramp outcome with substantially less code and no new state. A fully fair scheduler can be added as a follow-up KIP if global max.per.run proves insufficient for asymmetric recoveries.

5. Gate at broker unfencing. Keep the broker fenced from leadership

   until log catch-up completes. Rejected: unfencing is metadata-catch-up based, not log-data based, and conflating the two risks broader availability impact (e.g., the broker also cannot serve fetches, which slows the very replication we are waiting for).

6. Reviving KIP-491 (preferred leader blacklist). KIP-491 was

   discarded; it offered manual deprioritization rather than an automatic recovery-aware policy, and would still require an operator to drain and undrain at the right moment. The mechanism proposed here is automatic and tied to ISR state.

Test Plan

Unit tests (ReplicationControlManagerTest):

• leader.imbalance.election.max.per.run caps elections per run and the result correctly signals "more work remains" when capped.

• wait-for-sync with threshold.percent = 0 skips a broker that has any out-of-ISR preferred partition; elects once the broker is fully in sync.

• wait-for-sync with threshold.percent = 5 does not gate a broker with 4% out-of-ISR preferred partitions; does gate at 6%.

• max.wait.ms releases a gated broker after the configured duration even if it remains out of sync.

• Hysteresis: a broker that briefly drops out of ISR during one periodic run remains gated for one additional run after returning to sync.

• Round-robin selection: with two eligible brokers each having 1,000 imbalanced preferred partitions and max.per.run = 200, each broker receives 100 ± 1 elections per run.

• immediate (default) algorithm produces byte-for-byte identical records to the pre-KIP behavior, including with mixed in-sync / out-of-sync brokers and with multiple recovering brokers.

Periodic task (PeriodicTaskControlManagerTest):

• When electPreferred returns true and throttle.interval.ms > 0, the next execution is deferred by exactly the configured interval.

• When throttle.interval.ms = 0, immediate-reschedule behavior is preserved for electPreferred and is unchanged for all other periodic tasks (regression test: pass each of electUnclean, maybeFenceStaleBroker, generatePeriodicPerformanceMessage, expireDelegationTokens and assert each retains immediatePeriodNs()).

• Dynamic config change to throttle.interval.ms takes effect on the next run.

Integration (controller integration tests):

• Simulate a broker rejoining with staggered ISR re-entry across 5,000 partitions; assert no leadership is granted while >threshold.percent of its preferred partitions are out of ISR, and that leadership returns in max.per.run-sized batches after catch-up.

• Simulate two brokers rejoining simultaneously; assert that round-robin scheduling gives each broker an approximately equal share of each capped run.

Metrics:

• OutOfSyncPreferredPartitionCount{broker=N} reflects the actual count during a recovery and returns to 0 once the broker is in sync.

• PreferredLeaderElectionsSkippedByGatePerSec and PreferredLeaderElectionsDeferredByThrottlePerSec are non-zero during the gated/throttled phase and return to 0 after.

System / load test:

• On a load-test KRaft cluster with ≥10,000 partitions on the target broker, restart one broker and compare the recovery window with and without the feature. Acceptance criteria:

  • Produce P99 latency, measured over the 10-minute window starting at broker unfencing, stays within 1.5× of the prior 1-hour steady-state P99 (today: typically 3–10× spike).

  • No produce request times out for clients with request.timeout.ms ≥ 30000.

  • Total recovery time (broker unfenced → broker is leader for ≥99% of its preferred partitions) is at most 2× the today-baseline. A moderate slowdown is acceptable; a regression beyond 2× is not.

Upgrade / downgrade:

• Mixed-version controller quorum during a rolling upgrade: active controller's version determines behavior; cluster remains functional in all combinations.

• Downgrading a controller that had the new configs set leaves the cluster in a functional state (the older controller ignores the configs); the feature simply turns off.