[聚合文章] Kafka运维填坑

  1. 首页
  2. 聚合文章
  3. 文章详情
消息系统 2017-10-09 55 阅读
原文链接: http://www.jianshu.com/p/d2cbaae38014
  • 前提: 只针对Kafka 0.9.0.1版本;
  • 说是运维,其实偏重于问题解决;
  • 大部分解决方案都是google而来, 我只是作了次搬运工;
  • 有些问题的解决方案未必一定是通用的, 若应用到线上请慎重;
  • 如有疏漏之处, 欢迎大家批评指正;
  • 列表:
    1. Replica无法从leader同步消息
    2. Broker到zk集群的连接不时会断开重断
    3. Broker重启耗时很久
    4. 不允许脏主选举导致Broker被强制关闭
    5. Replica从错误的Partition leader上去同步数据
    6. __consumer_offsets日志无法被清除
    7. GC问题
    8. zk和kafka部署
    9. 监控很重要

Replica无法从leader同步消息

  • 现象: 集群上某topic原来只有单复本, 增加双复本后,发现有些partition没有从leader同步数据,导致isr列表中一直没有新增的replica;
  • 日志分析:
[2017-09-20 19:37:05,265] ERROR Found invalid messages during fetch for partition [xxxx,87] offset 1503297 error Message is corrupt (stored crc = 286782282, computed crc = 400317671) (kafka.server.ReplicaFetcherThread)
[2017-09-20 19:37:05,458] ERROR Found invalid messages during fetch for partition [xxxx,75] offset 1501373 error Message found with corrupt size (0) in shallow iterator (kafka.server.ReplicaFetcherThread)
[2017-09-20 19:37:07,455] ERROR [ReplicaFetcherThread-0-5], Error due to  (kafka.server.ReplicaFetcherThread)
kafka.common.KafkaException: error processing data for partition [xxxx,87] offset 1503346
        at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2$$anonfun$apply$mcV$sp$1$$anonfun$apply$2.apply(AbstractFetcherThread.scala:147)
        at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2$$anonfun$apply$mcV$sp$1$$anonfun$apply$2.apply(AbstractFetcherThread.scala:122)
        at scala.Option.foreach(Option.scala:257)
        at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2$$anonfun$apply$mcV$sp$1.apply(AbstractFetcherThread.scala:122)
        at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2$$anonfun$apply$mcV$sp$1.apply(AbstractFetcherThread.scala:120)
        at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:99)
        at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:99)
        at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:230)
        at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40)
        at scala.collection.mutable.HashMap.foreach(HashMap.scala:99)
        at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2.apply$mcV$sp(AbractFeherThread.scala:120)
        at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2.apply(AbstractFetcherThread.scala:120)
        at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2.apply(AbstractFetcherThread.scala:120)
        at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:262)
        at kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:118)
        at kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:93)
        at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
Caused by: java.lang.RuntimeException: Offset mismatch: fetched offset = 1503346, log end offset = 1503297.
        at kafka.server.ReplicaFetcherThread.processPartitionData(ReplicaFetcherThread.scala:110)
        at kafka.server.ReplicaFetcherThread.processPartitionData(ReplicaFetcherThread.scala:42)
        at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2$$anonfun$apply$mcV$sp$1$$anonfun$apply$2.apply(AbstractFetcherThread.scala:138)
  • 解决:
    1. Kafka 0.9.0.1版本的bug: ReplicaFetcherThread stopped after ReplicaFetcherThread received a corrupted message
    2. 升级版本 或者 按上面链接中Reporter给出的简单修复来避开这个问题;
  • 深究:
    这个bug被触发实际是上下面这个导致:
    ERROR Found invalid messages during fetch for partition [qssnews_download,87] offset 1503297 error Message is corrupt (stored crc = 286782282, computed crc = 400317671) (kafka.server.ReplicaFetcherThread)
    当时触发这个bug的时, 恰逢相应的broker机器上硬盘出现了多个坏块, 但不能完全确定这个crc错误跟这个有关.这个也有个Kafka的issue: Replication issues

