[KAFKA-3370] Add options to auto.offset.reset to reset offsets upon initialization only

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Labels: needs-kip
• Assignee: Vahid Hashemian
• Reporter: Guozhang Wang
• Created: 09/Mar/16 22:11
• Updated: 29/Feb/24 15:41

Description

Currently "auto.offset.reset" is applied in the following two cases:

1) upon starting the consumer for the first time (hence no committed offsets before);

2) upon fetching offsets out-of-range.

For scenarios where case 2) needs to be avoid (i.e. people need to be notified upon offsets out-of-range rather than silently offset reset), "auto.offset.reset" need to be set to "none". However for case 1) setting "auto.offset.reset" to "none" will cause NoOffsetForPartitionException upon polling. And in this case, seekToBeginning/seekToEnd is mistakenly applied trying to set the offset at initialization, which are actually designed for during the life time of the consumer (in rebalance callback, for example).

The fix proposal is to add two more options to "auto.offset.reset", "earliest-on-start", and "latest-on-start", whose semantics are "earliest" and "latest" for case 1) only, and "none" for case 2).

Issue Links

• is related to: KAFKA-6189 Loosing messages on OFFSET_OUT_OF_RANGE error in consumer
• is related to: KAFKA-12478 Consumer group may lose data for newly expanded partitions when add partitions for topic if the group is set to consume from the latest
• links to: GitHub Pull Request #9006