什么情况下会导致一个副本与 leader 失去同步

一个副本与 leader 失去同步的原因有很多，主要包括：

慢副本（Slow replica）：follower replica 在一段时间内一直无法赶上 leader 的写进度。造成这种情况的最常见原因之一是 follower replica 上的 I/O瓶颈，导致它持久化日志的时间比它从 leader 消费消息的时间要长；
卡住副本（Stuck replica）：follower replica 在很长一段时间内停止从 leader 获取消息。这可能是以为 GC 停顿，或者副本出现故障；
刚启动副本（Bootstrapping replica）：当用户给某个主题增加副本因子时，新的 follower replicas 是不同步的，直到它跟上 leader 的日志。

当副本落后于 leader 分区时，这个副本被认为是不同步或滞后的。在 Kafka 0.8.2 中，副本的滞后于 leader 是根据 replica.lag.max.messages 或 replica.lag.time.max.ms 来衡量的；前者用于检测慢副本（Slow replica），而后者用于检测卡住副本（Stuck replica）。

如何确认某个副本处于滞后状态

通过 replica.lag.time.max.ms 来检测卡住副本（Stuck replica）在所有情况下都能很好地工作。它跟踪 follower 副本没有向 leader 发送获取请求的时间，通过这个可以推断 follower 是否正常。另一方面，使用消息数量检测不同步慢副本（Slow replica）的模型只有在为单个主题或具有同类流量模式的多个主题设置这些参数时才能很好地工作，但我们发现它不能扩展到生产集群中所有主题。

在我之前的示例的基础上，如果主题 foo 以 2 msg/sec 的速率写入数据，其中 leader 收到的单个批次通常永远不会超过3条消息，那么我们知道这个主题的 replica.lag.max.messages 参数可以设置为 4。为什么？因为我们以最大速度往 leader 写数据并且在 follower 副本复制这些消息之前，follower 的日志落后于 leader 不超过3条消息。同时，如果主题 foo 的 follower 副本始终落后于 leader 超过3条消息，则我们希望 leader 删除慢速 follower 副本以防止消息写入延迟增加。

这本质上是 replica.lag.max.messages 的目标 - 能够检测始终与 leader 不同步的副本。假设现在这个主题的流量由于峰值而增加，生产者最终往 foo 发送了一批包含4条消息，等于 replica.lag.max.messages = 4 的配置值。此时，两个 follower 副本将被视为与 leader 不同步，并被移除 ISR。

但是，由于两个 follower 副本都处于活动状态，因此它们将在下一个 fetch 请求中赶上 leader 的日志结束偏移量并被添加回 ISR。如果生产者继续向 leader 发送大量的消息，则将重复上述相同的过程。这证明了 follower 副本进出 ISR 时触发不必要的错误警报的情况。

replica.lag.max.messages 参数的核心问题是，用户必须猜测如何配置这个值，因为我们不知道 Kafka 的传入流量到底会到多少，特别是在网络峰值的情况下。

一个参数搞定一切

我们意识到，检测卡住或慢速副本真正重要的事情，是副本与 leader 不同步的时间。我们删除了通过猜测来设置的 replica.lag.max.messages 参数。现在，我们只需要在服务器上配置 replica.lag.time.max.ms 参数即可；这个参数的含义为副本与 leader 不同步的时间。

检测卡住副本（Stuck replica）的方式与以前相同 - 如果副本未能在 replica.lag.time.max.ms 时间内发送 fetch 请求，则会将其视为已死的副本并从 ISR 中删除；
检测慢副本的机制已经改变 - 如果副本落后于 leader 的时间超过 replica.lag.time.max.ms，则认为它太慢并且从 ISR 中删除。

因此，即使在峰值流量下，生产者往 leader 发送大量的消息，除非副本始终和 leader 保持 replica.lag.time.max.ms 时间的落后，否则它不会随机进出 ISR。

常见问题（ISR）

Leader丢失导致的Consumer挂起故障解决

当时的现象是Leader的值均为-1，Isr的值也均为-1

至此我们猜测是因为KakfaController丢失导致的partition leader为-1，进而导致的Consumer端无法正常消费。

我们可以发现Replica的状态机管理是在KafkaController中完成的，也就是说Controller丢失的情况下，也就失去了与Zookeeper交互的能力。默认情况下Leader必须从ISR列表中选择，我们发现列表为空（经过排查发现是Kafka的bug，在Controller和Zookeeper通信过程中出现问题时，可能导致leader丢失而无法通信的情况，这个可能性是很大的，因为zookeeper在高并发环境是容易超时，这就是为什么在kafka 0.8.2.1之后更建议我们使用kafka topic的方式存储offset而不是存储在zookeeper中。

