[KAFKA-16107] Ensure consumer does not start fetching from added partitions until onPartitionsAssigned completes

• Type: Sub-task
• Status: Resolved
• Priority: Critical
• Resolution: Fixed
• Fix Version/s: 3.8.0
• Component/s: clients, consumer
• Labels: kip-848-client-support reconciliation
• Assignee: Lianet Magrans
• Reporter: Lianet Magrans
• Reviewer: David Jacot
• Created: 10/Jan/24 14:23
• Updated: 02/Feb/24 22:20
• Resolved: 24/Jan/24 12:34

Description

In the new consumer implementation, when new partitions are assigned, the subscription state is updated and then the #onPartitionsAssigned triggered. This sequence seems sensible but we need to ensure that no data is fetched until the onPartitionsAssigned completes (where the user could be setting the committed offsets it want to start fetching from).
We should pause the partitions newly added partitions until onPartitionsAssigned completes, similar to how it's done on revocation to avoid positions getting ahead of the committed offsets.

Issue Links

• is related to: KAFKA-16134 kafka.api.PlaintextConsumerTest.testPerPartitionLagMetricsCleanUpWithSubscribe(String, String).quorum=kraft+kip848.groupProtocol=consumer is flaky
• is related to: KAFKA-16135 kafka.api.PlaintextConsumerTest.testPerPartitionLeadMetricsCleanUpWithSubscribe(String, String).quorum=kraft+kip848.groupProtocol=consumer is flaky
• links to: GitHub Pull Request #15215
• links to: GitHub Pull Request #15218
• links to: GitHub Pull Request #15285