1、概念

水位标记：

        水位或水印（watermark）一词，表示位置信息，即位移（offset）。Kafka源码中使用的名字是高水位，HW（high watermark）。

副本角色：

        Kafka分区使用多个副本(replica)提供高可用。

LEO和HW：

  每个分区副本对象都有两个重要的属性：LEO和HW。

• LEO：即日志末端位移(log end offset)，记录了该副本日志中下一条消息的位移值。如果LEO=10，那么表示该副本保存了10条消息，位移值范围是[0, 9]。另外，Leader LEO和Follower LEO的更新是有区别的。
• HW：即上面提到的水位值。对于同一个副本对象而言，其HW值不会大于LEO值。小于等于HW值的所有消息都被认为是“已备份”的（replicated）。Leader副本和Follower副本的HW更新不同。

        上图中，HW值是7，表示位移是0~7 的所有消息都已经处于“已提交状态”（committed），而LEO值是14，8~13的消息就是未完全备份（fully replicated）——为什么没有14？LEO指向的是下一条消息到来时的位移。 

消费者无法消费分区下Leader副本中位移大于分区HW的消息。

2、Follower副本何时更新LEO

        Follower副本不停地向Leader副本所在的broker发送FETCH请求，一旦获取消息后写入自己的日志中进行备份。那么Follower副本的LEO是何时更新的呢？首先我必须言明，Kafka有两套Follower副本LEO：

1. 一套LEO保存在Follower副本所在Broker的副本管理机中；
2. 另一套LEO保存在Leader副本所在Broker的副本管理机中。Leader副本机器上保存了所有的follower副本的LEO。

Kafka使用前者帮助Follower副本更新其HW值；利用后者帮助Leader副本更新其HW。

（1）Follower副本的本地LEO何时更新？

        Follower副本的LEO值就是日志的LEO值，每当新写入一条消息，LEO值就会被更新。当Follower发送FETCH请求后，Leader将数据返回给Follower，此时Follower开始Log写数据，从而自动更新LEO值。

（2）Leader端Follower的LEO何时更新？

        Leader端的Follower的LEO更新发生在Leader在处理Follower FETCH请求时。一旦Leader接收到Follower发送的FETCH请求，它先从Log中读取相应的数据，给Follower返回数据前，先更新Follower的LEO。

3、Follower副本何时更新HW

• Follower更新HW发生在其更新LEO之后，一旦Follower向Log写完数据，尝试更新自己的HW值。
• 比较当前LEO值与FETCH响应中Leader的HW值，取两者的小者作为新的HW值。
• 即：如果Follower的LEO大于Leader的HW，Follower HW值不会大于Leader的HW值。

 Leader副本何时更新LEO？

        和Follower更新LEO相同，Leader写Log时自动更新自己的LEO值。

4、Leader副本何时更新HW值

Leader的HW值就是分区HW值，直接影响分区数据对消费者的可见性 。

Leader会尝试去更新分区HW的四种情况：

• Follower副本成为Leader副本时：Kafka会尝试去更新分区HW。
• Broker崩溃导致副本被踢出ISR时：检查下分区HW值是否需要更新是有必要的。
• 生产者向Leader副本写消息时：因为写入消息会更新Leader的LEO，有必要检查HW值是否需要更新
• Leader处理Follower FETCH请求时：首先从Log读取数据，之后尝试更新分区HW值

结论：

当Kafka broker都正常工作时，分区HW值的更新时机有两个：

• Leader处理PRODUCE请求时
• Leader处理FETCH请求时。

        Leader如何更新自己的HW值？Leader broker上保存了一套Follower副本的LEO以及自己的LEO。当尝试确定分区HW时，它会选出所有满足条件的副本，比较它们的LEO(包括Leader的LEO)，并选择最小的LEO值作为HW值。

需要满足的条件，（二选一）：

1. 处于ISR中
2. 副本LEO落后于Leader LEO的时长不大于replica.lag.time.max.ms 参数值(默认是10s)

        如果Kafka只判断第一个条件的话，确定分区HW值时就不会考虑这些未在ISR中的副本，但这些副本已经具备了“立刻进入ISR”的资格，因此就可能出现分区HW值越过ISR中副本LEO的情况——不允许。因为分区HW定义就是ISR中所有副本LEO的最小值。