kafka broker Leader -1引起消费异常

Kafka Broker Leader 为-1，表示有partition在选举Leader的时候失败了，因此引起了消费该Topic的实时任务都出现了异常，经过排除发现挂掉的cdh-003机器正好是broker id为257。(但为何192没有被选举为leader呢？？)

192没选举为leader的原因是选择了unclean.leader.election.enable=false，即非ISR不能参与leader选举

kafka producer生产数据server端堆外内存频繁OOM

先调发送缓存大小。

buffer.memory 生产者用来缓存等待发送到服务器的消息的内存总字节数。如果消息发送比可传递到服务器的快，生产者将阻塞max.block.ms之后，抛出异常。
此设置应该大致的对应生产者将要使用的总内存，但不是硬约束，因为生产者所使用的所有内存都用于缓冲。一些额外的内存将用于压缩（如果启动压缩），以及用于保持发送中的请求。

比如 Topic: __consumer_offsets Partition: 8 Leader: 2 Replicas: 2,4,1,3 Isr: 2
isr 只有一个 2, 一旦 2 节点掉了,是不是部分数据就没办法访问

root@zx001 kafka]# su kafka
[kafka@zx001 kafka]$ bin/kafka-topics.sh --zookeeper zx008:2181 --topic  __consumer_offsets  --describe
Topic:__consumer_offsets        PartitionCount:50       ReplicationFactor:4     Configs:segment.bytes=104857600,cleanup.policy=compact,compression.type=uncompressed
        Topic: __consumer_offsets       Partition: 0    Leader: 2       Replicas: 2,1,3,4       Isr: 2
        Topic: __consumer_offsets       Partition: 1    Leader: 3       Replicas: 3,2,4,1       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 2    Leader: 4       Replicas: 4,3,1,2       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 3    Leader: 1       Replicas: 1,4,2,3       Isr: 1,3,4,2
        Topic: __consumer_offsets       Partition: 4    Leader: 2       Replicas: 2,3,4,1       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 5    Leader: 3       Replicas: 3,4,1,2       Isr: 1,3,4,2
        Topic: __consumer_offsets       Partition: 6    Leader: 4       Replicas: 4,1,2,3       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 7    Leader: 1       Replicas: 1,2,3,4       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 8    Leader: 2       Replicas: 2,4,1,3       Isr: 2
        Topic: __consumer_offsets       Partition: 9    Leader: 3       Replicas: 3,1,2,4       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 10   Leader: 4       Replicas: 4,2,3,1       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 11   Leader: 1       Replicas: 1,3,4,2       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 12   Leader: 2       Replicas: 2,1,3,4       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 13   Leader: 3       Replicas: 3,2,4,1       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 14   Leader: 4       Replicas: 4,3,1,2       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 15   Leader: 1       Replicas: 1,4,2,3       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 16   Leader: 2       Replicas: 2,3,4,1       Isr: 1,3,4,2
        Topic: __consumer_offsets       Partition: 17   Leader: 3       Replicas: 3,4,1,2       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 18   Leader: 4       Replicas: 4,1,2,3       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 19   Leader: 1       Replicas: 1,2,3,4       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 20   Leader: 2       Replicas: 2,4,1,3       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 21   Leader: 3       Replicas: 3,1,2,4       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 22   Leader: 4       Replicas: 4,2,3,1       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 23   Leader: 1       Replicas: 1,3,4,2       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 24   Leader: 2       Replicas: 2,1,3,4       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 25   Leader: 3       Replicas: 3,2,4,1       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 26   Leader: 4       Replicas: 4,3,1,2       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 27   Leader: 1       Replicas: 1,4,2,3       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 28   Leader: 2       Replicas: 2,3,4,1       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 29   Leader: 3       Replicas: 3,4,1,2       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 30   Leader: 4       Replicas: 4,1,2,3       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 31   Leader: 1       Replicas: 1,2,3,4       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 32   Leader: 2       Replicas: 2,4,1,3       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 33   Leader: 3       Replicas: 3,1,2,4       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 34   Leader: 4       Replicas: 4,2,3,1       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 35   Leader: 1       Replicas: 1,3,4,2       Isr: 1,3,4,2
        Topic: __consumer_offsets       Partition: 36   Leader: 2       Replicas: 2,1,3,4       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 37   Leader: 3       Replicas: 3,2,4,1       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 38   Leader: 4       Replicas: 4,3,1,2       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 39   Leader: 1       Replicas: 1,4,2,3       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 40   Leader: 2       Replicas: 2,3,4,1       Isr: 2
        Topic: __consumer_offsets       Partition: 41   Leader: 3       Replicas: 3,4,1,2       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 42   Leader: 4       Replicas: 4,1,2,3       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 43   Leader: 1       Replicas: 1,2,3,4       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 44   Leader: 2       Replicas: 2,4,1,3       Isr: 2
        Topic: __consumer_offsets       Partition: 45   Leader: 3       Replicas: 3,1,2,4       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 46   Leader: 4       Replicas: 4,2,3,1       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 47   Leader: 1       Replicas: 1,3,4,2       Isr: 1,4,3,2
        Topic: __consumer_offsets       Partition: 48   Leader: 2       Replicas: 2,1,3,4       Isr: 2
        Topic: __consumer_offsets       Partition: 49   Leader: 3       Replicas: 3,2,4,1       Isr: 1,4,3,2
