Skip to content

KAFKA-7104: Consistent leader's state in fetch response #5305

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ijuma merged 2 commits into from
Jun 28, 2018
+10 −11
Merged

KAFKA-7104: Consistent leader's state in fetch response #5305

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bc23c6f
KAFKA-7104: More consistent leader's state in fetch response
apovzner Jun 28, 2018
5bf8b58
fixed comment
apovzner Jun 28, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 7 additions & 10 deletions core/src/main/scala/kafka/server/ReplicaManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,11 +90,6 @@ case class LogReadResult(info: FetchDataInfo,
case Some(e) => Errors.forException(e)
}

def updateLeaderReplicaInfo(leaderReplica: Replica): LogReadResult =
copy(highWatermark = leaderReplica.highWatermark.messageOffset,
leaderLogStartOffset = leaderReplica.logStartOffset,
leaderLogEndOffset = leaderReplica.logEndOffset.messageOffset)

def withEmptyFetchInfo: LogReadResult =
copy(info = FetchDataInfo(LogOffsetMetadata.UnknownOffsetMetadata, MemoryRecords.EMPTY))

Expand Down Expand Up @@ -1340,7 +1335,12 @@ class ReplicaManager(val config: KafkaConfig,

/**
* Update the follower's fetch state in the leader based on the last fetch request and update `readResult`,
* if necessary.
* if the follower replica is not recognized to be one of the assigned replicas. Do not update
* `readResult` otherwise, so that log start/end offset and high watermark is consistent with
* records in fetch response. Log start/end offset and high watermark may change not only due to
* this fetch request, e.g., rolling new log segment and removing old log segment may move log
* start offset further than the last offset in the fetched records. The followers will get the
* updated leader's state in the next fetch response.
*/
private def updateFollowerLogReadResults(replicaId: Int,
readResults: Seq[(TopicPartition, LogReadResult)]): Seq[(TopicPartition, LogReadResult)] = {
Expand All @@ -1351,10 +1351,7 @@ class ReplicaManager(val config: KafkaConfig,
case Some(partition) =>
partition.getReplica(replicaId) match {
case Some(replica) =>
if (partition.updateReplicaLogReadResult(replica, readResult))
partition.leaderReplicaIfLocal.foreach { leaderReplica =>
updatedReadResult = readResult.updateLeaderReplicaInfo(leaderReplica)
}
partition.updateReplicaLogReadResult(replica, readResult)
case None =>
warn(s"Leader $localBrokerId failed to record follower $replicaId's position " +
s"${readResult.info.fetchOffsetMetadata.messageOffset} since the replica is not recognized to be " +
Expand Down
4 changes: 3 additions & 1 deletion core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -463,7 +463,9 @@ class ReplicaManagerTest {

val tp0Status = responseStatusMap.get(tp0)
assertTrue(tp0Status.isDefined)
assertEquals(1, tp0Status.get.highWatermark)
// the response contains high watermark on the leader before it is updated based
// on this fetch request
assertEquals(0, tp0Status.get.highWatermark)
assertEquals(None, tp0Status.get.lastStableOffset)
assertEquals(Errors.NONE, tp0Status.get.error)
assertTrue(tp0Status.get.records.batches.iterator.hasNext)
Expand Down