I created KAFKA-1951 to improve the offset checker to let it print offsets
stored in both zookeeper and Kafka.
I don¹t think we have a way to wipe out offset topic in Kafka for a
consumer.
‹Jiangjie (Becket) Qin
On 2/12/15, 9:59 PM, "tao xiao" <xiaotao...@gmail.com> wrote:
>Thanks for the explanation. It there a way that I can wipe out the offset
>stored in kafka so that the checker can continue to work again?
>
>On Fri, Feb 13, 2015 at 1:31 PM, Jiangjie Qin <j...@linkedin.com.invalid>
>wrote:
>
>> I think this is the offset checker bug.
>> The offset checker will
>> 1. first check if the offset exists in offset topic on broker or not.
>> 2. If it is on broker then it will just return that offset.
>> 3. Otherwise it goes to zookeeper.
>>
>> So the problem you saw was actually following this logic.
>> After dual commit, offset topic already had the offsets for this
>>consumer
>> and topic.
>> Then you switched to zookeeper commit.
>> Because the offset topic has the offsets already, offset checker will
>>use
>> that and skip checking zookeeper. So the offset will not change anymore
>> because you are no longer committing to offset topic on broker, while
>> offset checker always use that offset.
>>
>> On 2/12/15, 7:30 PM, "tao xiao" <xiaotao...@gmail.com> wrote:
>>
>> >I used the one shipped with 0.8.2. It is pretty straightforward to
>> >reproduce the issue.
>> >
>> >Here are the steps to reproduce:
>> >1. I have a consumer using high level consumer API with initial
>>settings
>> >offsets.storage=kafka and dual.commit.enabled=false.
>> >2. After consuming messages for a while shutdown the consumer and
>>change
>> >setting dual.commit.enabled=true
>> >3. bounce the consumer and run for while. The lag looks good
>> >4. change setting offsets.storage=zookeeper and bounce the consumer.
>> >Starting from now the lag remain unchanged
>> >
>> >On Fri, Feb 13, 2015 at 11:01 AM, Joel Koshy <jjkosh...@gmail.com>
>>wrote:
>> >
>> >> That is weird. Are you by any chance running an older version of the
>> >> offset checker? Is this straightforward to reproduce?
>> >>
>> >> On Fri, Feb 13, 2015 at 09:57:31AM +0800, tao xiao wrote:
>> >> > Joel,
>> >> >
>> >> > No, the metric was not increasing. It was 0 all the time.
>> >> >
>> >> > On Fri, Feb 13, 2015 at 12:18 AM, Joel Koshy <jjkosh...@gmail.com>
>> >> wrote:
>> >> >
>> >> > > Actually I meant to say check that is not increasing.
>> >> > >
>> >> > > On Thu, Feb 12, 2015 at 08:15:01AM -0800, Joel Koshy wrote:
>> >> > > > Possibly a bug - can you also look at the MaxLag mbean in the
>> >> consumer
>> >> > > > to verify that the maxlag is zero?
>> >> > > >
>> >> > > > On Thu, Feb 12, 2015 at 11:24:42PM +0800, tao xiao wrote:
>> >> > > > > Hi Joel,
>> >> > > > >
>> >> > > > > When I set dual.commit.enabled=true the count value of both
>> >> metrics got
>> >> > > > > increased. After I set offsets.storage=zookeeper only
>> >> > > ZooKeeperCommitsPerSec
>> >> > > > > changed but not KafkaCommitsPerSec. I think this is expected
>>as
>> >> kafka
>> >> > > > > offset storage was turned off.
>> >> > > > >
>> >> > > > > But when I looked up the consumer lag via
>> >> > > kafka.tools.ConsumerOffsetChecker
>> >> > > > > the lag still remained unchanged.
>> >> > > > >
>> >> > > > > I scanned through the source code of ConsumerOffsetChecker it
>> >> doesn't
>> >> > > > > check the offset in zk unless offsetFetchResponse returns
>> >>NoOffset.
>> >> > > Since
>> >> > > > > the consumer used kafka as the offset storage before I don't
>> >>think
>> >> > > > > offsetFetchResponse would return NoOffset
>> >> > > > >
>> >> > > > > offsetFetchResponse.requestInfo.foreach { case
>> >>(topicAndPartition,
>> >> > > > > offsetAndMetadata) =>
>> >> > > > >
>> >> > > > > if (offsetAndMetadata ==
>> >>OffsetMetadataAndError.NoOffset) {
>> >> > > > >
>> >> > > > > val topicDirs = new ZKGroupTopicDirs(group,
>> >> > > topicAndPartition.
>> >> > > > > topic)
>> >> > > > >
>> >> > > > > // this group may not have migrated off zookeeper
>>for
>> >> offsets
>> >> > > > > storage (we don't expose the dual-commit option in this tool
>> >> > > > >
>> >> > > > > // (meaning the lag may be off until all the
>>consumers
>> >> in the
>> >> > > > > group have the same setting for offsets storage)
>> >> > > > >
>> >> > > > > try {
>> >> > > > >
>> >> > > > > val offset = ZkUtils.readData(zkClient,
>> >> > > topicDirs.consumerOffsetDir
>> >> > > > > + "/%d".format(topicAndPartition.partition))._1.toLong
>> >> > > > >
>> >> > > > > offsetMap.put(topicAndPartition, offset)
>> >> > > > >
>> >> > > > > } catch {
>> >> > > > >
>> >> > > > > case z: ZkNoNodeException =>
>> >> > > > >
>> >> > > > >
>> >> > > if(ZkUtils.pathExists(zkClient,topicDirs.consumerOffsetDir))
>> >> > > > >
>> >> > > > > offsetMap.put(topicAndPartition,-1)
>> >> > > > >
>> >> > > > > else
>> >> > > > >
>> >> > > > > throw z
>> >> > > > >
>> >> > > > > }
>> >> > > > >
>> >> > > > > }
>> >> > > > >
>> >> > > > > else if (offsetAndMetadata.error ==
>> >>ErrorMapping.NoError)
>> >> > > > >
>> >> > > > > offsetMap.put(topicAndPartition,
>> >> offsetAndMetadata.offset)
>> >> > > > >
>> >> > > > > else {
>> >> > > > >
>> >> > > > > println("Could not fetch offset for %s due to
>> >> %s.".format(
>> >> > > > > topicAndPartition,
>> >> ErrorMapping.exceptionFor(offsetAndMetadata.error)))
>> >> > > > >
>> >> > > > > }
>> >> > > > >
>> >> > > > > }
>> >> > > > >
>> >> > > > > On Thu, Feb 12, 2015 at 10:03 PM, Joel Koshy
>> >><jjkosh...@gmail.com>
>> >> > > wrote:
>> >> > > > >
>> >> > > > > > There are mbeans named KafkaCommitsPerSec and
>> >> ZooKeeperCommitsPerSec
>> >> > > -
>> >> > > > > > can you look those up and see what they report?
>> >> > > > > >
>> >> > > > > > On Thu, Feb 12, 2015 at 07:32:39PM +0800, tao xiao wrote:
>> >> > > > > > > Hi team,
>> >> > > > > > >
>> >> > > > > > > I was trying to migrate my consumer offset from kafka to
>> >> zookeeper.
>> >> > > > > > >
>> >> > > > > > > Here is the original settings of my consumer
>> >> > > > > > >
>> >> > > > > > > props.put("offsets.storage", "kafka");
>> >> > > > > > >
>> >> > > > > > > props.put("dual.commit.enabled", "false");
>> >> > > > > > > Here is the steps
>> >> > > > > > >
>> >> > > > > > > 1. set dual.commit.enabled=true
>> >> > > > > > > 2. restart my consumer and monitor offset lag with
>> >> > > > > > > kafka.tools.ConsumerOffsetChecker
>> >> > > > > > > 3. set offsets.storage=zookeeper
>> >> > > > > > > 4. restart my consumer and monitor offset lag with
>> >> > > > > > > kafka.tools.ConsumerOffsetChecker
>> >> > > > > > >
>> >> > > > > > > After step 4 my consumer was able to continually consume
>> >>data
>> >> from
>> >> > > topic
>> >> > > > > > > but the offset lag remained unchanged. Did I do anything
>> >>wrong?
>> >> > > > > > >
>> >> > > > > > > --
>> >> > > > > > > Regards,
>> >> > > > > > > Tao
>> >> > > > > >
>> >> > > > > >
>> >> > > > >
>> >> > > > >
>> >> > > > > --
>> >> > > > > Regards,
>> >> > > > > Tao
>> >> > > >
>> >> > >
>> >> > >
>> >> >
>> >> >
>> >> > --
>> >> > Regards,
>> >> > Tao
>> >>
>> >>
>> >
>> >
>> >--
>> >Regards,
>> >Tao
>>
>>
>
>
>--
>Regards,
>Tao