[kafka@zx001 kafka]$

是的，部分消费者会受影响。
节点恢复后会自行恢复。

0 0

•

ighack 半兽人 . 10月前

我想问一下如何恢复这些失效的节点。只能重启吗？

0 0

•

回复

半兽人 ighack . 10月前

嗯系统级别的故障。失效的节点只能通过重启恢复

如图所示，isr中还有机器，但是leader还是-1，没有选举到isr中的机器

个人的猜测原因是，最后leader 和 isr 都只剩一个了，且leader和isr的broker id是一样，但是后续leader 也出问题了，所以显示的isr有broker，其实是最后控制partition分区的leader的broker id，而且最后因为一些原因也挂掉了，ISR也停留在了最后一个leader id上。

如何确认不重复消费和丢数据的话，生产者用同步消息确认，消费者用手动提交offset

消费者无法消费数据

今天特意研究了一下这个问题，专门来登录回答一下楼主的疑问：

其实消费者无法消费有如下两种情况：

1）就是consumer_offsets副本数为1，当你所使用的消费者分组所对应的consumer_offsets分区所在的broker挂掉了的时候，消费者就无法读取offset和提交offset了，这就是楼主所分析的这种现象，通过根据集群规模修改offsets.topic.replication.factor大于1即可解决。
2）就是楼主的“未解之谜”，当你在未创建topic的情况下直接使用kafka-console-producer.sh创建生产者的时候，所创建的topic的默认分区数为1，副本数为1，也就是说此时此topic的所有分区和副本都只存在于一个kafka broker上（一般都是第一个启动的broker，具体源码还没来得及看），此时，无论你所用的消费者所属的consumer_offsets分区再哪个broker上，只要broker1挂了，生产者和消费者都不可用，因为topic所在分区挂了，生产者肯定无法写数据了，那么消费者需要从topic分区fetch数据，肯定也不行了。
如果，topic分区副本和消费者组所属的consumer-offsets不在一个broker上的话，此时，如果topic分区所在的broker挂了，那么生产者无法使用，如果消费者组所属的__consumer-offsets分区所在的broker挂了，那么消费者不可用。

经过我的反复测试，在三节点的集群中，只要把所使用的topic副本数设置为大于1即可，offsets.topic.replication.factor大于1即可解决这种问题！

一个消费者组消费消息的offset只会存在__consumer_offsets 里的一个分区上面，所以如果副本只有一份，那就只会在特定的broker上

最近碰到一个问题，多个业务往向一个Kafka topic发送消息，有些业务的消费量很大，有些业务的消息量很小。因Kafka尚未较好地支持按优先级来消费消息，导致某些业务的消息消费延时的问题。一种简单的解决方案是再增加几个Topic，面对一些系统遗留问题，增加Topic带来的是生产者和消费者处理逻辑复杂性。一种方法是使用Kafka Standalone consumer，先使用consumer.partitionFor("TOPIC_NAME")获取topic下的所有分区信息，再使用consumer.assign(partitions)显示地为consumer指定消费分区。另一种方法是基于consumer group 自定义Kafka consumer的分区分配策略，那这时候就得对Kafka目前已有的分区分配策略有所了解，并且明白什么时候、什么场景下触发rebalance？

