[lvthanh03]

Adds the kafka_cooperative_rebalance config. When enabled, consumers join their groups with librdkafka's cooperative-sticky assignor instead of the default eager assignors, so a rebalance only moves the partitions that change owner instead of revoking every partition.

Changes

• kafka_cooperative_rebalance sets partition.assignment.strategy: cooperative-sticky for every consumed topic in the pool.
• handle_events_cooperative runs one pipeline per partition, so an incremental revoke flushes and commits only the revoked partitions while the rest keep consuming. Eager mode keeps the existing handle_events with one shared pipeline.

With the config disabled, behavior is unchanged. Except for that InconsistentGroupProtocol no longer terminate the broker.

• This is to help with switching over a pool from eager to cooperative rebalancing.
• Since librdkafka can't mix eager and cooperative members in one group a broker that joins a group still on the other protocol now logs a warning, increments arroyo.consumer.inconsistent_group_protocol, and keeps retrying (they will successfully join when all current eager members leave the group)

Notes

I would only roll this out for push taskbroker pools. The reasons are that we could use maxSurge: 100% and maxUnavailable: 0 for a Deployment.

We would then have N cooperative pods start next to the N eager pods and push from the shared store while they wait to join, then all eager pods terminate together and the cooperative pods join the group at once.

This showed up in local load testing for me, but max_pending_count isn't strict in cooperative mode, each per-partition writer checks the pending count and then inserts, so concurrent writers can overshoot.

[untitaker]

let's just try it

[evanh]

Should we assert here? Is this really an error we should crash on?