Neha, I see how one might wish to implement onPartitionsAssigned and onPartitionsRevoked, but I don’t have a sense for how I might supply these implementations to a running consumer. What would the setup code look like to start a high-level consumer with these provided implementations?
thanks, Rob On Feb 27, 2014, at 3:48 AM, Neha Narkhede <neha.narkh...@gmail.com> wrote: > Rob, > > The use of the callbacks is explained in the javadoc here - > http://people.apache.org/~nehanarkhede/kafka-0.9-consumer-javadoc/doc/kafka/clients/consumer/ConsumerRebalanceCallback.html > > Let me know if it makes sense. The hope is to improve the javadoc so that > it is self explanatory. > > Thanks, > Neha > > > On Wed, Feb 26, 2014 at 9:16 AM, Robert Withers > <robert.w.with...@gmail.com>wrote: > >> Neha, what does the use of the RebalanceBeginCallback and >> RebalanceEndCallback look like? >> >> thanks, >> Rob >> >> On Feb 25, 2014, at 3:51 PM, Neha Narkhede <neha.narkh...@gmail.com> >> wrote: >> >>> How do you know n? The whole point is that you need to be able to fetch >> the >>> end offset. You can't a priori decide you will load 1m messages without >>> knowing what is there. >>> >>> Hmm. I think what you are pointing out is that in the new consumer API, >> we >>> don't have a way to issue the equivalent of the existing >> getOffsetsBefore() >>> API. Agree that is a flaw that we should fix. >>> >>> Will update the docs/wiki with a few use cases that I've collected so far >>> and see if the API covers those. >>> >>> I would prefer PartitionsAssigned and PartitionsRevoked as that seems >>> clearer to me >>> >>> Well the RebalanceBeginCallback interface will have >> onPartitionsAssigned() >>> as the callback. Similarly, the RebalanceEndCallback interface will have >>> onPartitionsRevoked() as the callback. Makes sense? >>> >>> Thanks, >>> Neha >>> >>> >>> On Tue, Feb 25, 2014 at 2:38 PM, Jay Kreps <jay.kr...@gmail.com> wrote: >>> >>>> 1. I would prefer PartitionsAssigned and PartitionsRevoked as that seems >>>> clearer to me. >>>> >>>> -Jay >>>> >>>> >>>> On Tue, Feb 25, 2014 at 10:19 AM, Neha Narkhede < >> neha.narkh...@gmail.com >>>>> wrote: >>>> >>>>> Thanks for the reviews so far! There are a few outstanding questions - >>>>> >>>>> 1. It will be good to make the rebalance callbacks forward compatible >>>> with >>>>> Java 8 capabilities. We can change it to PartitionsAssignedCallback >>>>> and PartitionsRevokedCallback or RebalanceBeginCallback and >>>>> RebalanceEndCallback? >>>>> >>>>> If there are no objections, I will change it to RebalanceBeginCallback >>>> and >>>>> RebalanceEndCallback. >>>>> >>>>> 2. The return type for committed() is List<TopicPartitionOffset>. >> There >>>>> was a suggestion to change it to either be Map<TopicPartition,Long> or >>>>> Map<TopicPartition, TopicPartitionOffset> >>>>> >>>>> Do people have feedback on this suggestion? >>>>> >>>>> >>>>> On Tue, Feb 25, 2014 at 9:56 AM, Neha Narkhede < >> neha.narkh...@gmail.com >>>>>> wrote: >>>>> >>>>>> Robert, >>>>>> >>>>>> Are you saying it is possible to get events from the high-level >>>>> consumerregarding various state machine changes? For instance, can we >>>> get a >>>>>> notification when a rebalance starts and ends, when a partition is >>>>>> assigned/unassigned, when an offset is committed on a partition, when >> a >>>>>> leader changes and so on? I call this OOB traffic, since they are not >>>>> the >>>>>> core messages streaming, but side-band events, yet they are still >>>>>> potentially useful to consumers. >>>>>> >>>>>> In the current proposal, you get notified when the state machine >>>> changes >>>>>> i.e. before and after a rebalance is triggered. Look at >>>>>> ConsumerRebalanceCallback< >>>>> >>>> >> http://people.apache.org/~nehanarkhede/kafka-0.9-consumer-javadoc/doc/kafka/clients/consumer/ConsumerRebalanceCallback.html >>>>>> >>>>>> .Leader changes do not count as state machine changes for consumer >>>>>> rebalance purposes. >>>>>> >>>>>> Thanks, >>>>>> Neha >>>>>> >>>>>> >>>>>> On Tue, Feb 25, 2014 at 9:54 AM, Neha Narkhede < >>>> neha.narkh...@gmail.com >>>>>> wrote: >>>>>> >>>>>>> Jay/Robert - >>>>>>> >>>>>>> >>>>>>> I think what Robert is saying is that we need to think through the >>>>> offset >>>>>>> API to enable "batch processing" of topic data. Think of a process >>>> that >>>>>>> periodically kicks off to compute a data summary or do a data load or >>>>>>> something like that. I think what we need to support this is an api >> to >>>>>>> fetch the last offset from the server for a partition. Something like >>>>>>> long lastOffset(TopicPartition tp) >>>>>>> and for symmetry >>>>>>> long firstOffset(TopicPartition tp) >>>>>>> >>>>>>> Likely this would have to be batched. >>>>>>> >>>>>>> A fixed range of data load can be done using the existing APIs as >>>>>>> follows. This assumes you know the endOffset which can be >>>> currentOffset >>>>> + n >>>>>>> (number of messages in the load) >>>>>>> >>>>>>> long startOffset = consumer.position(partition); >>>>>>> long endOffset = startOffset + n; >>>>>>> while(consumer.position(partition) <= endOffset) { >>>>>>> List<ConsumerRecord> messages = consumer.poll(timeout, >>>>>>> TimeUnit.MILLISECONDS); >>>>>>> process(messages, endOffset); // processes messages >>>> until >>>>>>> endOffset >>>>>>> } >>>>>>> >>>>>>> Does that make sense? >>>>>>> >>>>>>> >>>>>>> On Tue, Feb 25, 2014 at 9:49 AM, Neha Narkhede < >>>> neha.narkh...@gmail.com >>>>>> wrote: >>>>>>> >>>>>>>> Thanks for the review, Jun. Here are some comments - >>>>>>>> >>>>>>>> >>>>>>>> 1. The using of ellipsis: This may make passing a list of items from >>>> a >>>>>>>> collection to the api a bit harder. Suppose that you have a list of >>>>>>>> topics >>>>>>>> stored in >>>>>>>> >>>>>>>> ArrayList<String> topics; >>>>>>>> >>>>>>>> If you want subscribe to all topics in one call, you will have to >> do: >>>>>>>> >>>>>>>> String[] topicArray = new String[topics.size()]; >>>>>>>> consumer.subscribe(topics. >>>>>>>> toArray(topicArray)); >>>>>>>> >>>>>>>> A similar argument can be made for arguably the more common use case >>>> of >>>>>>>> subscribing to a single topic as well. In these cases, user is >>>> required >>>>>>>> to write more >>>>>>>> code to create a single item collection and pass it in. Since >>>>>>>> subscription is extremely lightweight >>>>>>>> invoking it multiple times also seems like a workable solution, no? >>>>>>>> >>>>>>>> 2. It would be good to document that the following apis are mutually >>>>>>>> exclusive. Also, if the partition level subscription is specified, >>>>> there >>>>>>>> is >>>>>>>> no group management. Finally, unsubscribe() can only be used to >>>> cancel >>>>>>>> subscriptions with the same pattern. For example, you can't >>>> unsubscribe >>>>>>>> at >>>>>>>> the partition level if the subscription is done at the topic level. >>>>>>>> >>>>>>>> *subscribe*(java.lang.String... topics) >>>>>>>> *subscribe*(java.lang.String topic, int... partitions) >>>>>>>> >>>>>>>> Makes sense. Made the suggested improvements to the docs< >>>>> >>>> >> http://people.apache.org/~nehanarkhede/kafka-0.9-consumer-javadoc/doc/kafka/clients/consumer/Consumer.html#subscribe%28java.lang.String...%29 >>>>>> >>>>>>>> >>>>>>>> >>>>>>>> 3.commit(): The following comment in the doc should probably say >>>>> "commit >>>>>>>> offsets for partitions assigned to this consumer". >>>>>>>> >>>>>>>> If no partitions are specified, commits offsets for the subscribed >>>>> list >>>>>>>> of >>>>>>>> topics and partitions to Kafka. >>>>>>>> >>>>>>>> Could you give more context on this suggestion? Here is the entire >>>> doc >>>>> - >>>>>>>> >>>>>>>> Synchronously commits the specified offsets for the specified list >> of >>>>>>>> topics and partitions to *Kafka*. If no partitions are specified, >>>>>>>> commits offsets for the subscribed list of topics and partitions. >>>>>>>> >>>>>>>> The hope is to convey that if no partitions are specified, offsets >>>> will >>>>>>>> be committed for the subscribed list of partitions. One improvement >>>>> could >>>>>>>> be to >>>>>>>> explicitly state that the offsets returned on the last poll will be >>>>>>>> committed. I updated this to - >>>>>>>> >>>>>>>> Synchronously commits the specified offsets for the specified list >> of >>>>>>>> topics and partitions to *Kafka*. If no offsets are specified, >>>> commits >>>>>>>> offsets returned on the last {@link #poll(long, TimeUnit) poll()} >> for >>>>>>>> the subscribed list of topics and partitions. >>>>>>>> >>>>>>>> 4. There is inconsistency in specifying partitions. Sometimes we use >>>>>>>> TopicPartition and some other times we use String and int (see >>>>>>>> examples below). >>>>>>>> >>>>>>>> void onPartitionsAssigned(Consumer consumer, >>>>>>>> TopicPartition...partitions) >>>>>>>> >>>>>>>> public void *subscribe*(java.lang.String topic, int... partitions) >>>>>>>> >>>>>>>> Yes, this was discussed previously. I think generally the consensus >>>>>>>> seems to be to use the higher level >>>>>>>> classes everywhere. Made those changes. >>>>>>>> >>>>>>>> What's the use case of position()? Isn't that just the nextOffset() >>>> on >>>>>>>> the >>>>>>>> last message returned from poll()? >>>>>>>> >>>>>>>> Yes, except in the case where a rebalance is triggered and poll() is >>>>> not >>>>>>>> yet invoked. Here, you would use position() to get the new fetch >>>>> position >>>>>>>> for the specific partition. Even if this is not a common use case, >>>> IMO >>>>> it >>>>>>>> is much easier to use position() to get the fetch offset than >>>> invoking >>>>>>>> nextOffset() on the last message. This also keeps the APIs >> symmetric, >>>>> which >>>>>>>> is nice. >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> On Mon, Feb 24, 2014 at 7:06 PM, Withers, Robert < >>>>>>>> robert.with...@dish.com> wrote: >>>>>>>> >>>>>>>>> That's wonderful. Thanks for kafka. >>>>>>>>> >>>>>>>>> Rob >>>>>>>>> >>>>>>>>> On Feb 24, 2014, at 9:58 AM, Guozhang Wang <wangg...@gmail.com >>>>> <mailto: >>>>>>>>> wangg...@gmail.com>> wrote: >>>>>>>>> >>>>>>>>> Hi Robert, >>>>>>>>> >>>>>>>>> Yes, you can check out the callback functions in the new API >>>>>>>>> >>>>>>>>> onPartitionDesigned >>>>>>>>> onPartitionAssigned >>>>>>>>> >>>>>>>>> and see if they meet your needs. >>>>>>>>> >>>>>>>>> Guozhang >>>>>>>>> >>>>>>>>> >>>>>>>>> On Mon, Feb 24, 2014 at 8:18 AM, Withers, Robert < >>>>>>>>> robert.with...@dish.com<mailto:robert.with...@dish.com>>wrote: >>>>>>>>> >>>>>>>>> Jun, >>>>>>>>> >>>>>>>>> Are you saying it is possible to get events from the high-level >>>>> consumer >>>>>>>>> regarding various state machine changes? For instance, can we get >> a >>>>>>>>> notification when a rebalance starts and ends, when a partition is >>>>>>>>> assigned/unassigned, when an offset is committed on a partition, >>>> when >>>>> a >>>>>>>>> leader changes and so on? I call this OOB traffic, since they are >>>> not >>>>>>>>> the >>>>>>>>> core messages streaming, but side-band events, yet they are still >>>>>>>>> potentially useful to consumers. >>>>>>>>> >>>>>>>>> Thank you, >>>>>>>>> Robert >>>>>>>>> >>>>>>>>> >>>>>>>>> Robert Withers >>>>>>>>> Staff Analyst/Developer >>>>>>>>> o: (720) 514-8963 >>>>>>>>> c: (571) 262-1873 >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> -----Original Message----- >>>>>>>>> From: Jun Rao [mailto:jun...@gmail.com] >>>>>>>>> Sent: Sunday, February 23, 2014 4:19 PM >>>>>>>>> To: users@kafka.apache.org<mailto:users@kafka.apache.org> >>>>>>>>> Subject: Re: New Consumer API discussion >>>>>>>>> >>>>>>>>> Robert, >>>>>>>>> >>>>>>>>> For the push orient api, you can potentially implement your own >>>>>>>>> MessageHandler with those methods. In the main loop of our new >>>>> consumer >>>>>>>>> api, you can just call those methods based on the events you get. >>>>>>>>> >>>>>>>>> Also, we already have an api to get the first and the last offset >>>> of a >>>>>>>>> partition (getOffsetBefore). >>>>>>>>> >>>>>>>>> Thanks, >>>>>>>>> >>>>>>>>> Jun >>>>>>>>> >>>>>>>>> >>>>>>>>> On Sat, Feb 22, 2014 at 11:29 AM, Withers, Robert >>>>>>>>> <robert.with...@dish.com<mailto:robert.with...@dish.com>>wrote: >>>>>>>>> >>>>>>>>> This is a good idea, too. I would modify it to include stream >>>>>>>>> marking, then you can have: >>>>>>>>> >>>>>>>>> long end = consumer.lastOffset(tp); >>>>>>>>> consumer.setMark(end); >>>>>>>>> while(consumer.beforeMark()) { >>>>>>>>> process(consumer.pollToMark()); >>>>>>>>> } >>>>>>>>> >>>>>>>>> or >>>>>>>>> >>>>>>>>> long end = consumer.lastOffset(tp); >>>>>>>>> consumer.setMark(end); >>>>>>>>> for(Object msg : consumer.iteratorToMark()) { >>>>>>>>> process(msg); >>>>>>>>> } >>>>>>>>> >>>>>>>>> I actually have 4 suggestions, then: >>>>>>>>> >>>>>>>>> * pull: stream marking >>>>>>>>> * pull: finite streams, bound by time range (up-to-now, >> yesterday) >>>>> or >>>>>>>>> offset >>>>>>>>> * pull: async api >>>>>>>>> * push: KafkaMessageSource, for a push model, with msg and OOB >>>>> events. >>>>>>>>> Build one in either individual or chunk mode and have a listener >> for >>>>>>>>> each msg or a listener for a chunk of msgs. Make it composable and >>>>>>>>> policy driven (chunked, range, commitOffsets policy, retry policy, >>>>>>>>> transactional) >>>>>>>>> >>>>>>>>> Thank you, >>>>>>>>> Robert >>>>>>>>> >>>>>>>>> On Feb 22, 2014, at 11:21 AM, Jay Kreps <jay.kr...@gmail.com >>>> <mailto: >>>>>>>>> jay.kr...@gmail.com><mailto: >>>>>>>>> jay.kr...@gmail.com<mailto:jay.kr...@gmail.com>>> wrote: >>>>>>>>> >>>>>>>>> I think what Robert is saying is that we need to think through the >>>>>>>>> offset API to enable "batch processing" of topic data. Think of a >>>>>>>>> process that periodically kicks off to compute a data summary or do >>>> a >>>>>>>>> data load or something like that. I think what we need to support >>>> this >>>>>>>>> is an api to fetch the last offset from the server for a partition. >>>>>>>>> Something like >>>>>>>>> long lastOffset(TopicPartition tp) >>>>>>>>> and for symmetry >>>>>>>>> long firstOffset(TopicPartition tp) >>>>>>>>> >>>>>>>>> Likely this would have to be batched. Essentially we should add >> this >>>>>>>>> use case to our set of code examples to write and think through. >>>>>>>>> >>>>>>>>> The usage would be something like >>>>>>>>> >>>>>>>>> long end = consumer.lastOffset(tp); >>>>>>>>> while(consumer.position < end) >>>>>>>>> process(consumer.poll()); >>>>>>>>> >>>>>>>>> -Jay >>>>>>>>> >>>>>>>>> >>>>>>>>> On Sat, Feb 22, 2014 at 1:52 AM, Withers, Robert >>>>>>>>> <robert.with...@dish.com<mailto:robert.with...@dish.com> >>>>>>>>> <mailto:robert.with...@dish.com>>wrote: >>>>>>>>> >>>>>>>>> Jun, >>>>>>>>> >>>>>>>>> I was originally thinking a non-blocking read from a distributed >>>>>>>>> stream should distinguish between "no local messages, but a fetch >> is >>>>>>>>> occurring" >>>>>>>>> versus "you have drained the stream". The reason this may be >>>> valuable >>>>>>>>> to me is so I can write consumers that read all known traffic then >>>>>>>>> terminate. >>>>>>>>> You caused me to reconsider and I think I am conflating 2 things. >>>> One >>>>>>>>> is a sync/async api while the other is whether to have an infinite >>>> or >>>>>>>>> finite stream. Is it possible to build a finite KafkaStream on a >>>>>>>>> range of messages? >>>>>>>>> >>>>>>>>> Perhaps a Simple Consumer would do just fine and then I could start >>>>>>>>> off getting the writeOffset from zookeeper and tell it to read a >>>>>>>>> specified range per partition. I've done this and forked a simple >>>>>>>>> consumer runnable for each partition, for one of our analyzers. >> The >>>>>>>>> great thing about the high-level consumer is that rebalance, so I >>>> can >>>>>>>>> fork however many stream readers I want and you just figure it out >>>> for >>>>>>>>> me. In that way you offer us the control over the resource >>>>>>>>> consumption within a pull model. This is best to regulate message >>>>>>>>> pressure, they say. >>>>>>>>> >>>>>>>>> Combining that high-level rebalance ability with a ranged partition >>>>>>>>> drain could be really nice...build the stream with an ending >>>> position >>>>>>>>> and it is a finite stream, but retain the high-level rebalance. >>>> With >>>>>>>>> a finite stream, you would know the difference of the 2 async >>>>>>>>> scenarios: fetch-in-progress versus end-of-stream. With an >> infinite >>>>>>>>> stream, you never get end-of-stream. >>>>>>>>> >>>>>>>>> Aside from a high-level consumer over a finite range within each >>>>>>>>> partition, the other feature I can think of is more complicated. A >>>>>>>>> high-level consumer has state machine changes that the client >> cannot >>>>>>>>> access, to my knowledge. Our use of kafka has us invoke a message >>>>>>>>> handler with each message we consumer from the KafkaStream, so we >>>>>>>>> convert a pull-model to a push-model. Including the idea of >>>> receiving >>>>>>>>> notifications from state machine changes, what would be really nice >>>> is >>>>>>>>> to have a KafkaMessageSource, that is an eventful push model. If >> it >>>>>>>>> were thread-safe, then we could register listeners for various >>>> events: >>>>>>>>> >>>>>>>>> * opening-stream >>>>>>>>> * closing-stream >>>>>>>>> * message-arrived >>>>>>>>> * end-of-stream/no-more-messages-in-partition (for finite >> streams) >>>>>>>>> * rebalance started >>>>>>>>> * partition assigned >>>>>>>>> * partition unassigned >>>>>>>>> * rebalance finished >>>>>>>>> * partition-offset-committed >>>>>>>>> >>>>>>>>> Perhaps that is just our use, but instead of a pull-oriented >>>>>>>>> KafkaStream, is there any sense in your providing a push-oriented >>>>>>>>> KafkaMessageSource publishing OOB messages? >>>>>>>>> >>>>>>>>> thank you, >>>>>>>>> Robert >>>>>>>>> >>>>>>>>> On Feb 21, 2014, at 5:59 PM, Jun Rao <jun...@gmail.com<mailto: >>>>>>>>> jun...@gmail.com><mailto: >>>>>>>>> jun...@gmail.com<mailto:jun...@gmail.com>><mailto: >>>>>>>>> jun...@gmail.com<mailto:jun...@gmail.com><mailto:jun...@gmail.com >>>>>>> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>> Robert, >>>>>>>>> >>>>>>>>> Could you explain why you want to distinguish btw >>>>>>>>> FetchingInProgressException and NoMessagePendingException? The >>>>>>>>> nextMsgs() method that you want is exactly what poll() does. >>>>>>>>> >>>>>>>>> Thanks, >>>>>>>>> >>>>>>>>> Jun >>>>>>>>> >>>>>>>>> >>>>>>>>> On Wed, Feb 19, 2014 at 8:45 AM, Withers, Robert >>>>>>>>> <robert.with...@dish.com<mailto:robert.with...@dish.com> <mailto: >>>>>>>>> robert.with...@dish.com> >>>>>>>>> <mailto:robert.with...@dish.com>>wrote: >>>>>>>>> >>>>>>>>> I am not clear on why the consumer stream should be positionable, >>>>>>>>> especially if it is limited to the in-memory fetched messages. >>>> Could >>>>>>>>> someone explain to me, please? I really like the idea of >> committing >>>>>>>>> the offset specifically on those partitions with changed read >>>> offsets, >>>>>>>>> only. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> 2 items I would like to see added to the KafkaStream are: >>>>>>>>> >>>>>>>>> * a non-blocking next(), throws several exceptions >>>>>>>>> (FetchingInProgressException and a NoMessagePendingException or >>>>>>>>> something) to differentiate between fetching or no messages left. >>>>>>>>> >>>>>>>>> * A nextMsgs() method which returns all locally available >>>>>>>>> messages >>>>>>>>> and kicks off a fetch for the next chunk. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> If you are trying to add transactional features, then formally >>>> define >>>>>>>>> a DTP capability and pull in other server frameworks to share the >>>>>>>>> implementation. Should it be XA/Open? How about a new peer2peer >>>> DTP >>>>>>>>> protocol? >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> Thank you, >>>>>>>>> >>>>>>>>> Robert >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> Robert Withers >>>>>>>>> >>>>>>>>> Staff Analyst/Developer >>>>>>>>> >>>>>>>>> o: (720) 514-8963 >>>>>>>>> >>>>>>>>> c: (571) 262-1873 >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> -----Original Message----- >>>>>>>>> From: Jay Kreps [mailto:jay.kr...@gmail.com] >>>>>>>>> Sent: Sunday, February 16, 2014 10:13 AM >>>>>>>>> To: users@kafka.apache.org<mailto:users@kafka.apache.org><mailto: >>>>>>>>> users@kafka.apache.org><mailto: >>>>>>>>> users@kafka.apache.org<mailto:users@kafka.apache.org>> >>>>>>>>> Subject: Re: New Consumer API discussion >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> +1 I think those are good. It is a little weird that changing the >>>>>>>>> +fetch >>>>>>>>> >>>>>>>>> point is not batched but changing the commit point is, but I >> suppose >>>>>>>>> there is no helping that. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> -Jay >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> On Sat, Feb 15, 2014 at 7:52 AM, Neha Narkhede >>>>>>>>> <neha.narkh...@gmail.com<mailto:neha.narkh...@gmail.com> <mailto: >>>>>>>>> neha.narkh...@gmail.com> >>>>>>>>> <mailto:neha.narkh...@gmail.com> >>>>>>>>> <mailto:neha.narkh...@gmail.com>>wrote: >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> Jay, >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> That makes sense. position/seek deal with changing the consumers >>>>>>>>> >>>>>>>>> in-memory data, so there is no remote rpc there. For some reason, I >>>>>>>>> >>>>>>>>> got committed and seek mixed up in my head at that time :) >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> So we still end up with >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> long position(TopicPartition tp) >>>>>>>>> >>>>>>>>> void seek(TopicPartitionOffset p) >>>>>>>>> >>>>>>>>> Map<TopicPartition, Long> committed(TopicPartition tp); >>>>>>>>> >>>>>>>>> void commit(TopicPartitionOffset...); >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> Thanks, >>>>>>>>> >>>>>>>>> Neha >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> On Friday, February 14, 2014, Jay Kreps <jay.kr...@gmail.com >>>> <mailto: >>>>>>>>> jay.kr...@gmail.com><mailto: >>>>>>>>> jay.kr...@gmail.com<mailto:jay.kr...@gmail.com>><mailto: >>>>>>>>> jay.kr...@gmail.com<mailto:jay.kr...@gmail.com><mailto: >>>>>>>>> jay.kr...@gmail.com>><mailto: >>>>>>>>> jay.kr...@gmail.com<mailto:jay.kr...@gmail.com><mailto: >>>>>>>>> jay.kr...@gmail.com><mailto:jay.kreps@gmail >>>>>>>>> .com>>> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> Oh, interesting. So I am assuming the following implementation: >>>>>>>>> >>>>>>>>> 1. We have an in-memory fetch position which controls the next >> fetch >>>>>>>>> >>>>>>>>> offset. >>>>>>>>> >>>>>>>>> 2. Changing this has no effect until you poll again at which point >>>>>>>>> >>>>>>>>> your fetch request will be from the newly specified offset 3. We >>>>>>>>> >>>>>>>>> then have an in-memory but also remotely stored committed offset. >>>>>>>>> >>>>>>>>> 4. Calling commit has the effect of saving the fetch position as >>>>>>>>> >>>>>>>>> both the in memory committed position and in the remote store 5. >>>>>>>>> >>>>>>>>> Auto-commit is the same as periodically calling commit on all >>>>>>>>> >>>>>>>>> positions. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> So batching on commit as well as getting the committed position >>>>>>>>> >>>>>>>>> makes sense, but batching the fetch position wouldn't, right? I >>>>>>>>> >>>>>>>>> think you are actually thinking of a different approach. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> -Jay >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> On Thu, Feb 13, 2014 at 10:40 PM, Neha Narkhede >>>>>>>>> >>>>>>>>> <neha.narkh...@gmail.com<mailto:neha.narkh...@gmail.com><mailto: >>>>>>>>> neha.narkh...@gmail.com><mailto: >>>>>>>>> neha.narkh...@gmail.com<mailto:neha.narkh...@gmail.com>> >>>>>>>>> >>>>>>>>> <javascript:;> >>>>>>>>> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> I think you are saying both, i.e. if you have committed on a >>>>>>>>> >>>>>>>>> partition it returns you that value but if you >>>>>>>>> >>>>>>>>> haven't >>>>>>>>> >>>>>>>>> it does a remote lookup? >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> Correct. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> The other argument for making committed batched is that commit() >>>>>>>>> >>>>>>>>> is batched, so there is symmetry. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> position() and seek() are always in memory changes (I assume) so >>>>>>>>> >>>>>>>>> there >>>>>>>>> >>>>>>>>> is >>>>>>>>> >>>>>>>>> no need to batch them. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> I'm not as sure as you are about that assumption being true. >>>>>>>>> >>>>>>>>> Basically >>>>>>>>> >>>>>>>>> in >>>>>>>>> >>>>>>>>> my example above, the batching argument for committed() also >>>>>>>>> >>>>>>>>> applies to >>>>>>>>> >>>>>>>>> position() since one purpose of fetching a partition's offset is >>>>>>>>> >>>>>>>>> to use >>>>>>>>> >>>>>>>>> it >>>>>>>>> >>>>>>>>> to set the position of the consumer to that offset. Since that >>>>>>>>> >>>>>>>>> might >>>>>>>>> >>>>>>>>> lead >>>>>>>>> >>>>>>>>> to a remote OffsetRequest call, I think we probably would be >>>>>>>>> >>>>>>>>> better off batching it. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> Another option for naming would be position/reposition instead of >>>>>>>>> >>>>>>>>> position/seek. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> I think position/seek is better since it aligns with Java file >> APIs. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> I also think your suggestion about ConsumerPosition makes sense. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> Thanks, >>>>>>>>> >>>>>>>>> Neha >>>>>>>>> >>>>>>>>> On Feb 13, 2014 9:22 PM, "Jay Kreps" <jay.kr...@gmail.com<mailto: >>>>>>>>> jay.kr...@gmail.com><mailto: >>>>>>>>> jay.kr...@gmail.com<mailto:jay.kr...@gmail.com>><mailto: >>>>>>>>> jay.kr...@gmail.com<mailto:jay.kr...@gmail.com><mailto: >>>>>>>>> jay.kr...@gmail.com>><mailto: >>>>>>>>> jay.kr...@gmail.com<mailto:jay.kr...@gmail.com><mailto: >>>>>>>>> jay.kr...@gmail.com><mailto:jay.kreps@gmail >>>>>>>>> .com>>> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> Hey Neha, >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> I actually wasn't proposing the name TopicOffsetPosition, that >>>>>>>>> >>>>>>>>> was >>>>>>>>> >>>>>>>>> just a >>>>>>>>> >>>>>>>>> typo. I meant TopicPartitionOffset, and I was just referencing >>>>>>>>> >>>>>>>>> what >>>>>>>>> >>>>>>>>> was >>>>>>>>> >>>>>>>>> in >>>>>>>>> >>>>>>>>> the javadoc. So to restate my proposal without the typo, using >>>>>>>>> >>>>>>>>> just >>>>>>>>> >>>>>>>>> the >>>>>>>>> >>>>>>>>> existing classes (that naming is a separate question): >>>>>>>>> >>>>>>>>> long position(TopicPartition tp) >>>>>>>>> >>>>>>>>> void seek(TopicPartitionOffset p) >>>>>>>>> >>>>>>>>> long committed(TopicPartition tp) >>>>>>>>> >>>>>>>>> void commit(TopicPartitionOffset...); >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> So I may be unclear on committed() (AKA lastCommittedOffset). Is >>>>>>>>> >>>>>>>>> it returning the in-memory value from the last commit by this >>>>>>>>> >>>>>>>>> consumer, >>>>>>>>> >>>>>>>>> or >>>>>>>>> >>>>>>>>> is >>>>>>>>> >>>>>>>>> it doing a remote fetch, or both? I think you are saying both, i.e. >>>>>>>>> >>>>>>>>> if >>>>>>>>> >>>>>>>>> you >>>>>>>>> >>>>>>>>> have committed on a partition it returns you that value but if >>>>>>>>> >>>>>>>>> you >>>>>>>>> >>>>>>>>> haven't >>>>>>>>> >>>>>>>>> it does a remote lookup? >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> The other argument for making committed batched is that commit() >>>>>>>>> >>>>>>>>> is batched, so there is symmetry. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> position() and seek() are always in memory changes (I assume) so >>>>>>>>> >>>>>>>>> there >>>>>>>>> >>>>>>>>> is >>>>>>>>> >>>>>>>>> no need to batch them. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> So taking all that into account what if we revise it to >>>>>>>>> >>>>>>>>> long position(TopicPartition tp) >>>>>>>>> >>>>>>>>> void seek(TopicPartitionOffset p) >>>>>>>>> >>>>>>>>> Map<TopicPartition, Long> committed(TopicPartition tp); >>>>>>>>> >>>>>>>>> void commit(TopicPartitionOffset...); >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> This is not symmetric between position/seek and commit/committed >>>>>>>>> >>>>>>>>> but >>>>>>>>> >>>>>>>>> it >>>>>>>>> >>>>>>>>> is >>>>>>>>> >>>>>>>>> convenient. Another option for naming would be >>>>>>>>> >>>>>>>>> position/reposition >>>>>>>>> >>>>>>>>> instead >>>>>>>>> >>>>>>>>> of position/seek. >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> With respect to the name TopicPartitionOffset, what I was trying >>>>>>>>> >>>>>>>>> to >>>>>>>>> >>>>>>>>> say >>>>>>>>> >>>>>>>>> is >>>>>>>>> >>>>>>>>> that I recommend we change that to something shorter. I think >>>>>>>>> >>>>>>>>> TopicPosition >>>>>>>>> >>>>>>>>> or ConsumerPosition might be better. Position does not refer to >>>>>>>>> >>>>>>>>> the variables in the object, it refers to the meaning of the >>>>>>>>> >>>>>>>>> object--it represents a position within a topic. The offset >>>>>>>>> >>>>>>>>> field in that object >>>>>>>>> >>>>>>>>> is >>>>>>>>> >>>>>>>>> still called the offset. TopicOffset, PartitionOffset, or >>>>>>>>> >>>>>>>>> ConsumerOffset >>>>>>>>> >>>>>>>>> would all be workable too. Basically I am just objecting to >>>>>>>>> >>>>>>>>> concatenating >>>>>>>>> >>>>>>>>> three nouns together. :-) >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> -Jay >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> On Thu, Feb 13, 2014 at 1:54 PM, Neha Narkhede < >>>>>>>>> >>>>>>>>> neha.narkh...@gmail.com<mailto:neha.narkh...@gmail.com><mailto: >>>>>>>>> neha.narkh...@gmail.com><mailto: >>>>>>>>> neha.narkh...@gmail.com<mailto:neha.narkh...@gmail.com>><mailto: >>>>>>>>> neha.narkh...@gmail.com<mailto:neha.narkh...@gmail.com><mailto: >>>>>>>>> neha.narkh...@gmail.com>> >>>>>>>>> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> 2. It returns a list of results. But how can you use the list? >>>>>>>>> >>>>>>>>> The >>>>>>>>> >>>>>>>>> only >>>>>>>>> >>>>>>>>> way >>>>>>>>> >>>>>>>>> to use the list is to make a map of tp=>offset and then look >>>>>>>>> >>>>>>>>> up >>>>>>>>> >>>>>>>>> results >>>>>>>>> >>>>>>>>> in >>>>>>>>> >>>>>>>>> this map (or do a for loop over the list for the partition you >>>>>>>>> >>>>>>>>> want). I >>>>>>>>> >>>>>>>>> recommend that if this is an in-memory check we just do one at >>>>>>>>> >>>>>>>>> a >>>>>>>>> >>>>>>>>> time. >>>>>>>>> >>>>>>>>> E.g. >>>>>>>>> >>>>>>>>> long committedPosition( >>>>>>>>> >>>>>>>>> TopicPosition). >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> This was discussed in the previous emails. There is a choic >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> -- >>>>>>>>> Robert Withers >>>>>>>>> robert.with...@dish.com<mailto:robert.with...@dish.com><mailto: >>>>>>>>> robert.with...@dish.com><mailto: >>>>>>>>> robert.with...@dish.com<mailto:robert.with...@dish.com>> >>>>>>>>> c: 303.919.5856 >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> -- >>>>>>>>> Robert Withers >>>>>>>>> robert.with...@dish.com<mailto:robert.with...@dish.com><mailto: >>>>>>>>> robert.with...@dish.com> >>>>>>>>> c: 303.919.5856 >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> -- >>>>>>>>> -- Guozhang >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> -- >>>>>>>>> Robert Withers >>>>>>>>> robert.with...@dish.com<mailto:robert.with...@dish.com> >>>>>>>>> c: 303.919.5856 >>>>>>>>> >>>>>>>>> >>>>>>>> >>>>>>> >>>>>> >>>>> >>>> >> >>