Kafka consumer要消费消息，哪些的分区的消息交给哪个consumer消费呢？这是consumer的分区分配策略，默认有三个：range、round-robin、sticky。说到round-robin这个算法，真是无处不在，它经常用在一些需要负载均衡的场景。比如Elasticsearch client向ES Server发送搜索请求时，因为默认情况下每台ES节点都可做为coordinator节点接收用户的查询请求，而在coordinator节点上需要汇总所有分片的查询结果，这需要消耗大量的内存和CPU，因此ES Client 也是基于round-robin算法选择将查询请求发送到哪个ES节点上。如果你仔细留意，会发现在Redis里面也会有这个算法的身影。再比如说：Redis LRU Cache中关于Key的access pattern，一般服从幂指分布(power-law distribution)：具有某一特征的一小部分的Key访问频率远远大于其他的Key，正如这种访问特性，LRU能达到很好的缓存效果。另外，Redis sorted set类型是基于skiplist实现，新的skipNode节点最大层数设置为多少合适呢？这也是个power-law distribution问题，其源码注释中：

Returns a random level for the new skiplist node we are going to create. The return value of this function is between 1 and ZSKIPLIST_MAXLEVEL (both inclusive), with a powerlaw-alike distribution where higher levels are less likely to be returned.

其实，我想表达的是有些思想或者说是解决方案，它是通用的，应用于各个不同的存储系统中，将它们对比起来看，能更好地理解系统背后的原理。

最近每次想写一些笔记时，脑海里总是出现一些其他各种各样的想法。这次本来主要是想写kafka 中这两个配置参数：session.timeout.ms 和 heartbeat.interval.ms的区别的，结果就先扯了一通数据存储相关的东西。

下面继续：

因为一个topic往往有多个分区，而我们又会在一个consumer group里面创建多个消费者消费这个topic，因此：就有了一个问题：哪些的分区的消息交给哪个consumer消费呢？这里涉及到三个概念：consumer group，consumer group里面的consumer，以及每个consumer group有一个 group coordinator。conusmer分区分配是通过组管理协议来实施的：具体如下：

consumer group里面的各个consumer都向 group coordinator发送JoinGroup请求，这样group coordinator就有了所有consumer的成员信息，于是它从中选出一个consumer作为Leader consumer，并告诉Leader consumer说：你拿着这些成员信息和我给你的topic分区信息去安排一下哪些consumer负责消费哪些分区吧

接下来，Leader consumer就根据我们配置的分配策略(由参数partition.assignment.strategy指定)为各个consumer计算好了各自待消费的分区。于是，各个consumer向 group coordinator 发送SyncGroup请求，但只有Leader consumer的请求中有分区分配策略，group coordinator 收到leader consumer的分区分配方案后，把该方案下发给各个consumer。画个图，就是下面这样的：

而在正常情况下，当有consumer进出consumer group时就会触发rebalance，所谓rebalance就是重新制订一个分区分配方案。而制订好了分区分配方案，就得及时告知各个consumer，这就与 heartbeat.interval.ms参数有关了。具体说来就是：每个consumer 都会根据 heartbeat.interval.ms 参数指定的时间周期性地向group coordinator发送 hearbeat，group coordinator会给各个consumer响应，若发生了 rebalance，各个consumer收到的响应中会包含 REBALANCE_IN_PROGRESS 标识，这样各个consumer就知道已经发生了rebalance，同时 group coordinator也知道了各个consumer的存活情况。

那为什么要把 heartbeat.interval.ms 与 session.timeout.ms 进行对比呢？session.timeout.ms是指：group coordinator检测consumer发生崩溃所需的时间。一个consumer group里面的某个consumer挂掉了，最长需要 session.timeout.ms 秒检测出来。举个示例session.timeout.ms=10，heartbeat.interval.ms=3

session.timeout.ms是个"逻辑"指标，它指定了一个阈值---10秒，在这个阈值内如果coordinator未收到consumer的任何消息，那coordinator就认为consumer挂了。而heartbeat.interval.ms是个"物理"指标，它告诉consumer要每3秒给coordinator发一个心跳包，heartbeat.interval.ms越小，发的心跳包越多，它是会影响发TCP包的数量的，产生了实际的影响，这也是我为什么将之称为"物理"指标的原因。

如果group coordinator在一个heartbeat.interval.ms周期内未收到consumer的心跳，就把该consumer移出group，这有点说不过去。就好像consumer犯了一个小错，就一棍子把它打死了。事实上，有可能网络延时，有可能consumer出现了一次长时间GC，影响了心跳包的到达，说不定下一个heartbeat就正常了。

而heartbeat.interval.ms肯定是要小于session.timeout.ms的，如果consumer group发生了rebalance，通过心跳包里面的REBALANCE_IN_PROGRESS，consumer就能及时知道发生了rebalance，从而更新consumer可消费的分区。而如果超过了session.timeout.ms，group coordinator都认为consumer挂了，那也当然不用把 rebalance信息告诉该consumer了。

