[KAFKA-14048] The Next Generation of the Consumer Rebalance Protocol

• Type: Improvement
• Status: Reopened
• Priority: Major
• Resolution: Unresolved
• Fix Version/s: 4.0.0
• Assignee: David Jacot
• Reporter: David Jacot
• Created: 06/Jul/22 08:28
• Updated: 26/Aug/24 08:21

Description

This Jira tracks the development of KIP-848: https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol.

Issue Links

• blocks: KAFKA-14524 Modularize `core` monolith
• links to: GitHub Pull Request #13203
• links to: GitHub Pull Request #14090
• links to: GitHub Pull Request #14327
• links to: GitHub Pull Request #16859
• links to: GitHub Pull Request #16937
• links to: GitHub Pull Request #16945
• mentioned in: Page Loading...

Sub-Tasks

• Automated protocol should support namespace: Open
• Add new `group-coordinator` module: Resolved
• Introduce `GroupCoordinator` interface: Resolved
• Add ConsumerGroupHeartbeat API: Resolved
• The Kafka protocol should support nullable structs: Resolved
• New Group Coordinator State Machine: Resolved
• Implement OffsetFetch/OffsetCommit APIs: Resolved
• Implement JoinGroup/SyncGroup APIs: Resolved
• Implement Heartbeat API: Resolved
• Implement LeaveGroup API: Resolved
• Implement ListGroups: Resolved
• Implement DescribeGroups API: Resolved
• Implement TnxOffsetCommit API: Resolved
• Implement DeleteGroups API and OffsetDelete API: Resolved
• Add ConsumerGroupPrepareAssignment API: Open
• Add ConsumerGroupInstallAssignment API: Open
• Add ConsumerGroupDescribe API: Resolved
• Extend DescribeConfigs API to support group configs: Resolved
• Extend AlterIncrementalConfigs API to support group config: Resolved
• Add broker side PartitionAssignor interface: Resolved
• Implement range broker side assignor: Resolved
• Implement optimized uniform broker side assignor: Resolved
• Implement static membeship: Resolved
• Implement regex subscriptions: In Progress
• Rebalance on topic/partition metadata changes: Resolved
• Add metrics to the new coordinator: Resolved
• Add high watermark listener to Partition/Log layers: Resolved
• Move __consumer_offsets records from core to group-coordinator: Resolved
• Add new __consumer_offsets records: Resolved
• OffsetCommit API Version 10: Patch Available
• OffsetFetch API Version 9: In Progress
• More broker side partition assignor to common: Resolved
• Extend server side assignor to support rack aware replica placement: Resolved
• Add ConsumerGroupHeartbeat API to AuthorizerIntegrationTest: Resolved
• Add support of topic id for OffsetCommitRequests in CommitRequestManager: Open
• Add ACL Authorizer integration test for authorized OffsetCommits with an unknown topic: Open
• Propagate Topic Ids to the Group Coordinator during Offsets Commit: Resolved
• Propagate Topic Ids to the Group Coordinator for Offset Fetch: Open
• Refactor ApiMessageAndVersion/Record: Open
• Validate that all partitions are assigned in TargetAssignmentBuilder: Resolved
• Implement Group/Offset expiration: Resolved
• CoordinatorPartitionWriter should reuse buffer: Resolved
• Consolidate all the group coordinator configs: Open
• Define constants for record types: Open
• Implement write operation timeout: Resolved
• CoordinatorContext should be protected by a lock: Resolved
• Consider replacing Subscription Metadata by a hash: Open
• Add support for OffsetCommit version 9 in admin client: Resolved
• Add support for OffsetCommit version 9 in consumer: Resolved
• Add support for OffsetFetch version 9 in consumer: Resolved
• Add support for OffsetFetch version 9 in admin: Resolved
• Add group type filter to ListGroups API: Resolved
• Add integration test for the ListGroup API: Resolved
• Add group type filter to the admin client: Resolved
• Implement general uniform broker side assignor: Resolved
• Handle heartbeat and revocation when consumer leaves group: Resolved
• Client consumer should commit offsets on close: Resolved
• Send HB request right after reconciliation completes: Resolved
• Ensure HeartbeatRequestManager only send out some fields once: Resolved
• Ensure coordinator node is removed upon disconnection exception: Resolved
• Enable existing client integration tests for new protocol: Resolved
• Implement client support for selecting KIP-848 server-side assignor: Resolved
• Implement client support for KIP-848 ConsumerGroupHeartbeat protocol RPC: Resolved
• Implement event plumbing for ConsumerRebalanceListener callbacks: Resolved
• Implement consumer group membership state machine: Resolved
• Client should stop fetching while partitions being revoked: Closed
• Update client state machine to align with protocol sending one assignment at a time: Resolved
• Implement auto-commit on partition assignment revocation: Closed
• Update states and transitions for membership manager state machine: Resolved
• Run System Tests for Old protocol in the New Coordinator: Resolved
• Member does not need to always try to join a group when a groupId is configured: Closed
• Add integration tests for DescribeGroups API, DeleteGroups API and OffsetDelete API: Resolved
• Add histogram metrics to GroupCoordinatorRuntimeMetrics: Resolved
• Refactor ConsumerRebalanceListener invocation for reuse: Resolved
• Do not send new heartbeat request while another one in-flight: Closed
• Improve unloading logging: Resolved
• Fix the different behavior in error handling between the old and new group coordinator: Resolved
• Client support for new consumer configs: Resolved
• New integration tests to have full coverage for preview: Closed
• Add integration tests for Heartbeat API and GroupLeave API: Resolved
• Migrate existing integration tests to run old protocol in new coordinator: Resolved
• Group Coordinator should not deliver new assignment before previous one is acknowledged: Resolved
• Implement listTopics() and partitionFor() for the AsyncKafkaConsumer: Resolved
• Correct initialization of ConsumerGroupHeartbeat by client: Closed
• Correct handling of KafkaConsumer.committed for new consumer: Resolved
• Fix ListGroups API when runtime partition size is zero: Resolved
• Add integration tests for JoinGroup API and SyncGroup API: Resolved
• Move new group coordinator metrics from Yammer to Metrics: Resolved
• Make General Assignor's rebalance incremental: Resolved
• New group coordinator needs to generate snapshots while loading: Resolved
• Refactor polling delays: Resolved
• update Group size only when groups size changes: Resolved
• Move GenericGroup state metrics to `GroupCoordinatorMetricsShard`: Resolved
• Upgrade existing generic group to consumer group: Resolved
• Ensure fairness in the uniform assignor: Resolved
• new group coordinator should check if partition goes offline during load: Resolved
• Add `group.coordinator.rebalance.protocols` and publish all new configs: Resolved
• Rename `Generic` to `Classic`: Resolved
• Trigger client reconciliation based on manager poll: Resolved
• Client support for OffsetFetch and OffsetCommit with topic ID: Open
• Review client retry logic of OffsetFetch and OffsetCommit responses: Resolved
• Client support for java regex based subscription: Resolved
• Group commit/callbacks triggering logic: Open
• Review consumer leave group request best effort and response handling: Resolved
• Review minimal effort approach on consumer last heartbeat on unsubscribe: Open
• Review consumer onPartitionsAssigned called with empty partitions: Open
• Topic ID integration in consumer subscription state: Open
• Update list group state type filter to include the states for the new consumer group type: Resolved
• Handle timeouts for AsyncKafkaConsumer.commitSync: Resolved
• AsyncKafkaConsumer: Always await async commit callbacks in commitSync and close: Resolved
• group size counters do not reflect the actual sizes when operations fail: Resolved
• Ensure consumer does not start fetching from added partitions until onPartitionsAssigned completes: Resolved
• Review client errors thrown on OffsetFetch and OffsetCommit failures: Resolved
• Add Integration test for checking if the correct assignor is chosen: Resolved
• Coordinator unloading fails when replica is deleted: Resolved
• Partition is assigned to two members at the same time: Resolved
• Implement GroupMetadataManager#onUnloaded: Resolved
• System test failing for new consumer on endOffsets with negative timestamps: Resolved
• Consumer invalid transition on expired poll interval: Resolved
• Implement GroupCoordinator.onPartitionsDeleted: Resolved
• Fix client reconciliation of same assignment received in different epochs: Resolved
• Extend admin to support ConsumerGroupDescribe API: Resolved
• Member should send full heartbeat when rejoining: Resolved
• KafkaConsumer.groupMetadata() should be correct when first records are returned: Closed
• Consumer does not rejoin after fenced on delayed revocation: Resolved
• Fix handling of deleted topic when auto-committing before revocation: Closed
• Console consumer fails with `IllegalStateException`: Closed
• Move code style exceptions from suppressions.xml to the code: Open
• Improve reconciliation state machine: Resolved
• Fenced member should not send heartbeats while waiting for onPartitionsLost to complete: Resolved
• Stale member should trigger onPartitionsLost when leaving group: Resolved
• Add group protocol migration enabling config: Resolved
• Ensure user callbacks exceptions are propagated to the user on consumer poll: Resolved
• Classic group error responses should contain a no generation id and an empty members list: Resolved
• Review fenced member unsubscribe/subscribe callbacks interaction: Open
• GroupCoordinatorService logger is not configured: Resolved
• fix EventAccumulator thread idle ratio metric: Resolved
• Offline group protocol migration: Resolved
• Add event rate in GroupCoordinatorRuntimeMetrics: Resolved
• Offline protocol migration integration tests: Resolved
• Full ConsumerGroupHeartbeat response must be sent when full request is received: Resolved
• High watermark updates should have a higher priority: Resolved
• Fix logic for discarding reconciliation if member rejoined: Resolved
• Coordinator flush time and event purgatory time metrics: Resolved
• Online upgrade triggering and group type conversion: Resolved
• kafka-dump-log --offsets-decoder should support new records: Resolved
• getOrMaybeCreateClassicGroup should not thrown GroupIdNotFoundException: Resolved
• Online downgrade triggering and group type conversion: Resolved
• Add JMH Benchmarks for assignor performance testing: Resolved
• Target Assignment Format Change: Resolved
• Store subscription model for consumer group in group state: Resolved
• JoinGroup API for upgrading ConsumerGroup: Resolved
• Reverse Lookup Partition to Member in Assignors: Resolved
• Uuid to String for subscribed topic names in assignment spec: Resolved
• CoordinatorRuntime write timer tasks should be cancelled once HWM advances: Resolved
• Re-add EventAccumulator.take(timeout): Resolved
• Optimize toTopicPartitions with ConsumerProtocolSubscription: Resolved
• Adjust classicGroupJoinToConsumerGroup to add subscription model: Resolved
• Remove rack aware code in assignors temporarily due to performance: Resolved
• SyncGroup API for upgrading ConsumerGroup: Resolved
• Coalesce records into bigger batches: Resolved
• Heartbeat API for upgrading ConsumerGroup: Resolved
• Create a new interface to store member metadata: Resolved
• CoordinatorRuntime should initialize MemoryRecordsBuilder with max batch size write limit: Resolved
• LeaveGroup API for upgrading ConsumerGroup: Resolved
• Introduce `group.version` feature flag: Resolved
• Copy on write in the Optimized Uniform Assignor: Resolved
• UniformHeterogeneousAssignmentBuilder throws NPE when member has not subscriptions: Resolved
• Range assignor doesn't co-partition with stickiness: Resolved
• Fix caught-up condition: Resolved
• Online protocol migration integration tests: Open
• Extend CoordinatorRuntime to support non-atomic writes: Resolved
• Static member using new protocol should always replace the one using the old protocol: Resolved
• System tests for protocol migration: Open
• Update upgrade notes for 4.0: Open
• Soften the validation when replaying tombstones: Resolved
• Validate and maybe trigger downgrade after static member replacement: Resolved
• Use the new group coordinator by default in 4.0: Resolved
• Re-introduce `group.version` feature flag: Resolved
• Delete kafka.common.OffsetAndMetadata: Open
• Factor out common setup from server-side assignor benchmarks: Resolved
• Add heterogeneous configuration to TargetAssignmentBuilderBenchmark: Resolved
• Revert #17219: Resolved
• Remove partitionRacks from TopicMetadata: Resolved
• Reduce logging verbosity on ConsumerGroupHeartbeat path: Resolved
• Rework UniformHeterogeneousAssignor to improve performance: Open