Jack, Could you just change "auto.offset.reset" to smallest and see if this issue goes away? It is not related to the producer end.
Guozhang On Mon, Apr 6, 2015 at 4:14 PM, Jack <jac...@gmail.com> wrote: > Hi Guozhang, > > Thanks so much for replying, first of all. > > Here is the config we have: > > group.id -> 'some unique id' > zookeeper.connect -> 'zookeeper host' > auto.commit.enabled -> false > 'auto.offset.reset' -> largest > consumer.timeout.ms -> -1 > fetch.message.max.bytes -> 10M > > So it seems like we need to make sure the submitted future returns before > performing action actions which eventually generate the message we expect. > > Cheers, > > -Jack > > > > On Mon, Apr 6, 2015 at 4:04 PM, Guozhang Wang <wangg...@gmail.com> wrote: > > > Jack, > > > > Your theory is correct if your consumer config set auto.offset.reset to > > latest and you do not have any committed offsets before. Could you list > > your consumer configs and see if that is the case? > > > > Guozhang > > > > On Mon, Apr 6, 2015 at 3:15 PM, Jack <jac...@gmail.com> wrote: > > > > > Hi folks, > > > > > > I have a quick question. > > > > > > We are using 0.8.1 and running into this weird problem. We are using > > > HighLevelConsumer for this topic. We created 64 partitions for this > > > message. > > > > > > In our service, we first create a Consumer object as usual, and then we > > > went ahead, calls 'createMessageStreans' with Map('topic_name'->64). It > > > returns us a Se[KafkaStream], For each stream object in the sequence, > we > > > submit a task like the following to the pool. > > > > > > threadpool.submit(new Runnable { > > > override def run() = { > > > stream.iterator().foreach { msg => ...} > > > } > > > } > > > > > > The problem we ran into is that after all the above established, any > > > message showing up in kafka, we should be able to get it from consumer > > > side. But in reality, for some reason, occasionally, we don't see these > > > message (we do see these message in the log though). > > > > > > Some team members believe that the stream might get a later offset, > thus > > > not being able to see the earlier messages. > > > > > > I really doubt that statement and want to see if anyone could shed any > > > light upon this? > > > > > > One possible theory from me is that the offset won't be given until > > > stream.iterator().next is called, but since the task submission is > > > asynchronous (we don't wait for each submission and then produce > message > > to > > > kafka), that could get us a later offset, which might not contains the > > > message we want). One possible solution to that is perform any action > > which > > > produce messages to kafka, after all these submitted tasks returns. > > > > > > Any thoughts? > > > > > > Thanks, > > > > > > -Jack > > > > > > > > > > > -- > > -- Guozhang > > > -- -- Guozhang