[KAFKA-17480] New consumer commit all consumed should retrieve offsets in background thread

• Type: Bug
• Status: Patch Available
• Priority: Major
• Resolution: Unresolved
• Fix Version/s: 4.0.0
• Component/s: clients, consumer
• Labels: consumer-threading-refactor kip-848-client-support
• Assignee: PoAn Yang
• Reporter: Lianet Magrans
• Created: 04/Sep/24 15:21
• Updated: 27/Sep/24 14:03

Description

When committing all consumed offsets (sync, async, or on close), the new consumer retrieves the offsets from subscriptionState.allConsumed() in the app thread (and generates events containing those offsets, that are sent to the background thread to be processed). We should review this to consider retrieving the allConsumed in the background when processing the events, to avoid inconsistencies given that the subscription state could be modified in the background thread since the moment the allConsumed was retrieved in the app thread.

These are the places in the app thread that currently retrieve the allConsumed to pass it into the events. 

• https://github.com/apache/kafka/blob/59f5d91d8f1412c50c74a776a624879101402284/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L735
• https://github.com/apache/kafka/blob/59f5d91d8f1412c50c74a776a624879101402284/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L1299
• https://github.com/apache/kafka/blob/59f5d91d8f1412c50c74a776a624879101402284/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L1323

We could consider to keep the same CommitSync and CommitAsync events, but with the option of not taking offsets as input (maybe a flag for commitAllConsumed). We could also create new events for the allConsumed alternatives (I lean towards keeping just CommitSync and CommitAsync but open to whatever seems clearer)

Issue Links

• links to: GitHub Pull Request #17150