Broker到zk集群的连接不时会断开重断

  • 现象: broker不时地和zk重新建立session;
  • 日志分析: broker日志里报zk连接超时或不能从zk读取任何数据
  • 解决: 增加broker的zk的session timeout时间, 不能完全解决,但会改善很多;
  • 深究:
    1. 目前用的kafka集群还是相对比较稳定, 但是这个zk超时问题真是百思不得其解啊.
      broker在启动时会在zk上注册一个临时节点,表时自己已上线, 一旦session超时,此临时节点将被删除, 相当于此broker下线, 必然引起整个集群的抖动,可参考 KafkaController分析8-broker挂掉
    2. zk为何会timeout, 根本原因未能准确定位,目前看到跟诸多因素有关,比如磁盘IO, CPU负载, GC等等吧;

Broker重启耗时很久

  • 现象: broker重启下分耗时
  • 日志分析: 重启时加载所有的log segments, rebuild index;
  • 解决: 应该是stop时, 没有优雅的shutdown, 直接 kill -9导致;
  • 深究:
    1. 停止broker服务请使用kafka本身提供的脚本优雅shutdown；
    2. 在shutdown broker时确保相应的zk集群是可用状态, 否则可能无法优雅地shutdown broker.

不允许脏主选举导致Broker被强制关闭

  • 现象: 监控到集群中某台broker挂掉
  • 日志分析:
    [2016-02-25 00:29:39,236] FATAL [ReplicaFetcherThread-0-1], Halting because log truncation is not allowed for topic test, Current leader 1's latest offset 0 is less than replica 2's latest offset 151 (kafka.server.ReplicaFetcherThread)
  • 解决: 实际上是设置了 unclean.leader.election.enable=false , 然后走到了代码里下面这段逻辑 
if (leaderEndOffset < replica.logEndOffset.messageOffset) {
      // Prior to truncating the follower's log, ensure that doing so is not disallowed by the configuration for unclean leader election.
      // This situation could only happen if the unclean election configuration for a topic changes while a replica is down. Otherwise,
      // we should never encounter this situation since a non-ISR leader cannot be elected if disallowed by the broker configuration.
      if (!LogConfig.fromProps(brokerConfig.originals, AdminUtils.fetchEntityConfig(replicaMgr.zkUtils,
        ConfigType.Topic, topicAndPartition.topic)).uncleanLeaderElectionEnable) {
        // Log a fatal error and shutdown the broker to ensure that data loss does not unexpectedly occur.
        fatal("...")
        Runtime.getRuntime.halt(1)
      }

调用 Runtime.getRuntime.halt(1) 直接暴力退出了.

可参考Kafka issue: Unclean leader election and "Halting because log truncation is not allowed"

Replica从错误的Partition leader上去同步数据

  • 现象: 集群里若干台机器先后磁盘空间报警, 经查是kafka log占用大量磁盘空间,接着看log, 里面有大量的
WARN [Replica Manager on Broker 3]: While recording the replica LEO, the partition [orderservice.production,0] hasn't been created. (kafka.server.ReplicaManager)

和

ERROR [ReplicaFetcherThread-0-58], Error for partition [reptest,0] to broker 58:org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition. (kafka.server.ReplicaFetcherThread)
  • 日志分析:
    从上面的日志结合当前topic的partiton的复本和isr情况,可知是错误的 replica 从错误的 partition leader 上去同步数据了, 这理论上不应该啊;
    1. 之前每个集群因硬件原因挂掉了一台机器, 然后想删掉上面的一个partiton, 但因为kafka本身不支持partiton的删除, 就在zk上的 /brokers/[topic] 节点的内容里直接去掉了这个partiton的信息, 但是 kafka controller 并不会处理partiton减少的情况, 可参考KafkaController分析
    2. 为了触发这个topic的partition的删除, 我又迁移了其他的partiton;
    3. 然后还删除了zk上的 /controller 临时节点;
    4. 最后连自己都晕了;
    5. 然后之前坏的机器修好又上线了, 然后问题出现了;
  • 解决: 将broker都重启了一遍;
  • 深究:
    1. 最终原因没有完全确认, 发现问题的时候之前的kafka debug log被删除了;
    2. kafka 上有类以的issue: can't create as many partitions as brokers exists
    3. 尽量不要手动更新zk上的kafka相关节点内容;
    4. 考虑在kafka源码里加个delete partition的功能, 这个不会太难;

__consumer_offsets日志无法被清除

  • 现象: 集群中若干台机器磁盘空间报警, 上去查看是__consumer_offsets的一个partition占用了几十G的空间
  • 日志分析: 之前的日志被清理了,没有有效的日志了.为了debug这个问题,我把这个partition下的index和log文件打包拷贝到了测试集群, 然后重启了当前的broker, 发现了下面的日志:
[2017-09-30 10:49:36,126] ERROR [kafka-log-cleaner-thread-0], Error due to  (kafka.log.LogCleaner)
java.lang.IllegalArgumentException: requirement failed: 138296566648 messages in segment __consumer_offsets-5/00000000000000000000.log but offset map can fit only 5033164. You can increase log.cleaner.dedupe.buffer.size or decrease log.cleaner.threads
        at scala.Predef$.require(Predef.scala:219)
        at kafka.log.Cleaner$$anonfun$buildOffsetMap$4.apply(LogCleaner.scala:584)
        at kafka.log.Cleaner$$anonfun$buildOffsetMap$4.apply(LogCleaner.scala:580)
        at scala.collection.immutable.Stream$StreamWithFilter.foreach(Stream.scala:570)
        at kafka.log.Cleaner.buildOffsetMap(LogCleaner.scala:580)
        at kafka.log.Cleaner.clean(LogCleaner.scala:322)
        at kafka.log.LogCleaner$CleanerThread.cleanOrSleep(LogCleaner.scala:230)
        at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:208)
        at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)

  • 问题分析:

    结合 LogCleaner 的源码可知,是 00000000000000000000.log 这个 logSegment 的 segment.nextOffset() - segment.baseOffset 大于了 maxDesiredMapSize , 导致了 LogClean 线程的终止, 从而无法清理, 这不应该啊?! 