5、HW和LEO正常更新案例

        我们假设有一个topic，单分区，副本因子是2，即一个Leader副本和一个Follower副本。我们看下当producer发送一条消息时，broker端的副本到底会发生什么事情以及分区HW是如何被更新的。

（1）初始状态

        初始时Leader和Follower的HW和LEO都是0(严格来说源代码会初始化LEO为-1，不过这不影响之后的讨论)。Leader中的Remote LEO指的就是Leader端保存的Follower LEO，也被初始化成0。此时，生产者没有发送任何消息给Leader，而Follower已经开始不断地给Leader发送FETCH请求了，但因为没有数据因此什么都不会发生。值得一提的是，Follower发送过来的FETCH请求因为无数据而暂时会被寄存到Leader端的purgatory中，待500ms ( replica.fetch.wait.max.ms 参数)超时后会强制完成。倘若在寄存期间生产者发来数据，则Kafka会自动唤醒该FETCH请求，让Leader继续处理。

（2） Follower发送FETCH请求在Leader处理完PRODUCE请求之后

• producer给该topic分区发送了一条消息
• 此时的状态如下图所示：

如上图所示，Leader接收到PRODUCE请求主要做两件事情： 

• 把消息写入Log，同时自动更新Leader自己的LEO
• 尝试更新Leader HW值。假设此时Follower尚未发送FETCH请求，Leader端保存的Remote LEO依然是0，因此Leader会比较它自己的LEO值和Remote LEO值，发现最小值是0，与当前HW值相同，故不会更新分区HW值（仍为0）

        PRODUCE请求处理完成后各值如下，Leader端的HW值依然是0，而LEO是1，Remote LEO也是0。

属性	阶段	旧值	新值	备注
Leader LEO	PRODUCE处理完成	0	1	写入了一条数据
Remote LEO	PRODUCE处理完成	0	0	还未Fetch
Leader HW	PRODUCE处理完成	0	0	min(LeaderLEO=1, RemoteLEO=0)=0
Follower LEO	PRODUCE处理完成	0	0	还未Fetch
Follower HW	PRODUCE处理完成	0	0	min(LeaderHW=0, FollowerLEO=0)=0

假设此时follower发送了FETCH请求，则状态变更如下：

本例中当follower发送FETCH请求时，Leader端的处理依次是： 

• 读取Log数据
• 更新remote LEO = 0（为什么是0？ 因为此时Follower还没有写入这条消息。Leader如何确认Follower还未写入呢？这是通过Follower发来的FETCH请求中的Fetch offset来确定的）
• 尝试更新分区HW：此时Leader LEO = 1，Remote LEO = 0，故分区HW值= min(Leader LEO,Follower Remote LEO) = 0
• 把数据和当前分区HW值（依然是0）发送给Follower副本

而Follower副本接收到FETCH Response后依次执行下列操作：

1. 写入本地Log，同时更新Follower自己管理的 LEO为1
2. 更新Follower HW：比较本地LEO和 FETCH Response 中的当前Leader HW值，取较小者，Follower HW = 0

        此时，第一轮FETCH RPC结束，我们会发现虽然Leader和Follower都已经在Log中保存了这条消息，但分区HW值尚未被更新，仍为0。

属性	阶段	旧值	新值	备注
Leader LEO	PRODUCE和Follower FETCH处理完成	0	1	写入了一条数据
Remote LEO	PRODUCE和Follower FETCH处理完成	0	0	第一次fetch中offset为0
Leader HW	PRODUCE和Follower FETCH处理完成	0	0	min(LeaderLEO=1, RemoteLEO=0)=0
Follower LEO	PRODUCE和Follower FETCH处理完成	0	1	同步了一条数据
Follower HW	PRODUCE和Follower FETCH处理完成	0	0	min(LeaderHW=0, FollowerLEO=1)=0

Follower第二轮FETCH

        分区HW是在第二轮FETCH RPC中被更新的，如下图所示：

Follower发来了第二轮FETCH请求，Leader端接收到后仍然会依次执行下列操作： 

1. 读取Log数据
2. 更新Remote LEO = 1（这次为什么是1了？ 因为这轮FETCH RPC携带的fetch offset是1，那么为什么这轮携带的就是1了呢，因为上一轮结束后Follower LEO被更新为1了）
3. 尝试更新分区HW：此时leader LEO = 1，Remote LEO = 1，故分区HW值= min(Leader LEO,Follower Remote LEO) = 1。
4. 把数据（实际上没有数据）和当前分区HW值（已更新为1）发送给Follower副本作为Response