在kafka0.10.1之后的版本中，将session.timeout.ms 和 max.poll.interval.ms 解耦了。也就是说：new KafkaConsumer对象后，在while true循环中执行consumer.poll拉取消息这个过程中，其实背后是有2个线程的，即一个kafka consumer实例包含2个线程：一个是heartbeat 线程，另一个是processing线程，processing线程可理解为调用consumer.poll方法执行消息处理逻辑的线程，而heartbeat线程是一个后台线程，对程序员是"隐藏不见"的。如果消息处理逻辑很复杂，比如说需要处理5min，那么 max.poll.interval.ms可设置成比5min大一点的值。而heartbeat 线程则和上面提到的参数 heartbeat.interval.ms有关，heartbeat线程每隔heartbeat.interval.ms向coordinator发送一个心跳包，证明自己还活着。只要 heartbeat线程在 session.timeout.ms 时间内向 coordinator发送过心跳包，那么 group coordinator就认为当前的kafka consumer是活着的。
在kafka0.10.1之前，发送心跳包和消息处理逻辑这2个过程是耦合在一起的，试想：如果一条消息处理时长要5min，而session.timeout.ms=3000ms，那么等 kafka consumer处理完消息，group coordinator早就将consumer 移出group了，因为只有一个线程，在消息处理过程中就无法向group coordinator发送心跳包，超过3000ms未发送心跳包，group coordinator就将该consumer移出group了。而将二者分开，一个processing线程负责执行消息处理逻辑，一个heartbeat线程负责发送心跳包，那么：就算一条消息需要处理5min，只要底heartbeat线程在session.timeout.ms向group coordinator发送了心跳包，那consumer可以继续处理消息，而不用担心被移出group了。另一个好处是：如果consumer出了问题，那么在 session.timeout.ms内就能检测出来，而不用等到 max.poll.interval.ms 时长后才能检测出来。

一次kafka consumer 不断地 rebalance 分析

明白了session.timeout.ms 和 max.poll.interval.ms 和 heartbeat.interval.ms三个参数的意义后，现在来实际分析一下项目中经常碰到的 consumer rebalance 错误。
一般我们是在一个线程（用户线程）里面执行kafka consumer 的while true循环逻辑的，其实这里有2个线程：一个是用户线程，另一个是心跳线程。心跳线程，我想就是根据heartbeat.interval.ms参数配置的值周期性向coordinator发送心跳包以证明consumer还活着。
如果消息处理逻辑过重，也即用户线程需要执行很长的时间处理消息，然后再提交offset。咋一看，有一个后台心跳线程在不断地发送心跳啊，那为什么group coordinator怎么还老是将consumer移出group，然后导致不断地rebalance呢？
我想，问题应该是 max.poll.interval.ms这个参数引起的吧，因为在ERROR日志中，老是提示：消息处理逻辑花了太长的时间，要么减少max.poll.records值，要么增大session.timeout.ms的值。尽管有后台heartbeat 线程，但是如果consumer的消息处理逻辑时长超过了max.poll.interval.ms ，那么此consumer提交offset就会失败：

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.

此外，在用户线程中，一般会做一些失败的重试处理。比如通过线程池的 ThreadPoolExecutor#afterExecute()方法捕获到异常，再次提交Runnable任务重新订阅kafka topic。本来消费处理需要很长的时间，如果某个consumer处理超时：消息处理逻辑的时长大于max.poll.interval.ms (或者消息处理过程中发生了异常)，被coordinator移出了consumer组，这时由于失败的重试处理，自动从线程池中拿出一个新线程作为消费者去订阅topic，那么意味着有新消费者加入group，就会引发 rebalance，而可悲的是：新的消费者还是来不及处理完所有消息，又被移出group。如此循环，就发生了不停地 rebalance 的现象。

kafka的rebalance机制

在Kafka中，当有新消费者加入或者订阅的Topic数发生变化时，会触发Rebalance(再均衡：在同一个消费者组当中，分区的所有权从一个消费者转移到另外一个消费者)机制，Rebalance顾名思义就是重新均衡消费者消费。Rebalance的过程如下：

第一步：所有消费成员都向Coordinator发送请求，请求入Consumer Group。一旦所有成员都发送了请求，Coordinator会从中选择一个Consumer担任Leader的角色，并把组成员信息以及订阅信息发给Leader。

