Kafka 副本滞后说明

Question

kafka 的文档说明如下：

仍在从领导者那里获取消息但没有获取消息的副本 赶上replica.lag.time.max.ms 中的最新消息将是 被认为不同步。

我不确定这到底是什么意思。

1. 每个replica.lag.time.max.ms 被认为同步时，副本需要落后0 消息
2. 或副本获取的最新消息不应早于 副本.lag.time.max.ms

这 2 个定义不是一回事，因为如果它意味着 #2，则副本可能总是落后 2 或 3 条消息，但只要它的漂移不超过 replica.lag.time 就仍然保持同步.max.ms.

但如果这意味着 #1 副本需要严格比数据到达更快地消耗。

Answer 1

这是第二个。如果没有数据早于未复制的领导者的滞后时间，则副本是同步的。如果您认为应该更新措辞，请打开 jira，因为这很容易更新:)

Answer 2

我认为它更接近 #1，但不完全是。我贴一些源代码来帮助你。源代码版本为1.0.2。

Partition.getOutOfSyncReplicas(leaderReplica: Replica, maxLagMs: Long) 导致副本不同步：

def getOutOfSyncReplicas(leaderReplica: Replica, maxLagMs: Long): Set[Replica] = {
/**
 * there are two cases that will be handled here -
 * 1. Stuck followers: If the leo of the replica hasn't been updated for maxLagMs ms,
 *                     the follower is stuck and should be removed from the ISR
 * 2. Slow followers: If the replica has not read up to the leo within the last maxLagMs ms,
 *                    then the follower is lagging and should be removed from the ISR
 * Both these cases are handled by checking the lastCaughtUpTimeMs which represents
 * the last time when the replica was fully caught up. If either of the above conditions
 * is violated, that replica is considered to be out of sync
 *
 **/
val candidateReplicas = inSyncReplicas - leaderReplica

val laggingReplicas = candidateReplicas.filter(r => (time.milliseconds - r.lastCaughtUpTimeMs) > maxLagMs)
if (laggingReplicas.nonEmpty)
  debug("Lagging replicas are %s".format(laggingReplicas.map(_.brokerId).mkString(",")))

laggingReplicas

}

Replica.lastCaughtUpTimeMs 由 Replica.updateLogReadResult(logReadResult: LogReadResult) 更新：

/**
* If the FetchRequest reads up to the log end offset of the leader when the current fetch request is received,
* set `lastCaughtUpTimeMs` to the time when the current fetch request was received.
*
* Else if the FetchRequest reads up to the log end offset of the leader when the previous fetch request was received,
* set `lastCaughtUpTimeMs` to the time when the previous fetch request was received.
*
* This is needed to enforce the semantics of ISR, i.e. a replica is in ISR if and only if it lags behind leader's LEO
* by at most `replicaLagTimeMaxMs`. These semantics allow a follower to be added to the ISR even if the offset of its
* fetch request is always smaller than the leader's LEO, which can happen if small produce requests are received at
* high frequency.
**/
def updateLogReadResult(logReadResult: LogReadResult) {
if (logReadResult.info.fetchOffsetMetadata.messageOffset >= logReadResult.leaderLogEndOffset)
  _lastCaughtUpTimeMs = math.max(_lastCaughtUpTimeMs, logReadResult.fetchTimeMs)
else if (logReadResult.info.fetchOffsetMetadata.messageOffset >= lastFetchLeaderLogEndOffset)
  _lastCaughtUpTimeMs = math.max(_lastCaughtUpTimeMs, lastFetchTimeMs)

logStartOffset = logReadResult.followerLogStartOffset
logEndOffset = logReadResult.info.fetchOffsetMetadata
lastFetchLeaderLogEndOffset = logReadResult.leaderLogEndOffset
lastFetchTimeMs = logReadResult.fetchTimeMs
}