[KAFKA-12516] Standartize KafkaConsumer's commitSync/commitAsync/committed

• Type: Improvement
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Affects Version/s: 2.7.0
• Component/s: consumer
• Assignee: Unassigned
• Reporter: Ivan Bondarenko
• Created: 21/Mar/21 19:33
• Updated: 21/Mar/21 19:36

Description

Commit-related methods of KafkaConsumer are kind of asymmetrical.

The main axis of asymmetry is (let's call it "MAP"):

Map<TopicPartition, OffsetAndMetadata>

 

To illustrate asymmetry let's put methods to table (skipping deprecated and those having Duration to reduce table).

Before that, let's name "SAC" what is returned by code:

subscriptions.allConsumed()

 

So the table is...

name	params	returns	is SAC used?	can we get MAP?
commited	Set<TP>	MAP	no	via return
commitSync	–	–	used	no
commitSync	MAP	–	no	no
commitAsync	–	–	no	no
commitAsync	callback	–	used	SAC via callback
commitAsync	MAP, callback	–	no	no

What is asymmetric here...

1. Main problem. SAC can be retrieved only by 1 async method. So without workarounds we cannot just sync-commit (without parameters) and see what was committed.
To solve this we can (multiple solutions can be applied):

•  Introduce new method like "getAllConsumed()" which will return SAC.
• commitSync() without params can return SAC instead of void.

2. Minor. There is no "commited()" without params, so we need to create a Set manually.
Solutions:

•  We can add "commited()" without params. In this case the default set will be constructed based on one of:
  • SAC
  • all partitions for subscribed topic
  • assignment

 3. Looks like minor bug. This peace of code is used in all commitAsync() methods, but not in all commitSync() methods, it doesn't look like on purpose:

 

offsets.forEach(this::updateLastSeenEpochIfNewer);