第二步：Leader开始分配消费方案，指明具体哪个Consumer负责消费哪些Topic的哪些Partition。一旦完成分配，leader会将这个方案发给Coordinator。Coordinator接收到分配方案之后会把方案发给各个Consumer，这样组内的所有成员就都知道自己应该消费哪些分区了。

所以对于Rebalance来说，Coordinator起着至关重要的作用

二、rebalance可能发生的时机

1、分区个数的增加

2、对Topic的订阅发生变化

3、消费组成员的加入或离开（这个是我们最常遇到）

三、rebalance的影响

Rebalance对我们数据的影响主要有以下几点：

1、可能重复消费: Consumer被踢出消费组，可能还没有提交offset，Rebalance时会Partition重新分配其它Consumer,会造成重复消费，虽有幂等操作但耗费消费资源，亦增加集群压力

2、集群不稳定：Rebalance扩散到整个ConsumerGroup的所有消费者，因为一个消费者的退出，导致整个Group进行了Rebalance，并在一个比较慢的时间内达到稳定状态，影响面较大

3、影响消费速度：频繁的Rebalance反而降低了消息的消费速度，大部分时间都在重复消费和Rebalance

四、避免rebalance措施

1、业务需要不可避免

（1）针对分区个数的增加，一般不会常有，是需要增加的时候都是业务及数据需求，不可避免

（2）对Topic的订阅增加或取消亦不可避免

2、合理设置消费者参数

下边是我们遇到的，要格外关注及重视

（1）未能及时发送心跳而Rebalance

session.timeout.ms 一次session的连接超时时间

heartbeat.interval.ms 心跳时间，一般为超时时间的1/3，Consumer在被判定为死亡之前，能够发送至少 3 轮的心跳请求

（2）Consumer消费超时而Rebalance

max.poll.interval.ms 每隔多长时间去拉取消息。合理设置预期值，尽量但间隔时间消费者处理完业务逻辑，否则就会被coordinator判定为死亡，踢出Consumer Group，进行Rebalance

max.poll.records 一次从拉取出来的数据条数。根据消费业务处理耗费时长合理设置，如果每次max.poll.interval.ms 设置的时间较短，可以max.poll.records设置小点儿，少拉取些，这样不会超时。

总之，尽可能在max.poll.interval.ms时间间隔内处理完max.poll.records条消息，让Coordinator认为消费Consumer还活着

1 flume的source是kafka，sink是hdfs，怎样判断flume是否堆积，或者是说怎么样保证落地的速度和消费的速度是平衡的

2 怎么判断flume的agent程序是否挂掉

3 挂掉时tmp文件爱呢怎么处理（hdfs上的tmp文件）

4 我遇到一个问题，当agent是6个时，一小时约生成26.5G文件，当有3个agent时，一小时16.9G，按我的理解，1个agent相当于一个consumer，这6个agent是同一个consumer group，那么3个和6个agent1小时都应该生成那么多的文件，但实际不是，怎么回事？

先说第4个问题，其实同事理解的不错，kafka的consumer必须归于一个group，以group为单位去消费topic上的数据，group内的consumer不论多或少，最终输出的数据量按理应该是一样多的。至于为啥当consumer多时消费量多，consumer少时消费少，那就说明了，每个consumer都到了它所能处理的上限了，而topic中整体的数据量太大。我的建议是增大consumer的数量，果然当增大consumer的数量后，消费的总量也上去了，说明consumer的数量还有提高的空间，当然，不论消费的多少，数据是不会丢的

再来说第一个问题，因为数据量很大了，conusmer的处理能力达到上限，然后flume作为kafka 的consumer，出现了消息的积压（这哥们在flume中加了个interceptor），按理说，我flume消费多少，我就从kafka中pull多少，根据我的消费能力我来消费数据，如果真是这样就不会有堆积了，kafkasource 默认batchsize 是1000条数据，而且auto.commit.enable 可能设置为true，所以，在flume往channel写的中间，一个interceptor处理能力上不去的话，kafkasource不停的pull数据过来，就导致了数据的来不及处理，有积压的出现。我的理解kafkasource 和 interceptor 之间或许没有建立一种ack机制，kafkasource 根据自己的batchsize和commit设置去不断的拉取数据，在interceptor处出现了瓶颈，如果source和interceptor能统一起来，就解决问题了。或者治标的办法是，降低batchsize 的大小，设置为500，auto.commit.enable = false 。在没改设置之前，出现了kafkasource的内存溢出，报的是由于GC导致的OOM，我分析了他自定义的interceptor代码，发现，它在event processor处理方法内，new 对象过多，对代码进行了优化后，问题暂时解决。治本的方法就是在可以在batch event processor中加上offset的处理，例如在init方法中初始化一个zookeeper的客户端，在batch event processor中对offset更新到zookeeper中，或者是更新到kafka的一个topic中。kafkasource根据offset来pull数据（或者是ack机制，当处理完这一批数据后，ack后，kafkasource再去拉取下一批数据，当数据没有处理成功的话，超时的话，也直接拉取下一批数据，有数据丢失的情况）

