You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by Bhavesh Mistry <mi...@gmail.com> on 2014/09/03 01:38:15 UTC
High Level Consumer and Commit
Hi Kafka Group,
I have to pull the data from the Topic and index into Elastic Search with
Bulk API and wanted to commit only batch that has been committed and still
continue to read from topic further on same topic. I have auto commit to
be off.
List<Message> batch .....
while (iterator.hasNext()) {
batch.add(iterator.next().message());
if(batch size is 50 ){
//===>>>> Once the bulk API is successful it will commit the offset
to zookeeper...
executor.submit(new Thread() process batch and commit batch,
cconsumerConnector)
batch = new batch buffer....
}
}
This commitOffset API commits all messages that have been read so far.
What is best way to continue reading and only commit another thread finish
batch process is successful. This will lead to fragmentation of the
Consumer offset so what is best way to implement continuous reading stream
and commit the rage offset.
Is Simple Consumer a better approach for this.
Thanks,
Bhavesh
Thanks,
Bhavesh
Re: Manual Leader Assignment
Posted by Jun Rao <ju...@gmail.com>.
With ack=1, some acked msgs could be lost when the leader of a partition
changes.
Thanks,
Jun
On Tue, Sep 9, 2014 at 6:22 AM, Andrew Otto <ao...@wikimedia.org> wrote:
> > (2)
> > configure enough retries + backoff time in the producer (so that new
> > leaders can be elected during failure).
> Ya, I’ve been doing this, I need to get it just right.
>
> > (1) use ack=-1 in the producer;
> We use ack=1. If I configure (2) properly, I shouldn’t need ack=-1, will
> I? I don’t think this will help me in my case anyway, as it is the
> producers themselves that are dropping the messages when their buffers fill
> up, while they are waiting for acks on messages before the new leadership
> metadata has been propagated.
>
> Anyway, my question is about being able to test if I have done (2)
> properly. Right now
>
>
>
> On Sep 9, 2014, at 12:10 AM, Jun Rao <ju...@gmail.com> wrote:
>
> > To avoid data loss, you will need to (1) use ack=-1 in the producer; (2)
> > configure enough retries + backoff time in the producer (so that new
> > leaders can be elected during failure).
> >
> > For controlled failure, you can reduce the unavailability window by using
> > controlled shutdown. See
> > http://kafka.apache.org/documentation.html#basic_ops_restarting
> >
> > Thanks,
> >
> > Jun
> >
> > On Wed, Sep 3, 2014 at 11:48 AM, Andrew Otto <ao...@wikimedia.org>
> wrote:
> >
> >> Hiya,
> >>
> >> During leader changes, we see short periods of message loss on some of
> our
> >> higher volume producers. I suspect that this is because it takes a
> couple
> >> of seconds for Zookeeper to notice and notify the producers of the
> metadata
> >> change. During this time, producer buffers can fill up and end up
> dropping
> >> some messages.
> >>
> >> I’d like to do some troubleshooting. Is it possible to manually change
> >> the leadership of a single partition? I see here[1] that I can start a
> >> leadership election for a particular partition, but the JSON doesn’t
> show a
> >> way to choose the new leader of the partition.
> >>
> >> Thanks!
> >> -Andrew Otto
> >>
> >> [1]
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/Replication+tools#Replicationtools-Howtousethetool?.1
>
>
Re: Manual Leader Assignment
Posted by Andrew Otto <ao...@wikimedia.org>.
> (2)
> configure enough retries + backoff time in the producer (so that new
> leaders can be elected during failure).
Ya, I’ve been doing this, I need to get it just right.
> (1) use ack=-1 in the producer;
We use ack=1. If I configure (2) properly, I shouldn’t need ack=-1, will I? I don’t think this will help me in my case anyway, as it is the producers themselves that are dropping the messages when their buffers fill up, while they are waiting for acks on messages before the new leadership metadata has been propagated.
Anyway, my question is about being able to test if I have done (2) properly. Right now
On Sep 9, 2014, at 12:10 AM, Jun Rao <ju...@gmail.com> wrote:
> To avoid data loss, you will need to (1) use ack=-1 in the producer; (2)
> configure enough retries + backoff time in the producer (so that new
> leaders can be elected during failure).
>
> For controlled failure, you can reduce the unavailability window by using
> controlled shutdown. See
> http://kafka.apache.org/documentation.html#basic_ops_restarting
>
> Thanks,
>
> Jun
>
> On Wed, Sep 3, 2014 at 11:48 AM, Andrew Otto <ao...@wikimedia.org> wrote:
>
>> Hiya,
>>
>> During leader changes, we see short periods of message loss on some of our
>> higher volume producers. I suspect that this is because it takes a couple
>> of seconds for Zookeeper to notice and notify the producers of the metadata
>> change. During this time, producer buffers can fill up and end up dropping
>> some messages.
>>
>> I’d like to do some troubleshooting. Is it possible to manually change
>> the leadership of a single partition? I see here[1] that I can start a
>> leadership election for a particular partition, but the JSON doesn’t show a
>> way to choose the new leader of the partition.
>>
>> Thanks!
>> -Andrew Otto
>>
>> [1]
>> https://cwiki.apache.org/confluence/display/KAFKA/Replication+tools#Replicationtools-Howtousethetool?.1
Re: Manual Leader Assignment
Posted by Jun Rao <ju...@gmail.com>.
To avoid data loss, you will need to (1) use ack=-1 in the producer; (2)
configure enough retries + backoff time in the producer (so that new
leaders can be elected during failure).
For controlled failure, you can reduce the unavailability window by using
controlled shutdown. See
http://kafka.apache.org/documentation.html#basic_ops_restarting
Thanks,
Jun
On Wed, Sep 3, 2014 at 11:48 AM, Andrew Otto <ao...@wikimedia.org> wrote:
> Hiya,
>
> During leader changes, we see short periods of message loss on some of our
> higher volume producers. I suspect that this is because it takes a couple
> of seconds for Zookeeper to notice and notify the producers of the metadata
> change. During this time, producer buffers can fill up and end up dropping
> some messages.
>
> I’d like to do some troubleshooting. Is it possible to manually change
> the leadership of a single partition? I see here[1] that I can start a
> leadership election for a particular partition, but the JSON doesn’t show a
> way to choose the new leader of the partition.
>
> Thanks!
> -Andrew Otto
>
> [1]
> https://cwiki.apache.org/confluence/display/KAFKA/Replication+tools#Replicationtools-Howtousethetool?.1
Manual Leader Assignment
Posted by Andrew Otto <ao...@wikimedia.org>.
Hiya,
During leader changes, we see short periods of message loss on some of our higher volume producers. I suspect that this is because it takes a couple of seconds for Zookeeper to notice and notify the producers of the metadata change. During this time, producer buffers can fill up and end up dropping some messages.
I’d like to do some troubleshooting. Is it possible to manually change the leadership of a single partition? I see here[1] that I can start a leadership election for a particular partition, but the JSON doesn’t show a way to choose the new leader of the partition.
Thanks!
-Andrew Otto
[1] https://cwiki.apache.org/confluence/display/KAFKA/Replication+tools#Replicationtools-Howtousethetool?.1
Re: High Level Consumer and Commit
Posted by Gwen Shapira <gs...@cloudera.com>.
Sorry, I guess I missed that. The followup discussion was around the
simple consumer :)
I'm not sure why the OP didn't find this solution acceptable.
On Wed, Sep 3, 2014 at 8:29 AM, Philip O'Toole <ph...@yahoo.com> wrote:
> That's what I said in my first reply. :-)
>
> -----------------------------------------
> http://www.philipotoole.com
>
>
> On Tuesday, September 2, 2014 10:37 PM, Gwen Shapira <gs...@cloudera.com>
> wrote:
>
>
> I believe a simpler solution would be to create multiple
> ConsumerConnector, each with 1 thread (single ConsumerStream) and use
> commitOffset API to commit all partitions managed by each
> ConsumerConnector after the thread finished processing the messages.
>
> Does that solve the problem, Bhavesh?
>
> Gwen
>
> On Tue, Sep 2, 2014 at 5:47 PM, Philip O'Toole
> <ph...@yahoo.com.invalid> wrote:
>> Yeah, from reading that I suspect you need the SimpleConsumer. Try it out
>> and see.
>>
>> Philip
>>
>>
>> -----------------------------------------
>> http://www.philipotoole.com
>>
>>
>> On Tuesday, September 2, 2014 5:43 PM, Bhavesh Mistry
>> <mi...@gmail.com> wrote:
>>
>>
>>
>> Hi Philip,
>>
>> Yes, We have disabled auto commit but, we need to be able to read from
>> particular offset if we manage the offset ourself in some storage(DB).
>> High Level consumer does not allow per partition management plug-ability.
>>
>> I like to have the High Level consumers Failover and auto rebalancing. We
>> just need plug ability of offset management.
>>
>> Thanks,
>>
>> Bhavesh
>>
>>
>>
>> On Tue, Sep 2, 2014 at 5:20 PM, Philip O'Toole <
>> philip.otoole@yahoo.com.invalid> wrote:
>>
>>> No, you'll need to write your own failover.
>>>
>>> I'm not sure I follow your second question, but the high-level Consumer
>>> should be able to do what you want if you disable auto-commit. I'm not
>>> sure
>>> what else you're asking.
>>>
>>>
>>> Philip
>>>
>>>
>>> -----------------------------------------
>>> http://www.philipotoole.com
>>>
>>>
>>> On Tuesday, September 2, 2014 5:15 PM, Bhavesh Mistry <
>>> mistry.p.bhavesh@gmail.com> wrote:
>>>
>>>
>>>
>>> Hi Philip,
>>>
>>> Thanks for the update. With Simple Consumer I will not get failover and
>>> rebalance that is provided out of box. what is other option not to block
>>> reading and keep processing and commit only when batch is done.
>>>
>>> Thanks,
>>>
>>> Bhavesh
>>>
>>>
>>>
>>> On Tue, Sep 2, 2014 at 4:43 PM, Philip O'Toole <
>>> philip.otoole@yahoo.com.invalid> wrote:
>>>
>>> > Either use the SimpleConsumer which gives you much finer-grained
>>> > control,
>>> > or (this worked with 0.7) spin up a ConsumerConnection (this is a
>>> HighLevel
>>> > consumer concept) per partition, turn off auto-commit.
>>> >
>>> > Philip
>>> >
>>> >
>>> > -----------------------------------------
>>> > http://www.philipotoole.com
>>> >
>>> >
>>> > On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <
>>> > mistry.p.bhavesh@gmail.com> wrote:
>>> >
>>> >
>>> >
>>> > Hi Kafka Group,
>>> >
>>> > I have to pull the data from the Topic and index into Elastic Search
>>> > with
>>> > Bulk API and wanted to commit only batch that has been committed and
>>> still
>>> > continue to read from topic further on same topic. I have auto commit
>>> > to
>>> > be off.
>>> >
>>> >
>>> > List<Message> batch .....
>>> >
>>> > while (iterator.hasNext()) {
>>> > batch.add(iterator.next().message());
>>> > if(batch size is 50 ){
>>> > //===>>>> Once the bulk API is successful it will commit the
>>> offset
>>> > to zookeeper...
>>> > executor.submit(new Thread() process batch and commit batch,
>>> > cconsumerConnector)
>>> > batch = new batch buffer....
>>> > }
>>> > }
>>> >
>>> > This commitOffset API commits all messages that have been read so far.
>>> > What is best way to continue reading and only commit another thread
>>> finish
>>> > batch process is successful. This will lead to fragmentation of the
>>> > Consumer offset so what is best way to implement continuous reading
>>> stream
>>> > and commit the rage offset.
>>> >
>>> > Is Simple Consumer a better approach for this.
>>> >
>>> >
>>> > Thanks,
>>> >
>>> > Bhavesh
>>> >
>>> >
>>> >
>>> >
>>> >
>>> >
>>> >
>>> > Thanks,
>>> > Bhavesh
>>> >
>>>
>
>
Re: High Level Consumer and Commit
Posted by Philip O'Toole <ph...@yahoo.com.INVALID>.
That's what I said in my first reply. :-)
-----------------------------------------
http://www.philipotoole.com
On Tuesday, September 2, 2014 10:37 PM, Gwen Shapira <gs...@cloudera.com> wrote:
I believe a simpler solution would be to create multiple
ConsumerConnector, each with 1 thread (single ConsumerStream) and use
commitOffset API to commit all partitions managed by each
ConsumerConnector after the thread finished processing the messages.
Does that solve the problem, Bhavesh?
Gwen
On Tue, Sep 2, 2014 at 5:47 PM, Philip O'Toole
<ph...@yahoo.com.invalid> wrote:
> Yeah, from reading that I suspect you need the SimpleConsumer. Try it out and see.
>
> Philip
>
>
> -----------------------------------------
> http://www.philipotoole.com
>
>
> On Tuesday, September 2, 2014 5:43 PM, Bhavesh Mistry <mi...@gmail.com> wrote:
>
>
>
> Hi Philip,
>
> Yes, We have disabled auto commit but, we need to be able to read from
> particular offset if we manage the offset ourself in some storage(DB).
> High Level consumer does not allow per partition management plug-ability.
>
> I like to have the High Level consumers Failover and auto rebalancing. We
> just need plug ability of offset management.
>
> Thanks,
>
> Bhavesh
>
>
>
> On Tue, Sep 2, 2014 at 5:20 PM, Philip O'Toole <
> philip.otoole@yahoo.com.invalid> wrote:
>
>> No, you'll need to write your own failover.
>>
>> I'm not sure I follow your second question, but the high-level Consumer
>> should be able to do what you want if you disable auto-commit. I'm not sure
>> what else you're asking.
>>
>>
>> Philip
>>
>>
>> -----------------------------------------
>> http://www.philipotoole.com
>>
>>
>> On Tuesday, September 2, 2014 5:15 PM, Bhavesh Mistry <
>> mistry.p.bhavesh@gmail.com> wrote:
>>
>>
>>
>> Hi Philip,
>>
>> Thanks for the update. With Simple Consumer I will not get failover and
>> rebalance that is provided out of box. what is other option not to block
>> reading and keep processing and commit only when batch is done.
>>
>> Thanks,
>>
>> Bhavesh
>>
>>
>>
>> On Tue, Sep 2, 2014 at 4:43 PM, Philip O'Toole <
>> philip.otoole@yahoo.com.invalid> wrote:
>>
>> > Either use the SimpleConsumer which gives you much finer-grained control,
>> > or (this worked with 0.7) spin up a ConsumerConnection (this is a
>> HighLevel
>> > consumer concept) per partition, turn off auto-commit.
>> >
>> > Philip
>> >
>> >
>> > -----------------------------------------
>> > http://www.philipotoole.com
>> >
>> >
>> > On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <
>> > mistry.p.bhavesh@gmail.com> wrote:
>> >
>> >
>> >
>> > Hi Kafka Group,
>> >
>> > I have to pull the data from the Topic and index into Elastic Search with
>> > Bulk API and wanted to commit only batch that has been committed and
>> still
>> > continue to read from topic further on same topic. I have auto commit to
>> > be off.
>> >
>> >
>> > List<Message> batch .....
>> >
>> > while (iterator.hasNext()) {
>> > batch.add(iterator.next().message());
>> > if(batch size is 50 ){
>> > //===>>>> Once the bulk API is successful it will commit the
>> offset
>> > to zookeeper...
>> > executor.submit(new Thread() process batch and commit batch,
>> > cconsumerConnector)
>> > batch = new batch buffer....
>> > }
>> > }
>> >
>> > This commitOffset API commits all messages that have been read so far.
>> > What is best way to continue reading and only commit another thread
>> finish
>> > batch process is successful. This will lead to fragmentation of the
>> > Consumer offset so what is best way to implement continuous reading
>> stream
>> > and commit the rage offset.
>> >
>> > Is Simple Consumer a better approach for this.
>> >
>> >
>> > Thanks,
>> >
>> > Bhavesh
>> >
>> >
>> >
>> >
>> >
>> >
>> >
>> > Thanks,
>> > Bhavesh
>> >
>>
Re: High Level Consumer and Commit
Posted by Gwen Shapira <gs...@cloudera.com>.
This is not how I'd expect this to work.
Offsets are per-partition and each thread reads its own partition
(Assuming you use Balaji's solution).
So:
Thread 1 reads messages 1..50 from partition 1, processes, indexes,
whatever and commits.
Thread 2 reads messages 1..55 from partition 2, processes, indexes,
whatever and commits.
....
Next iteration thread 1 reads messages 51...103 from partition 1,
thread 2 reads messages 56...160 from partition 2, etc.
No fragmentation :)
Gwen
On Wed, Sep 3, 2014 at 3:42 PM, Bhavesh Mistry
<mi...@gmail.com> wrote:
> Thanks got the idea !! But it will create a fragments for example
>
> Main Thread reads 0-50 messages give to Thread 1 for bulk index and commit
> 0 to 50 offset...
> Main Thread reads 51-100 message give to Thread 2 for bulk index and
> commit 51 100 offset...
>
> So Zookeeper might have offset that will overridden by thread that finish
> first Indexing (Thread 2 will finish first and Thread 1 will commit offset
> then ZK will have older offset). I guess it does not matter in our case
> since indexing same document will over write.
>
> Thanks,
>
> Bhavesh
>
>
>
> On Wed, Sep 3, 2014 at 3:20 PM, Gwen Shapira <gs...@cloudera.com> wrote:
>
>> Thanks, Balaji!
>>
>> It looks like your approach depends on specific implementation
>> details, such as the directory structure in ZK.
>> In this case it doesn't matter much since the APIs are not stable yet,
>> but in general, wouldn't you prefer to use public APIs, even if it
>> means multiple consumers without threads?
>>
>> Gwen
>>
>> On Wed, Sep 3, 2014 at 3:06 PM, Seshadri, Balaji
>> <Ba...@dish.com> wrote:
>> > We can still do with single ConsumerConnector with multiple threads.
>> >
>> > Each thread updates its own data in zookeeper.The below one is our own
>> implementation of commitOffset.
>> >
>> > public void commitOffset(DESMetadata metaData) {
>> > log.debug("Update offsets only for ->"+
>> metaData.toString());
>> > String key =
>> metaData.getTopic()+"/"+metaData.getPartitionNumber();
>> > Long nextOffset = metaData.getOffSet()+1;
>> > if(nextOffset!=checkPointedOffset.get(key)){
>> > ZKGroupTopicDirs topicDirs = new
>> ZKGroupTopicDirs(metaData.getGroupId(),metaData.getTopic());
>> > ZkUtils.updatePersistentPath(zkClient,
>> topicDirs.consumerOffsetDir()+"/"+metaData.getPartitionNumber(),nextOffset+"");
>> > checkPointedOffset.put(key,nextOffset);
>> > }
>> > }
>> >
>> > -----Original Message-----
>> > From: Gwen Shapira [mailto:gshapira@cloudera.com]
>> > Sent: Tuesday, September 02, 2014 11:38 PM
>> > To: users@kafka.apache.org; Philip O'Toole
>> > Subject: Re: High Level Consumer and Commit
>> >
>> > I believe a simpler solution would be to create multiple
>> ConsumerConnector, each with 1 thread (single ConsumerStream) and use
>> commitOffset API to commit all partitions managed by each ConsumerConnector
>> after the thread finished processing the messages.
>> >
>> > Does that solve the problem, Bhavesh?
>> >
>> > Gwen
>> >
>> > On Tue, Sep 2, 2014 at 5:47 PM, Philip O'Toole
>> <ph...@yahoo.com.invalid> wrote:
>> >> Yeah, from reading that I suspect you need the SimpleConsumer. Try it
>> out and see.
>> >>
>> >> Philip
>> >>
>> >>
>> >> -----------------------------------------
>> >> http://www.philipotoole.com
>> >>
>> >>
>> >> On Tuesday, September 2, 2014 5:43 PM, Bhavesh Mistry <
>> mistry.p.bhavesh@gmail.com> wrote:
>> >>
>> >>
>> >>
>> >> Hi Philip,
>> >>
>> >> Yes, We have disabled auto commit but, we need to be able to read from
>> >> particular offset if we manage the offset ourself in some storage(DB).
>> >> High Level consumer does not allow per partition management
>> plug-ability.
>> >>
>> >> I like to have the High Level consumers Failover and auto rebalancing.
>> >> We just need plug ability of offset management.
>> >>
>> >> Thanks,
>> >>
>> >> Bhavesh
>> >>
>> >>
>> >>
>> >> On Tue, Sep 2, 2014 at 5:20 PM, Philip O'Toole <
>> >> philip.otoole@yahoo.com.invalid> wrote:
>> >>
>> >>> No, you'll need to write your own failover.
>> >>>
>> >>> I'm not sure I follow your second question, but the high-level
>> >>> Consumer should be able to do what you want if you disable
>> >>> auto-commit. I'm not sure what else you're asking.
>> >>>
>> >>>
>> >>> Philip
>> >>>
>> >>>
>> >>> -----------------------------------------
>> >>> http://www.philipotoole.com
>> >>>
>> >>>
>> >>> On Tuesday, September 2, 2014 5:15 PM, Bhavesh Mistry <
>> >>> mistry.p.bhavesh@gmail.com> wrote:
>> >>>
>> >>>
>> >>>
>> >>> Hi Philip,
>> >>>
>> >>> Thanks for the update. With Simple Consumer I will not get failover
>> >>> and rebalance that is provided out of box. what is other option not
>> >>> to block reading and keep processing and commit only when batch is
>> done.
>> >>>
>> >>> Thanks,
>> >>>
>> >>> Bhavesh
>> >>>
>> >>>
>> >>>
>> >>> On Tue, Sep 2, 2014 at 4:43 PM, Philip O'Toole <
>> >>> philip.otoole@yahoo.com.invalid> wrote:
>> >>>
>> >>> > Either use the SimpleConsumer which gives you much finer-grained
>> >>> > control, or (this worked with 0.7) spin up a ConsumerConnection
>> >>> > (this is a
>> >>> HighLevel
>> >>> > consumer concept) per partition, turn off auto-commit.
>> >>> >
>> >>> > Philip
>> >>> >
>> >>> >
>> >>> > -----------------------------------------
>> >>> > http://www.philipotoole.com
>> >>> >
>> >>> >
>> >>> > On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <
>> >>> > mistry.p.bhavesh@gmail.com> wrote:
>> >>> >
>> >>> >
>> >>> >
>> >>> > Hi Kafka Group,
>> >>> >
>> >>> > I have to pull the data from the Topic and index into Elastic
>> >>> > Search with Bulk API and wanted to commit only batch that has been
>> >>> > committed and
>> >>> still
>> >>> > continue to read from topic further on same topic. I have auto
>> >>> > commit to be off.
>> >>> >
>> >>> >
>> >>> > List<Message> batch .....
>> >>> >
>> >>> > while (iterator.hasNext()) {
>> >>> > batch.add(iterator.next().message());
>> >>> > if(batch size is 50 ){
>> >>> > //===>>>> Once the bulk API is successful it will commit the
>> >>> offset
>> >>> > to zookeeper...
>> >>> > executor.submit(new Thread() process batch and commit batch,
>> >>> > cconsumerConnector)
>> >>> > batch = new batch buffer....
>> >>> > }
>> >>> > }
>> >>> >
>> >>> > This commitOffset API commits all messages that have been read so
>> far.
>> >>> > What is best way to continue reading and only commit another thread
>> >>> finish
>> >>> > batch process is successful. This will lead to fragmentation of
>> >>> > the Consumer offset so what is best way to implement continuous
>> >>> > reading
>> >>> stream
>> >>> > and commit the rage offset.
>> >>> >
>> >>> > Is Simple Consumer a better approach for this.
>> >>> >
>> >>> >
>> >>> > Thanks,
>> >>> >
>> >>> > Bhavesh
>> >>> >
>> >>> >
>> >>> >
>> >>> >
>> >>> >
>> >>> >
>> >>> >
>> >>> > Thanks,
>> >>> > Bhavesh
>> >>> >
>> >>>
>>
Re: High Level Consumer and Commit
Posted by Bhavesh Mistry <mi...@gmail.com>.
Thanks got the idea !! But it will create a fragments for example
Main Thread reads 0-50 messages give to Thread 1 for bulk index and commit
0 to 50 offset...
Main Thread reads 51-100 message give to Thread 2 for bulk index and
commit 51 100 offset...
So Zookeeper might have offset that will overridden by thread that finish
first Indexing (Thread 2 will finish first and Thread 1 will commit offset
then ZK will have older offset). I guess it does not matter in our case
since indexing same document will over write.
Thanks,
Bhavesh
On Wed, Sep 3, 2014 at 3:20 PM, Gwen Shapira <gs...@cloudera.com> wrote:
> Thanks, Balaji!
>
> It looks like your approach depends on specific implementation
> details, such as the directory structure in ZK.
> In this case it doesn't matter much since the APIs are not stable yet,
> but in general, wouldn't you prefer to use public APIs, even if it
> means multiple consumers without threads?
>
> Gwen
>
> On Wed, Sep 3, 2014 at 3:06 PM, Seshadri, Balaji
> <Ba...@dish.com> wrote:
> > We can still do with single ConsumerConnector with multiple threads.
> >
> > Each thread updates its own data in zookeeper.The below one is our own
> implementation of commitOffset.
> >
> > public void commitOffset(DESMetadata metaData) {
> > log.debug("Update offsets only for ->"+
> metaData.toString());
> > String key =
> metaData.getTopic()+"/"+metaData.getPartitionNumber();
> > Long nextOffset = metaData.getOffSet()+1;
> > if(nextOffset!=checkPointedOffset.get(key)){
> > ZKGroupTopicDirs topicDirs = new
> ZKGroupTopicDirs(metaData.getGroupId(),metaData.getTopic());
> > ZkUtils.updatePersistentPath(zkClient,
> topicDirs.consumerOffsetDir()+"/"+metaData.getPartitionNumber(),nextOffset+"");
> > checkPointedOffset.put(key,nextOffset);
> > }
> > }
> >
> > -----Original Message-----
> > From: Gwen Shapira [mailto:gshapira@cloudera.com]
> > Sent: Tuesday, September 02, 2014 11:38 PM
> > To: users@kafka.apache.org; Philip O'Toole
> > Subject: Re: High Level Consumer and Commit
> >
> > I believe a simpler solution would be to create multiple
> ConsumerConnector, each with 1 thread (single ConsumerStream) and use
> commitOffset API to commit all partitions managed by each ConsumerConnector
> after the thread finished processing the messages.
> >
> > Does that solve the problem, Bhavesh?
> >
> > Gwen
> >
> > On Tue, Sep 2, 2014 at 5:47 PM, Philip O'Toole
> <ph...@yahoo.com.invalid> wrote:
> >> Yeah, from reading that I suspect you need the SimpleConsumer. Try it
> out and see.
> >>
> >> Philip
> >>
> >>
> >> -----------------------------------------
> >> http://www.philipotoole.com
> >>
> >>
> >> On Tuesday, September 2, 2014 5:43 PM, Bhavesh Mistry <
> mistry.p.bhavesh@gmail.com> wrote:
> >>
> >>
> >>
> >> Hi Philip,
> >>
> >> Yes, We have disabled auto commit but, we need to be able to read from
> >> particular offset if we manage the offset ourself in some storage(DB).
> >> High Level consumer does not allow per partition management
> plug-ability.
> >>
> >> I like to have the High Level consumers Failover and auto rebalancing.
> >> We just need plug ability of offset management.
> >>
> >> Thanks,
> >>
> >> Bhavesh
> >>
> >>
> >>
> >> On Tue, Sep 2, 2014 at 5:20 PM, Philip O'Toole <
> >> philip.otoole@yahoo.com.invalid> wrote:
> >>
> >>> No, you'll need to write your own failover.
> >>>
> >>> I'm not sure I follow your second question, but the high-level
> >>> Consumer should be able to do what you want if you disable
> >>> auto-commit. I'm not sure what else you're asking.
> >>>
> >>>
> >>> Philip
> >>>
> >>>
> >>> -----------------------------------------
> >>> http://www.philipotoole.com
> >>>
> >>>
> >>> On Tuesday, September 2, 2014 5:15 PM, Bhavesh Mistry <
> >>> mistry.p.bhavesh@gmail.com> wrote:
> >>>
> >>>
> >>>
> >>> Hi Philip,
> >>>
> >>> Thanks for the update. With Simple Consumer I will not get failover
> >>> and rebalance that is provided out of box. what is other option not
> >>> to block reading and keep processing and commit only when batch is
> done.
> >>>
> >>> Thanks,
> >>>
> >>> Bhavesh
> >>>
> >>>
> >>>
> >>> On Tue, Sep 2, 2014 at 4:43 PM, Philip O'Toole <
> >>> philip.otoole@yahoo.com.invalid> wrote:
> >>>
> >>> > Either use the SimpleConsumer which gives you much finer-grained
> >>> > control, or (this worked with 0.7) spin up a ConsumerConnection
> >>> > (this is a
> >>> HighLevel
> >>> > consumer concept) per partition, turn off auto-commit.
> >>> >
> >>> > Philip
> >>> >
> >>> >
> >>> > -----------------------------------------
> >>> > http://www.philipotoole.com
> >>> >
> >>> >
> >>> > On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <
> >>> > mistry.p.bhavesh@gmail.com> wrote:
> >>> >
> >>> >
> >>> >
> >>> > Hi Kafka Group,
> >>> >
> >>> > I have to pull the data from the Topic and index into Elastic
> >>> > Search with Bulk API and wanted to commit only batch that has been
> >>> > committed and
> >>> still
> >>> > continue to read from topic further on same topic. I have auto
> >>> > commit to be off.
> >>> >
> >>> >
> >>> > List<Message> batch .....
> >>> >
> >>> > while (iterator.hasNext()) {
> >>> > batch.add(iterator.next().message());
> >>> > if(batch size is 50 ){
> >>> > //===>>>> Once the bulk API is successful it will commit the
> >>> offset
> >>> > to zookeeper...
> >>> > executor.submit(new Thread() process batch and commit batch,
> >>> > cconsumerConnector)
> >>> > batch = new batch buffer....
> >>> > }
> >>> > }
> >>> >
> >>> > This commitOffset API commits all messages that have been read so
> far.
> >>> > What is best way to continue reading and only commit another thread
> >>> finish
> >>> > batch process is successful. This will lead to fragmentation of
> >>> > the Consumer offset so what is best way to implement continuous
> >>> > reading
> >>> stream
> >>> > and commit the rage offset.
> >>> >
> >>> > Is Simple Consumer a better approach for this.
> >>> >
> >>> >
> >>> > Thanks,
> >>> >
> >>> > Bhavesh
> >>> >
> >>> >
> >>> >
> >>> >
> >>> >
> >>> >
> >>> >
> >>> > Thanks,
> >>> > Bhavesh
> >>> >
> >>>
>
Re: High Level Consumer and Commit
Posted by Philip O'Toole <ph...@yahoo.com.INVALID>.
>>Only problem is no of connections to Kafka is increased.
*Why* is it a problem?
Philip
RE: High Level Consumer and Commit
Posted by "Seshadri, Balaji" <Ba...@dish.com>.
Multiple consumers with single thread each will also work.
Only problem is no of connections to Kafka is increased.
-----Original Message-----
From: Gwen Shapira [mailto:gshapira@cloudera.com]
Sent: Wednesday, September 03, 2014 4:20 PM
To: users@kafka.apache.org
Cc: Philip O'Toole
Subject: Re: High Level Consumer and Commit
Thanks, Balaji!
It looks like your approach depends on specific implementation details, such as the directory structure in ZK.
In this case it doesn't matter much since the APIs are not stable yet, but in general, wouldn't you prefer to use public APIs, even if it means multiple consumers without threads?
Gwen
On Wed, Sep 3, 2014 at 3:06 PM, Seshadri, Balaji <Ba...@dish.com> wrote:
> We can still do with single ConsumerConnector with multiple threads.
>
> Each thread updates its own data in zookeeper.The below one is our own implementation of commitOffset.
>
> public void commitOffset(DESMetadata metaData) {
> log.debug("Update offsets only for ->"+ metaData.toString());
> String key = metaData.getTopic()+"/"+metaData.getPartitionNumber();
> Long nextOffset = metaData.getOffSet()+1;
> if(nextOffset!=checkPointedOffset.get(key)){
> ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(metaData.getGroupId(),metaData.getTopic());
> ZkUtils.updatePersistentPath(zkClient, topicDirs.consumerOffsetDir()+"/"+metaData.getPartitionNumber(),nextOffset+"");
> checkPointedOffset.put(key,nextOffset);
> }
> }
>
> -----Original Message-----
> From: Gwen Shapira [mailto:gshapira@cloudera.com]
> Sent: Tuesday, September 02, 2014 11:38 PM
> To: users@kafka.apache.org; Philip O'Toole
> Subject: Re: High Level Consumer and Commit
>
> I believe a simpler solution would be to create multiple ConsumerConnector, each with 1 thread (single ConsumerStream) and use commitOffset API to commit all partitions managed by each ConsumerConnector after the thread finished processing the messages.
>
> Does that solve the problem, Bhavesh?
>
> Gwen
>
> On Tue, Sep 2, 2014 at 5:47 PM, Philip O'Toole <ph...@yahoo.com.invalid> wrote:
>> Yeah, from reading that I suspect you need the SimpleConsumer. Try it out and see.
>>
>> Philip
>>
>>
>> -----------------------------------------
>> http://www.philipotoole.com
>>
>>
>> On Tuesday, September 2, 2014 5:43 PM, Bhavesh Mistry <mi...@gmail.com> wrote:
>>
>>
>>
>> Hi Philip,
>>
>> Yes, We have disabled auto commit but, we need to be able to read
>> from particular offset if we manage the offset ourself in some storage(DB).
>> High Level consumer does not allow per partition management plug-ability.
>>
>> I like to have the High Level consumers Failover and auto rebalancing.
>> We just need plug ability of offset management.
>>
>> Thanks,
>>
>> Bhavesh
>>
>>
>>
>> On Tue, Sep 2, 2014 at 5:20 PM, Philip O'Toole <
>> philip.otoole@yahoo.com.invalid> wrote:
>>
>>> No, you'll need to write your own failover.
>>>
>>> I'm not sure I follow your second question, but the high-level
>>> Consumer should be able to do what you want if you disable
>>> auto-commit. I'm not sure what else you're asking.
>>>
>>>
>>> Philip
>>>
>>>
>>> -----------------------------------------
>>> http://www.philipotoole.com
>>>
>>>
>>> On Tuesday, September 2, 2014 5:15 PM, Bhavesh Mistry <
>>> mistry.p.bhavesh@gmail.com> wrote:
>>>
>>>
>>>
>>> Hi Philip,
>>>
>>> Thanks for the update. With Simple Consumer I will not get failover
>>> and rebalance that is provided out of box. what is other option not
>>> to block reading and keep processing and commit only when batch is done.
>>>
>>> Thanks,
>>>
>>> Bhavesh
>>>
>>>
>>>
>>> On Tue, Sep 2, 2014 at 4:43 PM, Philip O'Toole <
>>> philip.otoole@yahoo.com.invalid> wrote:
>>>
>>> > Either use the SimpleConsumer which gives you much finer-grained
>>> > control, or (this worked with 0.7) spin up a ConsumerConnection
>>> > (this is a
>>> HighLevel
>>> > consumer concept) per partition, turn off auto-commit.
>>> >
>>> > Philip
>>> >
>>> >
>>> > -----------------------------------------
>>> > http://www.philipotoole.com
>>> >
>>> >
>>> > On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <
>>> > mistry.p.bhavesh@gmail.com> wrote:
>>> >
>>> >
>>> >
>>> > Hi Kafka Group,
>>> >
>>> > I have to pull the data from the Topic and index into Elastic
>>> > Search with Bulk API and wanted to commit only batch that has been
>>> > committed and
>>> still
>>> > continue to read from topic further on same topic. I have auto
>>> > commit to be off.
>>> >
>>> >
>>> > List<Message> batch .....
>>> >
>>> > while (iterator.hasNext()) {
>>> > batch.add(iterator.next().message());
>>> > if(batch size is 50 ){
>>> > //===>>>> Once the bulk API is successful it will commit
>>> > the
>>> offset
>>> > to zookeeper...
>>> > executor.submit(new Thread() process batch and commit batch,
>>> > cconsumerConnector)
>>> > batch = new batch buffer....
>>> > }
>>> > }
>>> >
>>> > This commitOffset API commits all messages that have been read so far.
>>> > What is best way to continue reading and only commit another
>>> > thread
>>> finish
>>> > batch process is successful. This will lead to fragmentation of
>>> > the Consumer offset so what is best way to implement continuous
>>> > reading
>>> stream
>>> > and commit the rage offset.
>>> >
>>> > Is Simple Consumer a better approach for this.
>>> >
>>> >
>>> > Thanks,
>>> >
>>> > Bhavesh
>>> >
>>> >
>>> >
>>> >
>>> >
>>> >
>>> >
>>> > Thanks,
>>> > Bhavesh
>>> >
>>>
Re: High Level Consumer and Commit
Posted by Gwen Shapira <gs...@cloudera.com>.
Thanks, Balaji!
It looks like your approach depends on specific implementation
details, such as the directory structure in ZK.
In this case it doesn't matter much since the APIs are not stable yet,
but in general, wouldn't you prefer to use public APIs, even if it
means multiple consumers without threads?
Gwen
On Wed, Sep 3, 2014 at 3:06 PM, Seshadri, Balaji
<Ba...@dish.com> wrote:
> We can still do with single ConsumerConnector with multiple threads.
>
> Each thread updates its own data in zookeeper.The below one is our own implementation of commitOffset.
>
> public void commitOffset(DESMetadata metaData) {
> log.debug("Update offsets only for ->"+ metaData.toString());
> String key = metaData.getTopic()+"/"+metaData.getPartitionNumber();
> Long nextOffset = metaData.getOffSet()+1;
> if(nextOffset!=checkPointedOffset.get(key)){
> ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(metaData.getGroupId(),metaData.getTopic());
> ZkUtils.updatePersistentPath(zkClient, topicDirs.consumerOffsetDir()+"/"+metaData.getPartitionNumber(),nextOffset+"");
> checkPointedOffset.put(key,nextOffset);
> }
> }
>
> -----Original Message-----
> From: Gwen Shapira [mailto:gshapira@cloudera.com]
> Sent: Tuesday, September 02, 2014 11:38 PM
> To: users@kafka.apache.org; Philip O'Toole
> Subject: Re: High Level Consumer and Commit
>
> I believe a simpler solution would be to create multiple ConsumerConnector, each with 1 thread (single ConsumerStream) and use commitOffset API to commit all partitions managed by each ConsumerConnector after the thread finished processing the messages.
>
> Does that solve the problem, Bhavesh?
>
> Gwen
>
> On Tue, Sep 2, 2014 at 5:47 PM, Philip O'Toole <ph...@yahoo.com.invalid> wrote:
>> Yeah, from reading that I suspect you need the SimpleConsumer. Try it out and see.
>>
>> Philip
>>
>>
>> -----------------------------------------
>> http://www.philipotoole.com
>>
>>
>> On Tuesday, September 2, 2014 5:43 PM, Bhavesh Mistry <mi...@gmail.com> wrote:
>>
>>
>>
>> Hi Philip,
>>
>> Yes, We have disabled auto commit but, we need to be able to read from
>> particular offset if we manage the offset ourself in some storage(DB).
>> High Level consumer does not allow per partition management plug-ability.
>>
>> I like to have the High Level consumers Failover and auto rebalancing.
>> We just need plug ability of offset management.
>>
>> Thanks,
>>
>> Bhavesh
>>
>>
>>
>> On Tue, Sep 2, 2014 at 5:20 PM, Philip O'Toole <
>> philip.otoole@yahoo.com.invalid> wrote:
>>
>>> No, you'll need to write your own failover.
>>>
>>> I'm not sure I follow your second question, but the high-level
>>> Consumer should be able to do what you want if you disable
>>> auto-commit. I'm not sure what else you're asking.
>>>
>>>
>>> Philip
>>>
>>>
>>> -----------------------------------------
>>> http://www.philipotoole.com
>>>
>>>
>>> On Tuesday, September 2, 2014 5:15 PM, Bhavesh Mistry <
>>> mistry.p.bhavesh@gmail.com> wrote:
>>>
>>>
>>>
>>> Hi Philip,
>>>
>>> Thanks for the update. With Simple Consumer I will not get failover
>>> and rebalance that is provided out of box. what is other option not
>>> to block reading and keep processing and commit only when batch is done.
>>>
>>> Thanks,
>>>
>>> Bhavesh
>>>
>>>
>>>
>>> On Tue, Sep 2, 2014 at 4:43 PM, Philip O'Toole <
>>> philip.otoole@yahoo.com.invalid> wrote:
>>>
>>> > Either use the SimpleConsumer which gives you much finer-grained
>>> > control, or (this worked with 0.7) spin up a ConsumerConnection
>>> > (this is a
>>> HighLevel
>>> > consumer concept) per partition, turn off auto-commit.
>>> >
>>> > Philip
>>> >
>>> >
>>> > -----------------------------------------
>>> > http://www.philipotoole.com
>>> >
>>> >
>>> > On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <
>>> > mistry.p.bhavesh@gmail.com> wrote:
>>> >
>>> >
>>> >
>>> > Hi Kafka Group,
>>> >
>>> > I have to pull the data from the Topic and index into Elastic
>>> > Search with Bulk API and wanted to commit only batch that has been
>>> > committed and
>>> still
>>> > continue to read from topic further on same topic. I have auto
>>> > commit to be off.
>>> >
>>> >
>>> > List<Message> batch .....
>>> >
>>> > while (iterator.hasNext()) {
>>> > batch.add(iterator.next().message());
>>> > if(batch size is 50 ){
>>> > //===>>>> Once the bulk API is successful it will commit the
>>> offset
>>> > to zookeeper...
>>> > executor.submit(new Thread() process batch and commit batch,
>>> > cconsumerConnector)
>>> > batch = new batch buffer....
>>> > }
>>> > }
>>> >
>>> > This commitOffset API commits all messages that have been read so far.
>>> > What is best way to continue reading and only commit another thread
>>> finish
>>> > batch process is successful. This will lead to fragmentation of
>>> > the Consumer offset so what is best way to implement continuous
>>> > reading
>>> stream
>>> > and commit the rage offset.
>>> >
>>> > Is Simple Consumer a better approach for this.
>>> >
>>> >
>>> > Thanks,
>>> >
>>> > Bhavesh
>>> >
>>> >
>>> >
>>> >
>>> >
>>> >
>>> >
>>> > Thanks,
>>> > Bhavesh
>>> >
>>>
RE: High Level Consumer and Commit
Posted by "Seshadri, Balaji" <Ba...@dish.com>.
We can still do with single ConsumerConnector with multiple threads.
Each thread updates its own data in zookeeper.The below one is our own implementation of commitOffset.
public void commitOffset(DESMetadata metaData) {
log.debug("Update offsets only for ->"+ metaData.toString());
String key = metaData.getTopic()+"/"+metaData.getPartitionNumber();
Long nextOffset = metaData.getOffSet()+1;
if(nextOffset!=checkPointedOffset.get(key)){
ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(metaData.getGroupId(),metaData.getTopic());
ZkUtils.updatePersistentPath(zkClient, topicDirs.consumerOffsetDir()+"/"+metaData.getPartitionNumber(),nextOffset+"");
checkPointedOffset.put(key,nextOffset);
}
}
-----Original Message-----
From: Gwen Shapira [mailto:gshapira@cloudera.com]
Sent: Tuesday, September 02, 2014 11:38 PM
To: users@kafka.apache.org; Philip O'Toole
Subject: Re: High Level Consumer and Commit
I believe a simpler solution would be to create multiple ConsumerConnector, each with 1 thread (single ConsumerStream) and use commitOffset API to commit all partitions managed by each ConsumerConnector after the thread finished processing the messages.
Does that solve the problem, Bhavesh?
Gwen
On Tue, Sep 2, 2014 at 5:47 PM, Philip O'Toole <ph...@yahoo.com.invalid> wrote:
> Yeah, from reading that I suspect you need the SimpleConsumer. Try it out and see.
>
> Philip
>
>
> -----------------------------------------
> http://www.philipotoole.com
>
>
> On Tuesday, September 2, 2014 5:43 PM, Bhavesh Mistry <mi...@gmail.com> wrote:
>
>
>
> Hi Philip,
>
> Yes, We have disabled auto commit but, we need to be able to read from
> particular offset if we manage the offset ourself in some storage(DB).
> High Level consumer does not allow per partition management plug-ability.
>
> I like to have the High Level consumers Failover and auto rebalancing.
> We just need plug ability of offset management.
>
> Thanks,
>
> Bhavesh
>
>
>
> On Tue, Sep 2, 2014 at 5:20 PM, Philip O'Toole <
> philip.otoole@yahoo.com.invalid> wrote:
>
>> No, you'll need to write your own failover.
>>
>> I'm not sure I follow your second question, but the high-level
>> Consumer should be able to do what you want if you disable
>> auto-commit. I'm not sure what else you're asking.
>>
>>
>> Philip
>>
>>
>> -----------------------------------------
>> http://www.philipotoole.com
>>
>>
>> On Tuesday, September 2, 2014 5:15 PM, Bhavesh Mistry <
>> mistry.p.bhavesh@gmail.com> wrote:
>>
>>
>>
>> Hi Philip,
>>
>> Thanks for the update. With Simple Consumer I will not get failover
>> and rebalance that is provided out of box. what is other option not
>> to block reading and keep processing and commit only when batch is done.
>>
>> Thanks,
>>
>> Bhavesh
>>
>>
>>
>> On Tue, Sep 2, 2014 at 4:43 PM, Philip O'Toole <
>> philip.otoole@yahoo.com.invalid> wrote:
>>
>> > Either use the SimpleConsumer which gives you much finer-grained
>> > control, or (this worked with 0.7) spin up a ConsumerConnection
>> > (this is a
>> HighLevel
>> > consumer concept) per partition, turn off auto-commit.
>> >
>> > Philip
>> >
>> >
>> > -----------------------------------------
>> > http://www.philipotoole.com
>> >
>> >
>> > On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <
>> > mistry.p.bhavesh@gmail.com> wrote:
>> >
>> >
>> >
>> > Hi Kafka Group,
>> >
>> > I have to pull the data from the Topic and index into Elastic
>> > Search with Bulk API and wanted to commit only batch that has been
>> > committed and
>> still
>> > continue to read from topic further on same topic. I have auto
>> > commit to be off.
>> >
>> >
>> > List<Message> batch .....
>> >
>> > while (iterator.hasNext()) {
>> > batch.add(iterator.next().message());
>> > if(batch size is 50 ){
>> > //===>>>> Once the bulk API is successful it will commit the
>> offset
>> > to zookeeper...
>> > executor.submit(new Thread() process batch and commit batch,
>> > cconsumerConnector)
>> > batch = new batch buffer....
>> > }
>> > }
>> >
>> > This commitOffset API commits all messages that have been read so far.
>> > What is best way to continue reading and only commit another thread
>> finish
>> > batch process is successful. This will lead to fragmentation of
>> > the Consumer offset so what is best way to implement continuous
>> > reading
>> stream
>> > and commit the rage offset.
>> >
>> > Is Simple Consumer a better approach for this.
>> >
>> >
>> > Thanks,
>> >
>> > Bhavesh
>> >
>> >
>> >
>> >
>> >
>> >
>> >
>> > Thanks,
>> > Bhavesh
>> >
>>
Re: High Level Consumer and Commit
Posted by Gwen Shapira <gs...@cloudera.com>.
I believe a simpler solution would be to create multiple
ConsumerConnector, each with 1 thread (single ConsumerStream) and use
commitOffset API to commit all partitions managed by each
ConsumerConnector after the thread finished processing the messages.
Does that solve the problem, Bhavesh?
Gwen
On Tue, Sep 2, 2014 at 5:47 PM, Philip O'Toole
<ph...@yahoo.com.invalid> wrote:
> Yeah, from reading that I suspect you need the SimpleConsumer. Try it out and see.
>
> Philip
>
>
> -----------------------------------------
> http://www.philipotoole.com
>
>
> On Tuesday, September 2, 2014 5:43 PM, Bhavesh Mistry <mi...@gmail.com> wrote:
>
>
>
> Hi Philip,
>
> Yes, We have disabled auto commit but, we need to be able to read from
> particular offset if we manage the offset ourself in some storage(DB).
> High Level consumer does not allow per partition management plug-ability.
>
> I like to have the High Level consumers Failover and auto rebalancing. We
> just need plug ability of offset management.
>
> Thanks,
>
> Bhavesh
>
>
>
> On Tue, Sep 2, 2014 at 5:20 PM, Philip O'Toole <
> philip.otoole@yahoo.com.invalid> wrote:
>
>> No, you'll need to write your own failover.
>>
>> I'm not sure I follow your second question, but the high-level Consumer
>> should be able to do what you want if you disable auto-commit. I'm not sure
>> what else you're asking.
>>
>>
>> Philip
>>
>>
>> -----------------------------------------
>> http://www.philipotoole.com
>>
>>
>> On Tuesday, September 2, 2014 5:15 PM, Bhavesh Mistry <
>> mistry.p.bhavesh@gmail.com> wrote:
>>
>>
>>
>> Hi Philip,
>>
>> Thanks for the update. With Simple Consumer I will not get failover and
>> rebalance that is provided out of box. what is other option not to block
>> reading and keep processing and commit only when batch is done.
>>
>> Thanks,
>>
>> Bhavesh
>>
>>
>>
>> On Tue, Sep 2, 2014 at 4:43 PM, Philip O'Toole <
>> philip.otoole@yahoo.com.invalid> wrote:
>>
>> > Either use the SimpleConsumer which gives you much finer-grained control,
>> > or (this worked with 0.7) spin up a ConsumerConnection (this is a
>> HighLevel
>> > consumer concept) per partition, turn off auto-commit.
>> >
>> > Philip
>> >
>> >
>> > -----------------------------------------
>> > http://www.philipotoole.com
>> >
>> >
>> > On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <
>> > mistry.p.bhavesh@gmail.com> wrote:
>> >
>> >
>> >
>> > Hi Kafka Group,
>> >
>> > I have to pull the data from the Topic and index into Elastic Search with
>> > Bulk API and wanted to commit only batch that has been committed and
>> still
>> > continue to read from topic further on same topic. I have auto commit to
>> > be off.
>> >
>> >
>> > List<Message> batch .....
>> >
>> > while (iterator.hasNext()) {
>> > batch.add(iterator.next().message());
>> > if(batch size is 50 ){
>> > //===>>>> Once the bulk API is successful it will commit the
>> offset
>> > to zookeeper...
>> > executor.submit(new Thread() process batch and commit batch,
>> > cconsumerConnector)
>> > batch = new batch buffer....
>> > }
>> > }
>> >
>> > This commitOffset API commits all messages that have been read so far.
>> > What is best way to continue reading and only commit another thread
>> finish
>> > batch process is successful. This will lead to fragmentation of the
>> > Consumer offset so what is best way to implement continuous reading
>> stream
>> > and commit the rage offset.
>> >
>> > Is Simple Consumer a better approach for this.
>> >
>> >
>> > Thanks,
>> >
>> > Bhavesh
>> >
>> >
>> >
>> >
>> >
>> >
>> >
>> > Thanks,
>> > Bhavesh
>> >
>>
Re: High Level Consumer and Commit
Posted by Philip O'Toole <ph...@yahoo.com.INVALID>.
Yeah, from reading that I suspect you need the SimpleConsumer. Try it out and see.
Philip
-----------------------------------------
http://www.philipotoole.com
On Tuesday, September 2, 2014 5:43 PM, Bhavesh Mistry <mi...@gmail.com> wrote:
Hi Philip,
Yes, We have disabled auto commit but, we need to be able to read from
particular offset if we manage the offset ourself in some storage(DB).
High Level consumer does not allow per partition management plug-ability.
I like to have the High Level consumers Failover and auto rebalancing. We
just need plug ability of offset management.
Thanks,
Bhavesh
On Tue, Sep 2, 2014 at 5:20 PM, Philip O'Toole <
philip.otoole@yahoo.com.invalid> wrote:
> No, you'll need to write your own failover.
>
> I'm not sure I follow your second question, but the high-level Consumer
> should be able to do what you want if you disable auto-commit. I'm not sure
> what else you're asking.
>
>
> Philip
>
>
> -----------------------------------------
> http://www.philipotoole.com
>
>
> On Tuesday, September 2, 2014 5:15 PM, Bhavesh Mistry <
> mistry.p.bhavesh@gmail.com> wrote:
>
>
>
> Hi Philip,
>
> Thanks for the update. With Simple Consumer I will not get failover and
> rebalance that is provided out of box. what is other option not to block
> reading and keep processing and commit only when batch is done.
>
> Thanks,
>
> Bhavesh
>
>
>
> On Tue, Sep 2, 2014 at 4:43 PM, Philip O'Toole <
> philip.otoole@yahoo.com.invalid> wrote:
>
> > Either use the SimpleConsumer which gives you much finer-grained control,
> > or (this worked with 0.7) spin up a ConsumerConnection (this is a
> HighLevel
> > consumer concept) per partition, turn off auto-commit.
> >
> > Philip
> >
> >
> > -----------------------------------------
> > http://www.philipotoole.com
> >
> >
> > On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <
> > mistry.p.bhavesh@gmail.com> wrote:
> >
> >
> >
> > Hi Kafka Group,
> >
> > I have to pull the data from the Topic and index into Elastic Search with
> > Bulk API and wanted to commit only batch that has been committed and
> still
> > continue to read from topic further on same topic. I have auto commit to
> > be off.
> >
> >
> > List<Message> batch .....
> >
> > while (iterator.hasNext()) {
> > batch.add(iterator.next().message());
> > if(batch size is 50 ){
> > //===>>>> Once the bulk API is successful it will commit the
> offset
> > to zookeeper...
> > executor.submit(new Thread() process batch and commit batch,
> > cconsumerConnector)
> > batch = new batch buffer....
> > }
> > }
> >
> > This commitOffset API commits all messages that have been read so far.
> > What is best way to continue reading and only commit another thread
> finish
> > batch process is successful. This will lead to fragmentation of the
> > Consumer offset so what is best way to implement continuous reading
> stream
> > and commit the rage offset.
> >
> > Is Simple Consumer a better approach for this.
> >
> >
> > Thanks,
> >
> > Bhavesh
> >
> >
> >
> >
> >
> >
> >
> > Thanks,
> > Bhavesh
> >
>
Re: High Level Consumer and Commit
Posted by Bhavesh Mistry <mi...@gmail.com>.
Hi Philip,
Yes, We have disabled auto commit but, we need to be able to read from
particular offset if we manage the offset ourself in some storage(DB).
High Level consumer does not allow per partition management plug-ability.
I like to have the High Level consumers Failover and auto rebalancing. We
just need plug ability of offset management.
Thanks,
Bhavesh
On Tue, Sep 2, 2014 at 5:20 PM, Philip O'Toole <
philip.otoole@yahoo.com.invalid> wrote:
> No, you'll need to write your own failover.
>
> I'm not sure I follow your second question, but the high-level Consumer
> should be able to do what you want if you disable auto-commit. I'm not sure
> what else you're asking.
>
>
> Philip
>
>
> -----------------------------------------
> http://www.philipotoole.com
>
>
> On Tuesday, September 2, 2014 5:15 PM, Bhavesh Mistry <
> mistry.p.bhavesh@gmail.com> wrote:
>
>
>
> Hi Philip,
>
> Thanks for the update. With Simple Consumer I will not get failover and
> rebalance that is provided out of box. what is other option not to block
> reading and keep processing and commit only when batch is done.
>
> Thanks,
>
> Bhavesh
>
>
>
> On Tue, Sep 2, 2014 at 4:43 PM, Philip O'Toole <
> philip.otoole@yahoo.com.invalid> wrote:
>
> > Either use the SimpleConsumer which gives you much finer-grained control,
> > or (this worked with 0.7) spin up a ConsumerConnection (this is a
> HighLevel
> > consumer concept) per partition, turn off auto-commit.
> >
> > Philip
> >
> >
> > -----------------------------------------
> > http://www.philipotoole.com
> >
> >
> > On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <
> > mistry.p.bhavesh@gmail.com> wrote:
> >
> >
> >
> > Hi Kafka Group,
> >
> > I have to pull the data from the Topic and index into Elastic Search with
> > Bulk API and wanted to commit only batch that has been committed and
> still
> > continue to read from topic further on same topic. I have auto commit to
> > be off.
> >
> >
> > List<Message> batch .....
> >
> > while (iterator.hasNext()) {
> > batch.add(iterator.next().message());
> > if(batch size is 50 ){
> > //===>>>> Once the bulk API is successful it will commit the
> offset
> > to zookeeper...
> > executor.submit(new Thread() process batch and commit batch,
> > cconsumerConnector)
> > batch = new batch buffer....
> > }
> > }
> >
> > This commitOffset API commits all messages that have been read so far.
> > What is best way to continue reading and only commit another thread
> finish
> > batch process is successful. This will lead to fragmentation of the
> > Consumer offset so what is best way to implement continuous reading
> stream
> > and commit the rage offset.
> >
> > Is Simple Consumer a better approach for this.
> >
> >
> > Thanks,
> >
> > Bhavesh
> >
> >
> >
> >
> >
> >
> >
> > Thanks,
> > Bhavesh
> >
>
Re: High Level Consumer and Commit
Posted by Philip O'Toole <ph...@yahoo.com.INVALID>.
No, you'll need to write your own failover.
I'm not sure I follow your second question, but the high-level Consumer should be able to do what you want if you disable auto-commit. I'm not sure what else you're asking.
Philip
-----------------------------------------
http://www.philipotoole.com
On Tuesday, September 2, 2014 5:15 PM, Bhavesh Mistry <mi...@gmail.com> wrote:
Hi Philip,
Thanks for the update. With Simple Consumer I will not get failover and
rebalance that is provided out of box. what is other option not to block
reading and keep processing and commit only when batch is done.
Thanks,
Bhavesh
On Tue, Sep 2, 2014 at 4:43 PM, Philip O'Toole <
philip.otoole@yahoo.com.invalid> wrote:
> Either use the SimpleConsumer which gives you much finer-grained control,
> or (this worked with 0.7) spin up a ConsumerConnection (this is a HighLevel
> consumer concept) per partition, turn off auto-commit.
>
> Philip
>
>
> -----------------------------------------
> http://www.philipotoole.com
>
>
> On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <
> mistry.p.bhavesh@gmail.com> wrote:
>
>
>
> Hi Kafka Group,
>
> I have to pull the data from the Topic and index into Elastic Search with
> Bulk API and wanted to commit only batch that has been committed and still
> continue to read from topic further on same topic. I have auto commit to
> be off.
>
>
> List<Message> batch .....
>
> while (iterator.hasNext()) {
> batch.add(iterator.next().message());
> if(batch size is 50 ){
> //===>>>> Once the bulk API is successful it will commit the offset
> to zookeeper...
> executor.submit(new Thread() process batch and commit batch,
> cconsumerConnector)
> batch = new batch buffer....
> }
> }
>
> This commitOffset API commits all messages that have been read so far.
> What is best way to continue reading and only commit another thread finish
> batch process is successful. This will lead to fragmentation of the
> Consumer offset so what is best way to implement continuous reading stream
> and commit the rage offset.
>
> Is Simple Consumer a better approach for this.
>
>
> Thanks,
>
> Bhavesh
>
>
>
>
>
>
>
> Thanks,
> Bhavesh
>
Re: High Level Consumer and Commit
Posted by Bhavesh Mistry <mi...@gmail.com>.
Hi Philip,
Thanks for the update. With Simple Consumer I will not get failover and
rebalance that is provided out of box. what is other option not to block
reading and keep processing and commit only when batch is done.
Thanks,
Bhavesh
On Tue, Sep 2, 2014 at 4:43 PM, Philip O'Toole <
philip.otoole@yahoo.com.invalid> wrote:
> Either use the SimpleConsumer which gives you much finer-grained control,
> or (this worked with 0.7) spin up a ConsumerConnection (this is a HighLevel
> consumer concept) per partition, turn off auto-commit.
>
> Philip
>
>
> -----------------------------------------
> http://www.philipotoole.com
>
>
> On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <
> mistry.p.bhavesh@gmail.com> wrote:
>
>
>
> Hi Kafka Group,
>
> I have to pull the data from the Topic and index into Elastic Search with
> Bulk API and wanted to commit only batch that has been committed and still
> continue to read from topic further on same topic. I have auto commit to
> be off.
>
>
> List<Message> batch .....
>
> while (iterator.hasNext()) {
> batch.add(iterator.next().message());
> if(batch size is 50 ){
> //===>>>> Once the bulk API is successful it will commit the offset
> to zookeeper...
> executor.submit(new Thread() process batch and commit batch,
> cconsumerConnector)
> batch = new batch buffer....
> }
> }
>
> This commitOffset API commits all messages that have been read so far.
> What is best way to continue reading and only commit another thread finish
> batch process is successful. This will lead to fragmentation of the
> Consumer offset so what is best way to implement continuous reading stream
> and commit the rage offset.
>
> Is Simple Consumer a better approach for this.
>
>
> Thanks,
>
> Bhavesh
>
>
>
>
>
>
>
> Thanks,
> Bhavesh
>
Re: High Level Consumer and Commit
Posted by Philip O'Toole <ph...@yahoo.com.INVALID>.
Either use the SimpleConsumer which gives you much finer-grained control, or (this worked with 0.7) spin up a ConsumerConnection (this is a HighLevel consumer concept) per partition, turn off auto-commit.
Philip
-----------------------------------------
http://www.philipotoole.com
On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <mi...@gmail.com> wrote:
Hi Kafka Group,
I have to pull the data from the Topic and index into Elastic Search with
Bulk API and wanted to commit only batch that has been committed and still
continue to read from topic further on same topic. I have auto commit to
be off.
List<Message> batch .....
while (iterator.hasNext()) {
batch.add(iterator.next().message());
if(batch size is 50 ){
//===>>>> Once the bulk API is successful it will commit the offset
to zookeeper...
executor.submit(new Thread() process batch and commit batch,
cconsumerConnector)
batch = new batch buffer....
}
}
This commitOffset API commits all messages that have been read so far.
What is best way to continue reading and only commit another thread finish
batch process is successful. This will lead to fragmentation of the
Consumer offset so what is best way to implement continuous reading stream
and commit the rage offset.
Is Simple Consumer a better approach for this.
Thanks,
Bhavesh
Thanks,
Bhavesh
Re: High Level Consumer and Commit
Posted by Philip O'Toole <ph...@yahoo.com.INVALID>.
Either use the SimpleConsumer which gives you much finer-grained control, or (this worked with 0.7) spin up a ConsumerConnection (this is a HighLevel consumer concept) per partition, turn off auto-commit.
Philip
-----------------------------------------
http://www.philipotoole.com
On Tuesday, September 2, 2014 4:38 PM, Bhavesh Mistry <mi...@gmail.com> wrote:
Hi Kafka Group,
I have to pull the data from the Topic and index into Elastic Search with
Bulk API and wanted to commit only batch that has been committed and still
continue to read from topic further on same topic. I have auto commit to
be off.
List<Message> batch .....
while (iterator.hasNext()) {
batch.add(iterator.next().message());
if(batch size is 50 ){
//===>>>> Once the bulk API is successful it will commit the offset
to zookeeper...
executor.submit(new Thread() process batch and commit batch,
cconsumerConnector)
batch = new batch buffer....
}
}
This commitOffset API commits all messages that have been read so far.
What is best way to continue reading and only commit another thread finish
batch process is successful. This will lead to fragmentation of the
Consumer offset so what is best way to implement continuous reading stream
and commit the rage offset.
Is Simple Consumer a better approach for this.
Thanks,
Bhavesh
Thanks,
Bhavesh