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