同样地，Follower副本接收到FETCH response后依次执行下列操作：

• 写入本地Log，当然没东西可写，Follower LEO也不会变化，依然是1。
• 更新Follower HW：比较本地LEO和当前Leader LEO取小者。由于都是1，故更新follower HW = 1 。

属性	阶段	旧值	新值	备注
Leader LEO	第二次Follower FETCH处理完成	1	1	未写入新数据
Remote LEO	第二次Follower FETCH处理完成	0	1	第2次fetch中offset为1
Leader HW	第二次Follower FETCH处理完成	0	1	min(RemoteLEO,LeaderLEO)=1
Follower LEO	第二次Follower FETCH处理完成	1	1	未写入新数据
Follower HW	第二次Follower FETCH处理完成	0	1	第2次fetch resp中的LeaderHW和本地Follower LEO都是1

        此时消息已经成功地被复制到Leader和Follower的Log中且分区HW是1，表明消费者能够消费offset = 0的消息。

（3）FETCH请求保存在purgatory中，PRODUCE请求到来。

        当Leader无法立即满足FECTH返回要求的时候(比如没有数据)，那么该FETCH请求被暂存到Leader端的purgatory中（炼狱），待时机成熟尝试再次处理。Kafka不会无限期缓存，默认有个超时时间（500ms），一旦超时时间已过，则这个请求会被强制完成。当寄存期间还没超时，生产者发送PRODUCE请求从而使之满足了条件以致被唤醒。此时，Leader端处理流程如下：

1. Leader写Log（自动更新Leader LEO）
2. 尝试唤醒在purgatory中寄存的FETCH请求
3. 尝试更新分区HW

6、HW和LEO异常案例

        Kafka使用HW值来决定副本备份的进度，而HW值的更新通常需要额外一轮FETCH RPC才能完成。但这种设计是有问题的，可能引起的问题包括：

• 备份数据丢失
• 备份数据不一致

（1）数据丢失

        使用HW值来确定备份进度时其值的更新是在下一轮RPC中完成的。如果Follower副本在标记上方的的第一步与第二步之间发生崩溃，那么就有可能造成数据的丢失。

• 上图中有两个副本：A和B。开始状态是A是Leader。
• 假设生产者min.insync.replicas 为1，那么当生产者发送两条消息给A后，A写入Log，此时Kafka会通知生产者这两条消息写入成功。 

代	属性	阶段	旧值	新值	备注
1	Leader LEO	PRODUCE和Follower FETCH处理完成	0	1	写入了一条数据
1	Remote LEO	PRODUCE和Follower FETCH处理完成	0	0	第一次fetch中offset为0
1	Leader HW	PRODUCE和Follower FETCH处理完成	0	0	min(LeaderLEO=1,FollowerLEO=0)=0
1	Follower LEO	PRODUCE和Follower FETCH处理完成	0	1	同步了一条数据
1	Follower HW	PRODUCE和Follower FETCH处理完成	0	0	min(LeaderHW=0, FollowerLEO=1)=0
2	Leader LEO	第二次Follower FETCH处理完成	1	2	写入了第二条数据
2	Remote LEO	第二次Follower FETCH处理完成	0	1	第2次fetch中offset为1
2	Leader HW	第二次Follower FETCH处理完成	0	1	min(RemoteLEO=1,LeaderLEO=2)=1
2	Follower LEO	第二次Follower FETCH处理完成	1	2	写入了第二条数据
2	Follower HW	第二次Follower FETCH处理完成	0	1	min(LeaderHW=1,FollowerLEO=2)=1
3	Leader LEO	第三次Follower FETCH处理完成	2	2	未写入新数据
3	Remote LEO	第三次Follower FETCH处理完成	1	2	第3次fetch中offset为2
3	Leader HW	第三次Follower FETCH处理完成	1	2	min(RemoteLEO=2,LeaderLEO)=2
3	Follower LEO	第三次Follower FETCH处理完成	2	2	未写入新数据
3	Follower HW	第三次Follower FETCH处理完成	1	2	第3次fetch resp中的LeaderHW和本地FollowerLEO都是2

        但是在broker端，Leader和Follower的Log虽都写入了2条消息且分区HW已经被更新到2，但Follower HW尚未被更新还是1，也就是上面标记的第二步尚未执行，表中最后一条未执行。

        倘若此时副本B所在的broker宕机，那么重启后B会自动把LEO调整到之前的HW值1，故副本B会做日志截断(log truncation)，将offset = 1的那条消息从log中删除，并调整LEO = 1。此时follower副本底层log中就只有一条消息，即offset = 0的消息！

        B重启之后需要给A发FETCH请求，但若A所在broker机器在此时宕机，那么Kafka会令B成为新的Leader，而当A重启回来后也会执行日志截断，将HW调整回1。这样，offset=1的消息就从两个副本的log中被删除，也就是说这条已经被生产者认为发送成功的数据丢失。

        丢失数据的前提是min.insync.replicas=1 时，一旦消息被写入Leader端Log即被认为是committed 。延迟一轮FETCH RPC 更新HW值的设计使follower HW值是异步延迟更新，若在这个过程中Leader发生变更，那么成为新Leader的Follower的HW值就有可能是过期的，导致生产者本是成功提交的消息被删除。

