Hi All,
I have updated the KIP with recent suggestions.
Thanks.
On Fri, Nov 8, 2024 at 11:53 AM Manikumar <manikumar.re...@gmail.com> wrote:
>
> Hi Matthias,
>
> Thanks for the comments.
>
> > I am personally not convinced that adding a new config
> `auto.offset.reset.by.duration` is the best way.
>
> I think in this case, we felt it's better to add the strategy name to
> auto.offset.reset config and add additional strategy specific configs.
> This will be easy to understand and extend in future. I am open to
> change if there are similar concerns.
>
> > One question about "earliest_local": if we fetch this offset from the
> server, and seek it later, and start fetching, won't there be a race
> condition? the server would just need to reload it from the tier.
>
> Yes, your understanding is correct. the server will read from the tier
> storage.
> Anyhow, I will consider dropping earliest_local based on Jun's suggestion.
>
> > For Kafka Streams, just adding the new option so the existing enum does
> not work IMHO. If a user would use BY_DURATION, there is no option to pass
> in the actual duration (and we might need to throw an exception).
>
> I was assuming we will be using the underlying consumer config here. I
> will update based on your suggestions.
>
>
> Thanks,
>
>
>
>
> On Fri, Nov 8, 2024 at 6:21 AM Matthias J. Sax <mj...@apache.org> wrote:
> >
> > Thanks for updating the KIP.
> >
> > I am personally not convinced that adding a new config
> > `auto.offset.reset.by.duration` is the best way. Kafka in general has
> > way too many configs and trying to avoid adding more configs seems to be
> > desirable? -- It seems this might be a point of contention, and if the
> > majority of people wants this new config so be it. I just wanted to
> > stress my concerns about it.
> >
> > One question about "earliest_local": if we fetch this offset from the
> > server, and seek to it later, and start fetching, won't there be a race
> > condition? The oldest segment could have been discarded locally in the
> > mean time? Maybe not too much of a big deal: the server would just need
> > to reload it from the tier. Just want to ensure my understanding is
> > correct, and make sure nobody has concerns about it.
> >
> > For Kafka Streams, just adding the new option so the existing enum does
> > not work IMHO.
> >
> > For example, we have a method:
> >
> > addSource(final AutoOffsetReset offsetReset,
> > final String name,
> > final String... topics)
> >
> > If a user would use BY_DURATION, there is no option to pass in the
> > actual duration (and we might need to throw an exception). And if we
> > overload `addSource` adding a fourth parameter, it would be odd too, as
> > users could pass in `EARLIEST` plus a duration, and we would need to
> > throw an exception or ignore the given duration. Overall not an easy to
> > use API.
> >
> > Thus, I would suggest to deprecate the existing `AutoOffsetReset` enum,
> > and add a new class `AutoOffsetReset` for which we can guide users
> > accordingly avoiding corner cases:
> >
> > public class AutoOffsetReset() {
> > public static AutoOffsetReset earliest();
> > public static AutoOffsetReset latest();
> > public static AutoOffsetReset byDuration(Duration duration);
> > public static AutoOffsetReset earliestLocal();
> > }
> >
> > The existing enum is contained inside `Topology` so we can add the new
> > class just for package `org.apache.kafka.streams` to avoid naming conflicts.
> >
> > Additionally, we would deprecate all methods taking existing enum
> > `AutoOffsetReset` (only two classed `Topology` and `Consumed` are
> > affected) and add new overload which accepts the new `class
> > AutoOffsetReset` instead.
> >
> >
> > -Matthias
> >
> > On 11/7/24 1:57 PM, Jun Rao wrote:
> > > Hi, Manikumar,
> > >
> > > Thanks for the KIP. A couple of comments.
> > >
> > > JR1. It doesn't seem that we need earliest_local. Intuitively, it makes
> > > sense for a consumer app to bootstrap based on time. The time that a topic
> > > is kept locally seems irrelevant to a consumer application.
> > >
> > > JR2. I took a look at the proposal of KIP-842. It seems that this KIP
> > > covers the common scenario in KIP-842 and is simpler. The scenario
> > > described in KIP-842 is about which initial offset to use when a new
> > > partition is added. This is more or less the same problem for choosing the
> > > initial offset for existing partitions. Having a time based option
> > > provides
> > > more flexibility than earliest or latest, and is likely good enough for
> > > most usage.
> > >
> > > Jun
> > >
> > >
> > > On Wed, Nov 6, 2024 at 4:00 AM Apoorv Mittal <apoorvmitta...@gmail.com>
> > > wrote:
> > >
> > >> Thanks Manikumar and Andrew.
> > >>
> > >> For AM6: Though I was thinking of enhancing OffsetFetcher and adding
> > >> "seekToLocalBeginning" where OffsetFetcher does use Admin.listOffsets API
> > >> itself. However I agree that this is just another convenient way and not
> > >> necessary to support functionality. Skipping it sounds good to me.
> > >>
> > >> Regards,
> > >> Apoorv Mittal
> > >>
> > >>
> > >> On Wed, Nov 6, 2024 at 9:12 AM Andrew Schofield <
> > >> andrew_schofield_j...@outlook.com> wrote:
> > >>
> > >>> Hi Manikumar,
> > >>>
> > >>> AS7: In response to AM6, I agree that in general the combination
> > >>> of offsetsForTimes() and seek() will do what is required. The one case
> > >> that
> > >>> I think is more tricky is seeking to the local earliest/beginning.
> > >>>
> > >>> The Consumer API has Consumer.beginningOffsets and
> > >>> seekToBeginning, which correspond to OffsetResetStrategy.EARLIEST.
> > >>> Similarly there are methods for OffsetResetStrategy.LATEST.
> > >>>
> > >>> With the Consumer API, I don't think you can find the local earliest
> > >>> offset.
> > >>> That can be found using Admin.listOffsets with OffsetSpec.EarliestLocal.
> > >>> We could add equivalents of Consumer.beginningOffsets and
> > >>> Consumer.seekToBeginning. An application could always use the
> > >>> Admin API with the Consumer API.
> > >>>
> > >>> Personally, I would not complicate this KIP with these new methods when
> > >>> the focus is more on duration-based offset reset and configs.
> > >>>
> > >>> Thanks,
> > >>> Andrew
> > >>> ________________________________________
> > >>> From: Manikumar <manikumar.re...@gmail.com>
> > >>> Sent: 06 November 2024 04:20
> > >>> To: dev@kafka.apache.org <dev@kafka.apache.org>
> > >>> Subject: Re: [DISCUSS] KIP-1106: Add duration based offset reset option
> > >>> for consumer clients
> > >>>
> > >>> Hi Apoorv,
> > >>>
> > >>> Thanks for the review.
> > >>>
> > >>> AM5: Thanks, Updated the KIP.
> > >>>
> > >>> AM6: Currently we can achieve this with the combination of
> > >>> offsetsForTimes() and seek() API..
> > >>> Maybe we can add a generic API like seekToTimes(Map<TopicPartition,
> > >>> Long> timestampsToSearch) if required.
> > >>> Let's see what others think.
> > >>>
> > >>> Thanks,
> > >>>
> > >>> On Wed, Nov 6, 2024 at 1:45 AM Apoorv Mittal <apoorvmitta...@gmail.com>
> > >>> wrote:
> > >>>>
> > >>>> Hi Manikumar,
> > >>>> Thanks for the changes. Just minor comment and a question:
> > >>>>
> > >>>> AM5: The description "How to initialize the share-partition start
> > >>> offset:"
> > >>>> for "*share.auto.offset.reset*" seems incomplete as it ends with `:`.
> > >>>> Should we write the details as defined in KIP-932 i.e.
> > >>>>
> > >>>> How to initialize the share-partition start offset:
> > >>>>
> > >>>> -
> > >>>>
> > >>>> "earliest" : automatically reset the offset to the earliest offset
> > >>>> -
> > >>>>
> > >>>> "latest" : automatically reset the offset to the latest offset
> > >>>> -
> > >>>>
> > >>>> "*earliest_local": *automatically resets the offset to the earliest
> > >>>> message stored in the local log on the broker.
> > >>>> -
> > >>>>
> > >>>> "*by_duration": *automatically resets the offset to the earliest
> > >>> offset
> > >>>> whose timestamp is greater than or equal to the configured duration
> > >>>> * (auto.offset.reset.by.duration).*
> > >>>>
> > >>>>
> > >>>> *or* maybe just replace ":" with "." i.e. "How to initialize the
> > >>>> share-partition start offset." as the details of the new configs are
> > >>>> defined at top in different sections as well.
> > >>>>
> > >>>> AM6: Question: KafkaConsumer has public methods defined to "seek" for
> > >> the
> > >>>> reset strategy like "seekToBeginning" and "seekToEnd" which uses
> > >>>> OffsetResetStrategy of EARLIEST and LATEST respectively. Do you think
> > >> it
> > >>>> would be sensible to add additional KafkaConsumer APIs for newly
> > >>> introduced
> > >>>> strategies?
> > >>>>
> > >>>> Regards,
> > >>>> Apoorv Mittal
> > >>>>
> > >>>>
> > >>>> On Tue, Nov 5, 2024 at 6:10 PM Manikumar <manikumar.re...@gmail.com>
> > >>> wrote:
> > >>>>
> > >>>>> Hi Matthias,
> > >>>>>
> > >>>>> Thanks for the review.
> > >>>>>
> > >>>>> MS1: looks like KIP-842 is stuck in the voting. Current changes
> > >> should
> > >>>>> fit into KIP-842 (few updates definitely required) as KIP-842 is
> > >>>>> proposing to add new configs (auto.offset.reset.on.no.initial.offset
> > >>>>> and auto.offset.reset.on.invalid.offset). I prefer to revive KIP-842
> > >>>>> after completing this KIP.
> > >>>>>
> > >>>>> MS2: Added support for earliest-local config to reset to oldest local
> > >>>>> offset and changes to kafka-consumer-group cmd.
> > >>>>>
> > >>>>> MS3: Based on suggestions, I have updated the config option to
> > >> support
> > >>>>> ISO format.
> > >>>>>
> > >>>>> MS4: I have included the AutoOffsetReset changes to the KIP. I
> > >>>>> definitely need your help on refining the API/solution for KS :)
> > >>>>>
> > >>>>> Thanks
> > >>>>>
> > >>>>>
> > >>>>>
> > >>>>>
> > >>>>>
> > >>>>> On Tue, Nov 5, 2024 at 7:26 AM Matthias J. Sax <mj...@apache.org>
> > >>> wrote:
> > >>>>>>
> > >>>>>> Thanks for the KIP.
> > >>>>>>
> > >>>>>> A somewhat orthogonal question I have is, if we should try to merge
> > >>> this
> > >>>>>> KIP with the existing KIP-842:
> > >>>>>>
> > >>>>>
> > >>>
> > >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-842%3A+Add+richer+group+offset+reset+mechanisms
> > >>>>>>
> > >>>>>> While KIP-842 has a different intention, ie, separating the case of
> > >>> "no
> > >>>>>> offset during initial startup" vs "offset out of bounds during
> > >>>>>> processing", it might still be worth it? My motivation for this
> > >>> question
> > >>>>>> is, that `auto.offset.reset` is one of the most central configs we
> > >>> have,
> > >>>>>> and thus, doing a single larger change (ie, both KIPs together)
> > >>> might be
> > >>>>>> better (less noise) than doing two independent changes? Thoughts? I
> > >>> know
> > >>>>>> it might be a little bit of a stretch, but asking cannot hurt :)
> > >>>>>>
> > >>>>>>
> > >>>>>> About the KIP itself, did we consider to add something like
> > >>>>>> "latest-local" to say reset to oldest local, but not fetch from
> > >>> tiered
> > >>>>>> storage?
> > >>>>>>
> > >>>>>> For the format of the config, did we consider what we do in
> > >>>>>> `bin/kafka-consumer-group.sh`? It has multiple different options
> > >> like
> > >>>>>>
> > >>>>>> --by-duration <duration: format `PnDTnHnMnS`>
> > >>>>>> --shift-by <number-of-records>
> > >>>>>>
> > >>>>>> --to-datetime <fixed point in time: format
> > >>> `YYYY-MM-DDTHH:mm:SS.sss`>
> > >>>>>> --to-offset <absolute offset>
> > >>>>>>
> > >>>>>> As it also uses ISO format, it might be good to also use them (even
> > >>> if I
> > >>>>>> can see the appeal and simplicity of what you proposed).
> > >>>>>>
> > >>>>>>
> > >>>>>> I also want to add to (AM1), as there is also
> > >>> `Topology.AutoOffsetReset`
> > >>>>>> enum in Kafka Streams. I think we would need to convert this into a
> > >>>>>> class. While it could be done is a follow up KIP, too, it seems
> > >> best
> > >>> to
> > >>>>>> do this holistically in a single KIP, because KS is not something
> > >>> on-top
> > >>>>>> of Kafka, but it's part of Kafka. I am happy to help with the
> > >> design
> > >>> and
> > >>>>>> even PRs if necessary, but would strongly prefer to do it all in a
> > >>>>>> single KIP.
> > >>>>>>
> > >>>>>>
> > >>>>>> Btw: if we add something like "latest-local", it might also be good
> > >>> to
> > >>>>>> extend `bin/kafka-consumer-group.sh` accordingly (even if the tool
> > >>> works
> > >>>>>> slightly different, as it does commit an offset and there could be
> > >>> some
> > >>>>>> race condition between committing "latest-local", tiering, and when
> > >>> the
> > >>>>>> consumer is actually started?
> > >>>>>>
> > >>>>>>
> > >>>>>> -Matthias
> > >>>>>>
> > >>>>>>
> > >>>>>> On 11/4/24 7:38 AM, Apoorv Mittal wrote:
> > >>>>>>> Hi Manikumar,
> > >>>>>>> Thanks for the KIP, this new strategy would be helpful in
> > >>> specifying
> > >>>>> fetch
> > >>>>>>> behaviour.
> > >>>>>>>
> > >>>>>>> AM1: The config `auto.offset.reset` is currently applied as per
> > >> the
> > >>>>> enum
> > >>>>>>> class OffsetResetStarategy which is part of kafka-clients javadoc
> > >>>>>>> <
> > >>>>>
> > >>> https://javadoc.io/doc/org.apache.kafka/kafka-clients/latest/index.html
> > >>> .
> > >>>>>>> Are we also proposing to somehow add new definitions in the same
> > >>> class?
> > >>>>>>> However as new configurations will be a string representation
> > >> hence
> > >>>>> are we
> > >>>>>>> moving away from OffsetResetStartegy enum class altogether?
> > >> Should
> > >>> we
> > >>>>>>> include the change in the KIP as OffsetResetStarategy is part of
> > >>> public
> > >>>>>>> javadoc?
> > >>>>>>>
> > >>>>>>> AM2: While I can see the ISO-8601 format is in the rejected
> > >>>>> alternative,
> > >>>>>>> should we not follow some standard of defining duration which has
> > >>>>> already
> > >>>>>>> been adopted in other systems?
> > >>>>>>>
> > >>>>>>> AM3: We've introduced new config values using the format
> > >>> minus-n-hours,
> > >>>>>>> minus-n-days, minus-n-months, and minus-n-years. Should we
> > >>> explicitly
> > >>>>>>> define the "minus" prefix, or is it implied?
> > >>>>>>>
> > >>>>>>> AM4: When supporting duration-based resets, should we also allow
> > >>> users
> > >>>>> to
> > >>>>>>> specify a specific checkpoint time? For example, if a checkpoint
> > >>>>> occurs 2
> > >>>>>>> days, 5 hours and 30 minutes earlier, the current four formats
> > >>>>>>> (minus-n-hours, minus-n-days, minus-n-months, minus-n-years)
> > >> might
> > >>> not
> > >>>>> be
> > >>>>>>> sufficient. Should we consider adding a format to accommodate
> > >>> specific
> > >>>>>>> checkpoint times, or is there a reason to limit the supported
> > >>> formats?
> > >>>>>>>
> > >>>>>>>
> > >>>>>>> Regards,
> > >>>>>>> Apoorv Mittal
> > >>>>>>>
> > >>>>>>>
> > >>>>>>> On Mon, Nov 4, 2024 at 9:23 AM Manikumar <
> > >>> manikumar.re...@gmail.com>
> > >>>>> wrote:
> > >>>>>>>
> > >>>>>>>> Hi everyone,
> > >>>>>>>> I would like to start a discussion on KIP-1106:
> > >>>>>>>>
> > >>>>>>>>
> > >>>>>
> > >>>
> > >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1106%3A+Add+duration+based+offset+reset+option+for+consumer+clients
> > >>>>>>>>
> > >>>>>>>> This KIP proposes to add an additional auto offset reset
> > >> strategy
> > >>> for
> > >>>>>>>> consumer clients.
> > >>>>>>>>
> > >>>>>>>> Regards,
> > >>>>>>>> Manikumar
> > >>>>>>>>
> > >>>>>>>
> > >>>>>
> > >>>
> > >>
> > >
