转载请注明原创地址：http://www.cnblogs.com/dongxiao-yang/p/7600561.html
一 broker启动后ReplicaFetcherThread OOM
版本：0.8.2.2
错误现象，server启动日志：
WARN [ReplicaFetcherThread-1-21], Error in fetch Name: FetchRequest; Version: 0; CorrelationId: 8; ClientId: ReplicaFetcherThread-1-21; ReplicaId: 20; MaxWait: 500 ms; MinBytes: 1 bytes; RequestInfo: [topic_a,28] -> PartitionFetchInfo(13372825,400000000),[topic_b-live,19] -> PartitionFetchInfo(2382784277,400000000),[topic_b-user,2] -> PartitionFetchInfo(2230117428,400000000),[topic_b_like,20] -> PartitionFetchInfo(0,400000000),[topic_b-group,32] -> PartitionFetchInfo(0,400000000),[topic_b_live,1] -> PartitionFetchInfo(1352237,400000000),[topic_b-log,1] -> PartitionFetchInfo(12598395616,400000000),[topic_b_microvideo,18] -> PartitionFetchInfo(0,400000000),[topic_b-group,92] -> PartitionFetchInfo(0,400000000),[bl_generic_event_detail,10] -> PartitionFetchInfo(27302121,400000000),[__consumer_offsets,14] -> PartitionFetchInfo(0,400000000),[__consumer_offsets,18] -> PartitionFetchInfo(0,400000000),[topic_b_group,12] -> PartitionFetchInfo(7323,400000000),[topic_b_feed,19] -> PartitionFetchInfo(232378,400000000),[topic_b_nearby,26] -> PartitionFetchInfo(0,400000000),[topic_b_user,20] -> PartitionFetchInfo(265,400000000),[topic_b_banners,2] -> PartitionFetchInfo(0,400000000). Possible cause: java.lang.OutOfMemoryError: Java heap space (kafka.server.ReplicaFetcherThread)
[2017-08-10 11:09:55,310] WARN [ReplicaFetcherThread-3-23], Error in fetch Name: FetchRequest; Version: 0; CorrelationId: 5; ClientId: ReplicaFetcherThread-3-23; ReplicaId: 20; MaxWait: 500 ms; MinBytes: 1 bytes; RequestInfo: [topic_b_microvideo,8] -> PartitionFetchInfo(37116,400000000),[topic_b-common,12] -> PartitionFetchInfo(5378147776,400000000),[topic_b_user,2] -> PartitionFetchInfo(76267,400000000),[topic_b-group,82] -> PartitionFetchInfo(0,400000000),[bl_generic_event_detail,24] -> PartitionFetchInfo(27457189,400000000),[topic_b-group,22] -> PartitionFetchInfo(0,400000000),[topic_b_like,10] -> PartitionFetchInfo(0,400000000),[topic_b-nearby,28] -> PartitionFetchInfo(471861516,400000000),[topic_b_nearby,16] -> PartitionFetchInfo(28,400000000),[topic_b_feed,9] -> PartitionFetchInfo(79605,400000000),[topic_a,10] -> PartitionFetchInfo(13365304,400000000),[topic_b-live,1] -> PartitionFetchInfo(2381944301,400000000),[topic_b_log,13] -> PartitionFetchInfo(366613,400000000). Possible cause: java.nio.channels.ClosedSelectorException (kafka.server.ReplicaFetcherThread)
[2017-08-10 11:09:55,310] INFO Reconnect due to socket error: java.nio.channels.ClosedChannelException (kafka.consumer.SimpleConsumer)
[2017-08-10 11:09:55,671] ERROR OOME with size 491278751 (kafka.network.BoundedByteBufferReceive)
java.lang.OutOfMemoryError: Java heap space
原因 ：replica.fetch.max.bytes参数设置的太大，在集群正常运行的情况下不会有问题，一旦某台broker挂掉重启后需要追的分区数过多，每个需要追数的partition都是申请一块replica.fetch.max.bytes大小的内存造成OOM。
解决办法：减少replica.fetch.max.bytes的值
二 重启后分区完全无法自平衡leader
版本：0.8.2.2
错误现象：1 broker挂掉重启后启动正常，追数据正常，所有副本进入isr后controller不会自动指定为leader。所有服务端日志正常，crontroller不打日志
2 jstack crontroller 线程栈发现delete-topics-thread一直block，另有Controller-xx-to-broker-yy-send-thread block在deleteTopicStopReplicaCallback位置
"Controller-29-to-broker-26-send-thread" prio=10 tid=0x00007f6520041000 nid=0x4d9d waiting on condition [0x00007f649e90e000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x00000005d3831d38> (a java.util.concurrent.locks.ReentrantLock$NonfairSync)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:838)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:871)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1201)
at java.util.concurrent.locks.ReentrantLock$NonfairSync.lock(ReentrantLock.java:214)
at java.util.concurrent.locks.ReentrantLock.lock(ReentrantLock.java:290)
at kafka.utils.Utils$.inLock(Utils.scala:533)
at kafka.controller.TopicDeletionManager.kafka$controller$TopicDeletionManager$$deleteTopicStopReplicaCallback(TopicDeletionManager.scala:371)
at kafka.controller.TopicDeletionManager$$anonfun$startReplicaDeletion$2$$anonfun$apply$3.apply(TopicDeletionManager.scala:338)
at kafka.controller.TopicDeletionManager$$anonfun$startReplicaDeletion$2$$anonfun$apply$3.apply(TopicDeletionManager.scala:338)
at kafka.controller.ControllerBrokerRequestBatch$$anonfun$addStopReplicaRequestForBrokers$2$$anonfun$apply$mcVI$sp$2.apply(ControllerChannelManager.scala:229)
at kafka.controller.ControllerBrokerRequestBatch$$anonfun$addStopReplicaRequestForBrokers$2$$anonfun$apply$mcVI$sp$2.apply(ControllerChannelManager.scala:229)
at kafka.controller.RequestSendThread.doWork(ControllerChannelManager.scala:160)
- locked <0x000000065d279300> (a java.lang.Object)
at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60)
"delete-topics-thread-29" prio=10 tid=0x00007f652006c000 nid=0x4da3 waiting on condition [0x00007f649dbdd000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x000000065d2800e0> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2043)
at java.util.concurrent.LinkedBlockingQueue.put(LinkedBlockingQueue.java:282)
at kafka.controller.ControllerChannelManager.sendRequest(ControllerChannelManager.scala:57)
- locked <0x000000065d279858> (a java.lang.Object)
at kafka.controller.KafkaController.sendRequest(KafkaController.scala:668)
at kafka.controller.ControllerBrokerRequestBatch$$anonfun$sendRequestsToBrokers$3$$anonfun$apply$9.apply(ControllerChannelManager.scala:312)
at kafka.controller.ControllerBrokerRequestBatch$$anonfun$sendRequestsToBrokers$3$$anonfun$apply$9.apply(ControllerChannelManager.scala:309)
at scala.collection.immutable.List.foreach(List.scala:318)
at kafka.controller.ControllerBrokerRequestBatch$$anonfun$sendRequestsToBrokers$3.apply(ControllerChannelManager.scala:309)
at kafka.controller.ControllerBrokerRequestBatch$$anonfun$sendRequestsToBrokers$3.apply(ControllerChannelManager.scala:302)
at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
at kafka.controller.ControllerBrokerRequestBatch.sendRequestsToBrokers(ControllerChannelManager.scala:302)
at kafka.controller.ReplicaStateMachine.handleStateChanges(ReplicaStateMachine.scala:115)
at kafka.controller.TopicDeletionManager$$anonfun$startReplicaDeletion$2.apply(TopicDeletionManager.scala:337)
at kafka.controller.TopicDeletionManager$$anonfun$startReplicaDeletion$2.apply(TopicDeletionManager.scala:327)
at scala.collection.immutable.Map$Map1.foreach(Map.scala:109)
at kafka.controller.TopicDeletionManager.startReplicaDeletion(TopicDeletionManager.scala:327)
at kafka.controller.TopicDeletionManager.kafka$controller$TopicDeletionManager$$onPartitionDeletion(TopicDeletionManager.scala:360)
at kafka.controller.TopicDeletionManager$$anonfun$kafka$controller$TopicDeletionManager$$onTopicDeletion$2.apply(TopicDeletionManager.scala:306)
at kafka.controller.TopicDeletionManager$$anonfun$kafka$controller$TopicDeletionManager$$onTopicDeletion$2.apply(TopicDeletionManager.scala:305)
at scala.collection.immutable.Set$Set1.foreach(Set.scala:74)
at kafka.controller.TopicDeletionManager.kafka$controller$TopicDeletionManager$$onTopicDeletion(TopicDeletionManager.scala:305)
at kafka.controller.TopicDeletionManager$DeleteTopicsThread$$anonfun$doWork$1$$anonfun$apply$mcV$sp$4.apply(TopicDeletionManager.scala:424)
at kafka.controller.TopicDeletionManager$DeleteTopicsThread$$anonfun$doWork$1$$anonfun$apply$mcV$sp$4.apply(TopicDeletionManager.scala:396)
at scala.collection.immutable.Set$Set1.foreach(Set.scala:74)
at kafka.controller.TopicDeletionManager$DeleteTopicsThread$$anonfun$doWork$1.apply$mcV$sp(TopicDeletionManager.scala:396)
at kafka.controller.TopicDeletionManager$DeleteTopicsThread$$anonfun$doWork$1.apply(TopicDeletionManager.scala:390)
at kafka.controller.TopicDeletionManager$DeleteTopicsThread$$anonfun$doWork$1.apply(TopicDeletionManager.scala:390)
at kafka.utils.Utils$.inLock(Utils.scala:535)
at kafka.controller.TopicDeletionManager$DeleteTopicsThread.doWork(TopicDeletionManager.scala:390)
at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60)
原因：kafka版本bug，原因参考https://issues.apache.org/jira/browse/KAFKA-2046和https://issues.apache.org/jira/browse/KAFKA-2122，controller内部事件处理共用一个与controller.message.queue.size大小相关的队列，如果这个参数配置为一个过小的值的话，会导致controller内部线程互相block。
解决办法：调大controller.message.queue.size或者注销掉（默认值为Int.MaxValue）然后滚动重启所有状态不正常的controller，直到controller落到一个配置正常的broker上