（2）Leader和Follower数据离散

• 除了可能造成的数据丢失以外，该设计还会造成Leader的Log和Follower的Log数据不一致。
• 如Leader端记录序列：m1,m2,m3,m4,m5,…；Follower端序列可能是m1,m3,m4,m5,…。

看图：

        假设：A是Leader，A的Log写入了2条消息，但B的Log只写了1条消息。分区HW更新到2，但B的HW还是1，同时生产者min.insync.replicas 仍然为1。

        假设A和B所在Broker同时宕机，B先重启回来，因此B成为Leader，分区HW = 1。假设此时生产者发送了第3条消息(红色表示)给B，于是B的log中offset = 1的消息变成了红框表示的消息，同时分区HW更新到2（A还没有回来，就B一个副本，故可以直接更新HW而不用理会A）之后A重启回来，需要执行日志截断，但发现此时分区HW=2而A之前的HW值也是2，故不做任何调整。此后A和B将以这种状态继续正常工作。

        显然，这种场景下，A和B的Log中保存在offset = 1的消息是不同的记录，从而引发不一致的情形出现。

7、Leader Epoch使用

Kafka解决方案

造成上述两个问题的根本原因在于

• HW值被用于衡量副本备份的成功与否。
• 在出现失败重启时作为日志截断的依据。

        但HW值的更新是异步延迟的，特别是需要额外的FETCH请求处理流程才能更新，故这中间发生的任何崩溃都可能导致HW值的过期。

        Kafka从0.11引入了leader epoch 来取代HW值。Leader端使用内存保存Leader的epoch信息，即使出现上面的两个场景也能规避这些问题。

所谓Leader epoch实际上是一对值：<epoch, offset>：

• epoch表示Leader的版本号，从0开始，Leader变更过1次，epoch+1
• offset对应于该epoch版本的Leader写入第一条消息的offset。因此假设有两对值：

<0, 0>
<1, 120>

        则表示第一个Leader从位移0开始写入消息；共写了120条[0, 119]；而第二个Leader版本号是1，从位移120处开始写入消息。

• Leader broker中会保存这样的一个缓存，并定期地写入到一个checkpoint 文件中。
• 当Leader写Log时它会尝试更新整个缓存：如果这个Leader首次写消息，则会在缓存中增加一个条目；否则就不做更新。
• 每次副本变为Leader时会查询这部分缓存，获取出对应Leader版本的位移，则不会发生数据不一致和丢失的情况。

（1）规避数据丢失

        只需要知道每个副本都引入了新的状态来保存自己当leader时开始写入的第一条消息的offset以及leader版本。这样在恢复的时候完全使用这些信息而非HW来判断是否需要截断日志。 

（2）规避数据不一致

依靠Leader epoch的信息可以有效地规避数据不一致的问题。 

8、消息重复的场景及解决方案

消息重复和丢失是kafka中很常见的问题，主要发生在以下三个阶段：

• 生产者阶段
• broke阶段
• 消费者阶段

8.1、生产者阶段重复场景

（1）根本原因

• 生产发送的消息没有收到正确的broke响应，导致生产者重试。
• 生产者发出一条消息，broke落盘以后因为网络等种种原因发送端得到一个发送失败的响应或者网络中断，然后生产者收到一个可恢复的Exception重试消息导致消息重复。

