kafka offset-check工具失效的问题
kafka offset-check工具失效的问题
转载请注明原创地址http://www.cnblogs.com/dongxiao-yang/p/5414077.html
由于平时业务预警等需求，针对现在公司的kafka系统部署了几套监控系统，包括调用kafka-consumer-offset-checker.sh脚本写的lag监控，kafkaoffsetmonitor开源监控以及kafka-manager管理系统。最近发现kafka-consumer-offset-checker.sh脚本在原本运行正常的情况下一直出现"Exiting due to:null"的错误，这个问题会导致脚本直接退出无法获取完整的partition的lag信息导致报警失效。尝试把监控程序部署到其他机器又发现脚本可以正常运行。
为了搞明白问题，直接把kafka-consumer-offset-checker.sh脚本调用的kafka类ConsumerOffsetChecker拿出来进行研究，发现最后输出lag结果的方法如下
private def processPartition(zkUtils: ZkUtils,
group: String, topic: String, pid: Int) {
val topicPartition = TopicAndPartition(topic, pid)
val offsetOpt = offsetMap.get(topicPartition)
val groupDirs = new ZKGroupTopicDirs(group, topic)
val owner = zkUtils.readDataMaybeNull(groupDirs.consumerOwnerDir + "/%s".format(pid))._1
zkUtils.getLeaderForPartition(topic, pid) match {
case Some(bid) =>
val consumerOpt = consumerMap.getOrElseUpdate(bid, getConsumer(zkUtils, bid))
consumerOpt match {
case Some(consumer) =>
val topicAndPartition = TopicAndPartition(topic, pid)
val request =
OffsetRequest(immutable.Map(topicAndPartition -> PartitionOffsetRequestInfo(OffsetRequest.LatestTime, 1)))
val logSize = consumer.getOffsetsBefore(request).partitionErrorAndOffsets(topicAndPartition).offsets.head
val lagString = offsetOpt.map(o => if (o == -1) "unknown" else (logSize - o).toString)
println("%-15s %-30s %-3s %-15s %-15s %-15s %s".format(group, topic, pid, offsetOpt.getOrElse("unknown"), logSize, lagString.getOrElse("unknown"),
owner match {case Some(ownerStr) => ownerStr case None => "none"}))
case None => // ignore
}
case None =>
println("No broker for partition %s - %s".format(topic, pid))
}
}其中函数processPartition通过传入的group，topic，pid三个参数唯一确定需要计算的lag。
val logSize = consumer.getOffsetsBefore(request).partitionErrorAndOffsets(topicAndPartition).offsets.head 获取logSize
val lagString = offsetOpt.map(o => if (o == -1) "unknown" else (logSize - o).toString) 用logSize减去offsetOpt这个map里对应的partition的offset得到lag。
把kafka这个类的源码搞到intellij idea在本地进行单步调试发现同样出现了Exiting due to:null的问题，并且永远是运行到某一特定分区后就问出题，调试到
val logSize = consumer.getOffsetsBefore(request).partitionErrorAndOffsets(topicAndPartition).offsets.head这个代码报错，尝试加入try catch并打印对应bid
val consumerOpt = consumerMap.getOrElseUpdate(bid, getConsumer(zkUtils, bid))
println(" brokerid ",bid)
.............
............
try {
//val
val logSize = consumer.getOffsetsBefore(request).partitionErrorAndOffsets(topicAndPartition).offsets.head
val lagString = offsetOpt.map(o => if (o == -1) "unknown" else (logSize - o).toString)
println("%-15s %-30s %-3s %-15s %-15s %-15s %s".format(group, topic, pid, offsetOpt.getOrElse("unknown"), logSize, lagString.getOrElse("unknown"),
owner match { case Some(ownerStr) => ownerStr case None => "none" }))
}
catch {
case ex: Exception => //ignore
}研究发现对于不同的topic，出现问题的分区对应的broker id都是一样的，至此怀疑是代码环境与broker服务器之间的连通性出现问题，查了下本机以及监控环境的host配置的都是不全的，把host补全后问题解决。
后续发现kafkaoffsetmonitor以及kafka-manager出现的lag查询页面出现的分区显示不全或者数据为空的情况都通过补全host解决了。
吐槽一下kafka对于host的强依赖。