MaxLag.Replica metric is reported incorrectly
On the follower side, for the empty `LogAppendInfo` retrieved from the leader, fetcherLagStats set the wrong lag for fetcherLagStats due to `nextOffset` is zero.
@@ -272,10 +272,10 @@ abstract class AbstractFetcherThread(name: String,
partitionData)
logAppendInfoOpt.foreach { logAppendInfo =>
- val nextOffset = logAppendInfo.lastOffset + 1
+ val validBytes = logAppendInfo.validBytes
+ val nextOffset = if (validBytes > 0) logAppendInfo.lastOffset + 1 else currentFetchState.fetchOffset
fetcherLagStats.getAndMaybePut(topicPartition).lag = Math.max(0L, partitionData.highWatermark - nextOffset)
- val validBytes = logAppendInfo.validBytes
// ReplicaDirAlterThread may have removed topicPartition from the partitionStates after processing the partition data
if (validBytes > 0 && partitionStates.contains(topicPartition)) {
// Update partitionStates only if there is no exception during processPartitionData
