Explain incremental cooperative rebalancing (KIP-415) in Kafka Connect: how does it differ from eager, and what is connect.protocol?

## The problem it solves The legacy **eager** protocol was *stop-the-world*: on any membership or config change, every worker revoked **all** its connectors and tasks before a new assignment was computed, pausing every pipeline in the cluster. **KIP-415 — Incremental Cooperative Rebalancing** (shipped in Apache Kafka 2.3) fixes this. ## Core idea: revoke only what must move Instead of asking everyone to drop everything, the leader: 1. Looks at the **current** assignment (who runs what now). 2. Computes the **target** assignment (who should run what after the change). 3. Revokes **only the delta** — the specific connectors/tasks that need to relocate to reach balance. 4. Leaves every other task running undisturbed. So adding a worker means the leader picks a handful of tasks to move onto it; all other tasks keep flowing. There is no cluster-wide pause. ## Two-round (incremental) mechanics A single round of the group protocol can't both revoke from old owners and assign to new owners safely (a task must be fully stopped on worker A before it starts on worker B, or you'd run it twice). So cooperative rebalancing is **incremental**, using two rebalance rounds: - **Round 1 (revoke):** the leader tells the relevant workers to revoke the to-be-moved tasks. Those workers stop just those tasks and rejoin. - **Round 2 (assign):** now that the tasks are free, the leader assigns them to their new owners, which start them. The word *cooperative* captures this: members cooperate by giving up only their share of the delta rather than everything. ## The connect.protocol setting A **worker-level** config (`connect.protocol`) chooses the strategy. All workers in a cluster negotiate the highest protocol they all support: - **`eager`** — legacy stop-the-world. Use only to force old behavior. - **`compatible`** — the **default** since 2.3. Workers use cooperative rebalancing, but during a **mixed-version upgrade** (some workers still on an old image) the cluster automatically **falls back to eager** until every worker supports cooperative. This makes upgrades safe with no config gymnastics. - **`sessioned`** — adds **KIP-507** session protection: the leader signs assignments with a key so a rogue/stale worker cannot forge or replay an assignment. It builds on cooperative behavior and adds security/correctness for the internal Connect topics. ## Why two rounds isn't a regression You might think two rounds is slower than eager's one. But round 1 only stops the *moving* tasks, not the whole cluster, and round 2 only starts those same few. The total disruption is proportional to the size of the **delta**, not the size of the **cluster** — which is the whole point. ## Companion: scheduled.rebalance.max.delay.ms Cooperative rebalancing introduced a delay knob (KIP-415, `scheduled.rebalance.max.delay.ms`, default 300000 ms / 5 min). When a worker **leaves**, the leader can leave that worker's tasks **unassigned** for up to this delay, betting the worker will come back (e.g. a quick restart or GC pause) and reclaim its own tasks — avoiding a move-out-then-move-back churn. If the worker doesn't return within the delay, the tasks are reassigned to the survivors. ## Edge cases / gotchas - **Mixed versions**: while any worker doesn't support cooperative, `compatible` keeps the whole cluster on eager — you don't get the benefit until the *last* old worker is upgraded. - **Lost tasks during the delay window**: with a non-zero `scheduled.rebalance.max.delay.ms`, a genuinely dead worker's tasks stay **down** until the delay elapses, trading availability for stability. Tune it to your tolerance. - **Imbalance after delay**: after the delay reassigns tasks, a later rebalance may rebalance again when the worker returns. Cooperative rebalancing's incremental moves keep that cheap.