一条消息只有被ISR中所有Follower都从Leader复制过去才会被认为已提交。这样就避免了部分数据被写进了Leader，还没来得及被任何Follower复制就宕机了，而造成数据丢失。而对于Producer而言，它可以选择是否等待消息commit，这可以通过request.required.acks来设置。这种机制确保了只要ISR中有一个或者以上的follower，一条被commit的消息就不会丢失。
什么是ISR？参考这里：【大数据哔哔集20210123】别问，问就是Kafka高可靠
有一个很重要的问题是当Leader宕机了，怎样在Follower中选举出新的Leader，因为Follower可能落后很多或者直接crash了，所以必须确保选择最新的Follower作为新的Leader。一个基本的原则就是，如果Leader不在了，新的Leader必须拥有原来的Leader commit的所有消息。这就需要做一个折中，如果Leader在一个消息被commit前等待更多的Follower确认，那么在它挂掉之后就有更多的Follower可以成为新的Leader，但这也会造成吞吐率的下降。
一种非常常用的选举Leader的方式是“少数服从多数“，Kafka并不是采用这种方式。这种模式下，如果我们有2f 1个副本，那么在commit之前必须保证有f 1个replica复制完消息，同时为了保证能正确选举出新的Leader，失败的副本数不能超过f个。这种方式有个很大的优势，系统的延迟取决于最快的几台机器，也就是说比如副本数为3，那么延迟就取决于最快的那个Follower而不是最慢的那个。“少数服从多数”的方式也有一些劣势，为了保证Leader选举的正常进行，它所能容忍的失败的Follower数比较少，如果要容忍1个Follower挂掉，那么至少要3个以上的副本，如果要容忍2个Follower挂掉，必须要有5个以上的副本。也就是说，在生产环境下为了保证较高的容错率，必须要有大量的副本，而大量的副本又会在大数据量下导致性能的急剧下降。这种算法更多用在Zookeeper这种共享集群配置的系统中而很少在需要大量数据的系统中使用的原因。HDFS的HA功能也是基于“少数服从多数”的方式，但是其数据存储并不是采用这样的方式。
实际上，Leader选举的算法非常多，比如Zookeeper的Zab、Raft以及Viewstamped Replication。而Kafka所使用的Leader选举算法更像是微软的PacificA算法。
Kafka在Zookeeper中为每一个Partition动态的维护了一个ISR，这个ISR里的所有replica都跟上了Leader，只有ISR里的成员才能有被选为Leader的可能（unclean.leader.election.enable=false）。在这种模式下，对于f 1个副本，一个Kafka Topic能在保证不丢失已经commit消息的前提下容忍f个副本的失败，在大多数使用场景下，这种模式是十分有利的。事实上，为了容忍f个副本的失败，“少数服从多数”的方式和ISR在commit前需要等待的副本的数量是一样的，但是ISR需要的总的副本的个数几乎是“少数服从多数”的方式的一半。
上文提到，在ISR中至少有一个Follower时，Kafka可以确保已经commit的数据不丢失，但如果某一个Partition的所有replica都挂了，就无法保证数据不丢失了。这种情况下有两种可行的方案：
- 等待ISR中任意一个replica“活”过来，并且选它作为Leader
- 选择第一个“活”过来的replica（并不一定是在ISR中）作为Leader
这就需要在可用性和一致性当中作出一个简单的抉择。如果一定要等待ISR中的replica“活”过来，那不可用的时间就可能会相对较长。而且如果ISR中所有的replica都无法“活”过来了，或者数据丢失了，这个Partition将永远不可用。选择第一个“活”过来的replica作为Leader，而这个replica不是ISR中的replica，那即使它并不保障已经包含了所有已commit的消息，它也会成为Leader而作为Consumer的数据源。默认情况下，Kafka采用第二种策略，即unclean.leader.election.enable=true，也可以将此参数设置为false来启用第一种策略。
unclean.leader.election.enable这个参数对于leader的选举、系统的可用性以及数据的可靠性都有至关重要的影响。下面我们来分析下几种典型的场景。
如果上图所示，假设某个Partition中的副本数为3，replica-0, replica-1, replica-2分别存放在Broker0, Broker1和Broker2中。AR=(0,1,2)，ISR=(0,1)。设置request.required.acks=-1, min.insync.replicas=2，unclean.leader.election.enable=false。这里将Broker0中的副本也称之为Broker0起初Broker0为Leader，Broker1为Follower。
- 当ISR中的replica-0出现crash的情况时，Broker1选举为新的Leader[ISR=(1)]，因为受min.insync.replicas=2影响，write不能服务，但是read能继续正常服务。此种情况恢复方案：
- 尝试恢复(重启)replica-0，如果能起来，系统正常；2. 如果replica-0不能恢复，需要将min.insync.replicas设置为1，恢复write功能。
- 当ISR中的replica-0出现crash，紧接着replica-1也出现了crash, 此时[ISR=(1),leader=-1],不能对外提供服务，此种情况恢复方案：
- 尝试恢复replica-0和replica-1，如果都能起来，则系统恢复正常；
- 如果replica-0起来，而replica-1不能起来，这时候仍然不能选出Leader，因为当设置unclean.leader.election.enable=false时，leader只能从ISR中选举，当ISR中所有副本都失效之后，需要ISR中最后失效的那个副本能恢复之后才能选举Leader，即replica-0先失效，replica-1后失效，需要replica-1恢复后才能选举Leader。保守的方案建议设置unclean.leader.election.enable=true，但是这样会有丢失数据的情况发生，这样可以恢复read服务。同样需要将min.insync.replicas设置为1，恢复write功能；
- replica-1恢复，replica-0不能恢复，这个情况上面遇到过，read服务可用，需要将min.insync.replicas设置为1，恢复write功能；
- replica-0和replica-1都不能恢复，这种情况可以参考情形2.
- 当ISR中的replica-0，replica-1同时宕机，此时[ISR=(0,1)]，不能对外提供服务，此种情况恢复方案：尝试恢复replica-0和replica-1，当其中任意一个副本恢复正常时，对外可以提供read服务。直到2个副本恢复正常，write功能才能恢复，或者将将min.insync.replicas设置为1。