第二个问题，如何判断agent挂掉，可以在interceptor中做文章，例如在init方法中初始化一个zookeeper客户端，然后在zk上注册一个watch，当agent挂掉后，interceptor也就挂掉了，zookeeper就能监测到。

第三个问题，tmp文件是因为当sink往hdfs写的过程中，正常写入的文件被命名为tmp文件，当agent当掉后，tmp不会自动改名字，但是tmp中也会有部分数据存在，所以可以写个程序将文件名中的tmp去掉。如果agent重启后，可以续传，就保证数据连接上了。

前面说过了，同事加了一个interceptor后，会导致flume的性能下降。

同时有报了如下的错误

org.apache.flume.source.kafka.KafkaSource.doProcess(KafkaSource.java:314)] KafkaSource EXCEPTION, {}

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed due to group rebalance

at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:552)

at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:493)

at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:665)

at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:644)

我的理解是interceptor，当时处理时间过长，consumer 等待下一次pull的时间间隔过长，导致发送hearbeat时间间隔过长，coordinator认为comsumer死了，就发生了rebalance。

增大参数 hearbeat.interval.ms

缩小参数 max.partition.fetch.bytes 让每次pull的数据少点

新增多个 Flume 实例后，Kafka 数据重复消费问题处理

我们使用 Flume 将数据从 Kafka 加载到 Hive 中。

（自己）假如启动2个agent消费，会触发 reblance，假如有10个分区，flume 会自动建立10个consumer，1个groupid，如果在启动一个，触发relance，2个flume agent 平分 kafka的partition ，也就是一个5个。reblance 一般会发生在消费组这端，生产者主要是isr问题。

由于启动一个 Flume 实例时，数据加载的速度只能达到 10MB/秒 (每条Kafka记录100B)。于是我们计划启动多个 Flume 实例 (指定同一个消费者组名称)。

我们知道 Kafka 数据消费是以 Partition 为单位的，即一个 Partition 只能被一个 Flume 实例消费。当启动第二个 Flume 实例时，Kafka 会把已经分配给第一个 Flume 实例的 Partition 回收(revoke)后，然后重新分配(完整过程在这里 https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Client-side+Assignment+Proposal )。

另一方面，Flume 为了满足事务语义，需要等每条 Kafka Record 真正放到 Channel 后，才能向 Kafka 确认 (调用 consumer.commitSync)。在 Kafka 回收，分配过程 (rebalance) 中，已经接收到 Flume 端但还未被 Flume 放到 Channel 的 Records (自然也就没有向 Kafka 确认) 就被Kafka 认为未被消费；特别是属于将来被分配给其他 Flume 实例的 Partition 上的 Records，就会再次被新的 Flume 实例消费，造成重复加载。

Kafka 提供 ConsumerRebalanceListener 接口允许消费者侦听这个回收(onPartitionsRevoked)、分配(onPartitionsAssigned)过程。Flume 中 Kafka Source 在侦测到rebalance时的代码如下:

// this flag is set to true in a callback when some partitions are revoked.

// If there are any records we commit them.

if (rebalanceFlag.get()) {

rebalanceFlag.set(false);

break;

}

