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 >>>>>>> >>>>>>> >>>>>> >>>>> >>>> >>> >>
smime.p7s
Description: S/MIME cryptographic signature
