kafka ack = all and min-isr

Question

摘要

Kafka的文档和代码注释表明,当生产者设置acks设置all为时,只有当所有同步副本都赶上时,ack才会被发送给生产者,但代码(Partition.Scala,checkEnoughReplicasReachOffset)似乎表明ack已发送一旦最小的同步复制品赶上了.

细节

kafka文档有这个:

acks = all这意味着领导者将等待完整的同步副本集以确认记录. 资源

另外,查看Kafka源代码 - partition.scala checkEnoughReplicasReachOffset()有以下评论(强调我的):

请注意,只有在requiredAcks = -1时才会调用此方法,并且在我们确认生成请求之前,我们正在等待ISR中的所有副本完全捕获到与此生成请求相对应的(本地)领导者的偏移量.

最后,关于Stack Overflow的这个答案(再次强调我的)

此同步最小同步副本设置指定了分区保持可用于写入时需要同步的最小副本数.当生产者指定ack(-1/all config)时,它仍将等待来自所有同步副本的 ack(与min in-sync副本的设置无关).

但是当我查看Partition.Scala中的代码时(注意minIsr < curInSyncReplicas.size):

def checkEnoughReplicasReachOffset(requiredOffset: Long): (Boolean, Errors) = {
  ...
  val minIsr = leaderReplica.log.get.config.minInSyncReplicas
  if (leaderReplica.highWatermark.messageOffset >= requiredOffset) {          
    if (minIsr <= curInSyncReplicas.size)
      (true, Errors.NONE)

调用它的代码返回ack:

if (error != Errors.NONE || hasEnough) {
  status.acksPending = false
  status.responseStatus.error = error
}

因此,只要同步副本集大于最小同步内副本,代码看起来就会返回确认.但是,文档和注释表明只有在所有同步副本都赶上后才会发送确认.我错过了什么？至少,上面的评论checkEnoughReplicasReachOffset看起来应该改变.

Answer 1

感谢 jira-dev 邮件列表上的 Ismael。

关键点是这一行：

if(leaderReplica.highWatermark.messageOffset >= requiredOffset) {

仅当 ISR 中的所有副本都具有该特定偏移量时，高水位线才会移动。