（2）重试过程

说明： 

• new KafkaProducer()后创建一个后台线程KafkaThread扫描RecordAccumulator中是否有消息；
• 调用KafkaProducer.send()发送消息，实际上只是把消息保存到RecordAccumulator中；
• 后台线程KafkaThread扫描到RecordAccumulator中有消息后，将消息发送到kafka集群；
• 如果发送成功，那么返回成功；
• 如果发送失败，那么判断是否允许重试。如果不允许重试，那么返回失败的结果；如果允许重试，把消息再保存到RecordAccumulator中，等待后台线程KafkaThread扫描再次发送；

（3）可恢复异常说明

        异常是RetriableException类型或者TransactionManager允许重试；RetriableException类继承关系如下：

（4） 记录顺序问题  

        如果设置max.in.flight.requests.per.connection 大于1（默认5，单个连接上发送的未确认请求的最大数量，表示上一个发出的请求没有确认下一个请求又发出了）。大于1可能会改变记录的顺序，因为如果将两个batch发送到单个分区，第一个batch处理失败并重试，但是第二个batch处理成功，那么第二个batch处理中的记录可能先出现被消费。

        设置max.in.flight.requests.per.connection 为1，可能会影响吞吐量，可以解决单个生产者发送顺序问题。如果多个生产者，生产者1先发送一个请求，生产者2后发送请求，此时生产者1返回可恢复异常，重试一定次数成功了。虽然生产者1先发送消息，但生产者2发送的消息会被先消费。

8.2、生产者发送重复解决方案

（1）启动kafka的幂等性

        要启动kafka的幂等性，设置： enable.idempotence=true ，以及ack=all 以及retries > 1 。

（2）ack=0，不重试

        可能会丢消息，适用于吞吐量指标重要性高于数据丢失，例如：日志收集。

（3）生产者和broker阶段消息丢失场景

• ack=0，不重试：生产者发送消息完，不管结果了，如果发送失败也就丢失了。
• ack=1，leader crash：生产者发送消息完，只等待Leader写入成功就返回了，Leader分区丢失了，此时Follower没来及同步，消息丢失。
• unclean.leader.election.enable 配置true：允许选举ISR以外的副本作为leader,会导致数据丢失，默认为false。生产者发送异步消息，只等待Lead写入成功就返回，Leader分区丢失，此时ISR中没有Follower，Leader从OSR中选举，因为OSR中本来落后于Leader造成消息丢失。

（4）解决生产者和broker阶段消息丢失

禁用unclean选举，ack=all

• ack=all / -1,tries > 1,unclean.leader.election.enable : false
• 生产者发完消息，等待Follower同步完再返回，如果异常则重试。副本的数量可能影响吞吐量，不超过5个，一般三个。
• 不允许unclean Leader选举。

配置：min.insync.replicas > 1

        当生产者将acks 设置为all (或-1 )时， min.insync.replicas>1 。指定确认消息写成功需要的最小副本数量。达不到这个最小值，生产者将引发一个异常(要么是NotEnoughReplicas，要么是NotEnoughReplicasAfterAppend)。

        当一起使用时， min.insync.replicas 和ack 允许执行更大的持久性保证。一个典型的场景是创建一个复制因子为3的主题，设置min.insync复制到2个，用all 配置发送。将确保如果大多数副本没有收到写操作，则生产者将引发异常。

失败的offset单独记录

        生产者发送消息，会自动重试，遇到不可恢复异常会抛出，这时可以捕获异常记录到数据库或缓存，进行单独处理。

（5）消费者数据重复场景及解决方案

• 根本原因：数据消费完没有及时提交offset到broker。
• 场景：消息消费端在消费过程中挂掉没有及时提交offset到broke，另一个消费端启动拿之前记录的offset开始消费，由于offset的滞后性可能会导致新启动的客户端有少量重复消费。

（6）解决方案

• 取消自动提交：每次消费完或者程序退出时手动提交。这可能也没法保证一条重复。
• 下游做幂等：一般是让下游做幂等或者尽量每消费一条消息都记录offset，对于少数严格的场景可能需要把offset或唯一ID（例如订单ID）和下游状态更新放在同一个数据库里面做事务来保证精确的一次更新或者在下游数据表里面同时记录消费offset，然后更新下游数据的时候用消费位移做乐观锁拒绝旧位移的数据更新。