副本提取程序崩溃 - 第一个偏移量 XXXXXXX 小于下一个偏移量 YYYYYYYYY

Question

我们有一个 2.3 的 Kafka 集群，包含 3 个盒子。几天前，当我们将其升级到 2.3 时，我们注意到那些日志消息使两个代理上的一个主题分区的 replicaFetcher 线程崩溃：

[2019-08-09 15:02:43,520] ERROR [ReplicaFetcher replicaId=4, leaderId=3, fetcherId=0] Unexpected error occurred while processing data for partition __consumer_offsets-21 at offset 57542337 (kafka.server.R
eplicaFetcherThread)
kafka.common.UnexpectedAppendOffsetException: Unexpected offset in append to __consumer_offsets-21. First offset 57542333 is less than the next offset 57542337. First 10 offsets in append: List(57542333,
57542334, 57542335, 57542336, 57542337, 57542338, 57542339, 57542340, 57542341, 57542342), last offset in append: 57570869. Log start offset = 56949140
at kafka.log.Log.$anonfun$append$2(Log.scala:929)
at kafka.log.Log.maybeHandleIOException(Log.scala:2065)
at kafka.log.Log.append(Log.scala:850)
at kafka.log.Log.appendAsFollower(Log.scala:830)
at kafka.cluster.Partition.$anonfun$doAppendRecordsToFollowerOrFutureReplica$1(Partition.scala:726)
at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:253)
at kafka.utils.CoreUtils$.inReadLock(CoreUtils.scala:259)
at kafka.cluster.Partition.doAppendRecordsToFollowerOrFutureReplica(Partition.scala:717)
at kafka.cluster.Partition.appendRecordsToFollowerOrFutureReplica(Partition.scala:733)
at kafka.server.ReplicaFetcherThread.processPartitionData(ReplicaFetcherThread.scala:161)
at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$7(AbstractFetcherThread.scala:317)
at scala.Option.foreach(Option.scala:274)
at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$6(AbstractFetcherThread.scala:306)
at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$6$adapted(AbstractFetcherThread.scala:305)
at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$5(AbstractFetcherThread.scala:305)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:253)
at kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:305)
at kafka.server.AbstractFetcherThread.$anonfun$maybeFetch$3(AbstractFetcherThread.scala:133)
at kafka.server.AbstractFetcherThread.$anonfun$maybeFetch$3$adapted(AbstractFetcherThread.scala:132)
at scala.Option.foreach(Option.scala:274)
at kafka.server.AbstractFetcherThread.maybeFetch(AbstractFetcherThread.scala:132)
at kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:114)
at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:89)
[2019-08-09 15:02:43,524] WARN [ReplicaFetcher replicaId=4, leaderId=3, fetcherId=0] Partition __consumer_offsets-21 marked as failed (kafka.server.ReplicaFetcherThread)

影响是，对于这个主题分区，一个代理不能成为 ISR(实际上第二个代理有同样的问题，所以我们只有一个 ISR 是领导者(。

我仍然对此消息感到困惑，并且无法正确理解它，因此我无法找到解决此问题的正确方法。 我真的很想了解这里发生了什么，但不确定我是否理解以下代码：

https://github.com/apache/kafka/blob/a48b5d900c6b5c9c52a97124a1b51aff3636c32c/core/src/main/scala/kafka/log/Log.scala#L1081-L1098

if (appendInfo.firstOrLastOffsetOfFirstBatch < nextOffsetMetadata.messageOffset) {

当副本提取器必须附加记录时，它如何访问下一个偏移量信息..？ 不确定了解此分析的确切作用(要附加的当前记录？

val appendInfo = analyzeAndValidateRecords(records, isFromClient = isFromClient)

而这个：

nextOffsetMetadata

这是下一个记录批次吗？它如何访问任何"下一个"记录元数据？

如果有人能澄清一下，那就太好了。 与此同时，摆脱这个问题的解决方案会很好，但是，我仍然希望清楚地理解它。

编辑：

经过一番研究，有些事情变得更加清晰。 nextOffset 只是活动段的最新偏移量+1(这些元数据来自 loadSegments(( 调用(。

所以总结一下，这里发生了什么： 副本从起始偏移量低于活动段的最新偏移量的引线获取段。 所以这是我的问题，为什么副本不只是截断？

雅尼克

Answer 1

我们的一个 kafka 集群正在运行 kafka 1.1.1，并且遇到了同样的问题，使用类固醇。在我们的例子中，ReplicaFetcher崩溃了，并完全停止了复制。 解决方案是删除受感染的分区，并让 kafka 从健康的副本重新创建 if。我们试图修复分区文件夹上不一致的条目，但没有运气。

Answer 2

一种解决方案是停止有故障的代理并从磁盘中删除有故障的分区日志，然后重新启动代理。 然后，副本提取程序将停止抱怨，代理将能够进行 ISR。