val segmentSize = segment.nextOffset() - segment.baseOffset
      require(segmentSize <= maxDesiredMapSize, "%d messages in segment %s/%s but offset map can fit only %d. You can in了crease log.cleaner.dedupe.buffer.size or decrease log.cleaner.threads".format(segmentSize,  log.name, segment.log.file.getName, maxDesiredMapSize))
      if (map.size + segmentSize <= maxDesiredMapSize)
        offset = buildOffsetMapForSegment(log.topicAndPartition, segment, map)
      else
        full = true
  • 解决: 我也没想到其他的好办法, 暴力删除了 00000000000000000000.log 和 00000000000000000000.index , 然后删掉了 cleaner-offset-checkpoint 中相关的项,重启broker, 日志开始了压缩清理
  • 深究:
    这个 logSegment 的 segment.nextOffset() - segment.baseOffset 大于了 maxDesiredMapSize , 猜测是有个业务是手动提交offset到这个partition, 没有控制好,导致每秒能提交8,9MByte上来;

GC问题

  • 现象: 集群报警某台broker down, 在zk上无此broker节点的注册信息
  • 日志分析:
    1. 看broker日志里报zk连接超时或不能从zk读取任何数据, 其实和上面的 Broker到zk集群的连接不时会断开重断 现象是一样的;
    2. 看broker的gc日志, 对应时间gc耗时很长, 导致 stop the world ,broker所有线程都停止工作, 自然也无法与zk保持心跳;
  • 解决: 暂时无解决方案, GC 是个大麻烦, 网上也搜了一圈, 没找到有效的解决方案, 个人水平有限, 哪位大神有什么好的方法, 可以留言给我,谢谢~

zk和kafka部署

  • zk和kafka broker 如果部署在同一台机器上, 请尽量将各自的data和log路径放在不同的磁盘, 避免磁盘io的竞争;
  • kafka对zk的波动很敏感, 因此zk最好是单独部署,保证其稳定运行;
  • 对zk不要有大量的写入操作, zk的写操作最后都会转移动leader上zk;
  • 如果采用了zk和broker是混部的方式,并且还有大量的zk写入操作,比如使用较旧版本的storm,其提交offset到zk上, 导致zk的IO较高, 在启动zk时可以加上 zookeeper.forceSync=no , 降低写盘IO, 这个配置有其副作用, 在线上使用时还需慎重;

监控很重要

  • 实时监控: 在集群上建立一个专门的topic, 监控程序实时的写入数据, 但无法写入或写入耗时达到阈值时报警, 这个实时监控真的真好用,基本上都第一时间发现问题;
  • 基础监控: cpu, 磁盘IO, 网卡流量, FD, 连接数等;
  • Topic流量监控: 监控topic的生产和消费流量, 特别是流量突增的情况, 快速找出害群之马, 可以通过kafka的jmx来获取相关的数据, 使用 Grafana 来显示和报警;

Kafka源码分析-汇总

注：本文内容来自互联网，旨在为开发者提供分享、交流的平台。如有涉及文章版权等事宜，请你联系站长进行处理。