这个措施很简单，遇到这种情况立即跳出去处理(放到 Channel 并向 kafka 确认)。从代码 (https://github.com/apache/flume/blob/flume-1.8/flume-ng-sources/flume-kafka-source/src/main/java/org/apache/flume/source/kafka/KafkaSource.java) 上下文可以看到这里有几个缺陷:

1. 由于把 event 放到 channel 是需要时间的 (执行 interceptor 和 channel selector)，不一定能够在 partition 分配给别的 Flume 实例之前完成并提交 Kafka (调研 consumer.commitSync)

2. 那些已经读取到 records (本地) 中，但是没有放到 eventList 中的记录没有处理

所以我们修改了 Flume 源码，对这两个缺陷进行补救:

// If there are any records we commit them.

if (rebalanceFlag.get()) {

rebalanceFlag.set(false);

// invalidate remaining records

it = null;

// and drop processed events

eventList.clear();

tpAndOffsetMetadata.clear();

// and seek to committed offsets

for (Map.Entry<String, List<PartitionInfo>> es : consumer.listTopics().entrySet()) {

for (PartitionInfo pi : es.getValue()) {

TopicPartition tp = new TopicPartition(pi.topic(), pi.partition());

try {

OffsetAndMetadata oam = consumer.committed(tp);

if (oam != null) {

consumer.seek(tp, oam.offset());

}

} catch (Exception e) {

// log.warn("ignore seeking exception, {}", e);

}

log.info("read-ahead records have been dropped.");

break;

}

详细请参考 https://github.com/hejiang2000/flume/blob/hejiang-kafka-source/flume-ng-sources/flume-kafka-source/src/main/java/org/apache/flume/source/kafka/KafkaSource.java 。

经测试，以上处理完全修正 Kafka Source 数据重复消费问题。

————————————————

原文链接：https://blog.csdn.net/hejiangtju/article/details/80364388

理解：意思是忽略已经加入到eventList里面的消息，重新设置seek，然后重新读取一遍

阿里云的文档

使用消息队列Kafka版时消费客户端频繁出现Rebalance

问题原因

可能导致故障的部分原因如下：

开源版本为v0.10的消息队列Kafka版有一定的概率会触发频繁Rebalance。
消息队列Kafka版的Consumer没有独立线程维持心跳，而是把心跳维持与poll接口耦合在一起。其结果就是，如果用户消费出现卡顿，就会导致Consumer心跳超时，引发Rebalance。

解决方案

首先您需要了解以下几点信息：

session.timeout.ms：配置控制心跳的超时时间，可以由客户端自行设置。
max.poll.records：控制每次poll返回的最大消息数量。
消息队列Kafka版的心跳是通过poll接口来实现的，没有内置的独立线程。

为了避免心跳超时，引发Rebalance，请参考以下步骤进行调整：

检查消息队列Kafka版实例的开源版本。如果实例的开源版本是v0.10，建议您将实例版本升级到稳定的v0.10.2。详情请参见升级实例服务版本。
参考以下说明调整参数值：
- session.timeout.ms：适当提高该参数值，但不要超过30s，建议设置为25s。
- max.poll.records：降低该参数值，建议远远小于<单个线程每秒消费的条数> * <消费线程的个数> * <session.timeout.ms>的积。
尽量提高客户端的消费速度。

GC优化，默认是使用CMS和UseParNewGC 可以采用G1垃圾回收器

-XX:+DisableExplicitGC:禁用显式GC 禁止system.gc对gc的触发。加上后system.gc（）将不会触发gc。
-XX:+ExplicitGCInvokesConcurrent:使用并发方式处理显式GC
去掉第一条命令，采用第二条命令（也是触发full gc 只不过在CMS在full gc 效率比较高。因为是并发的执行gc，可以防止reblance和isr的缺失）
但是，如果完全不允许手动调用gc,

* 使用了NIO或者NIO框架（Mina/Netty）
* 使用了DirectByteBuffer分配字节缓冲区
* 使用了MappedByteBuffer做内存映射

以上情况下的堆外内存又无法回收，所以，才提到了用ExplicitGCInvokesConcurrent参数来替代DisableExplicitGC （这个关系让我一直疑问很久，究竟为什么要用ExplicitGCInvokesConcurrent参数来替代DisableExplicitGC，网上的发部分言论都在这么说，但是没有人清楚明白的说明清楚为什么！而这才是最重要的），当然，是要在使用CMS做full GC的时候，才能使用这个参数
1、JVM参数配置优化

如果使用的CMS GC算法，建议JVM Heap不要太大，在4GB以内就可以。JVM太大，导致Major GC或者Full GC产生的“stop the world”时间过长，导致broker和zk之间的session超时，比如重新选举controller节点和提升follow replica为leader replica。

JVM也不能过小，否则会导致频繁地触发gc操作，也影响Kafka的吞吐量。另外，需要避免CMS GC过程中的发生promotion failure和concurrent failure问题。CMSInitiatingOccupancyFraction=70可以预防concurrent failure问题，提前出发Major GC。

kafka 学习

一 Kafka 都有哪些特点？

二 Kafka的应用场景

三 Kafka 的设计架构

四 Kafka 分区的目的？

五 Kafka 是如何做到消息的有序性？

六 Kafka 的再均衡

七 Kafka 在什么情况下会出现消息丢失

Kafka 复制机制

副本在什么情况下才算跟上 leader