Skip to content

KAFKA-9491; Increment high watermark after full log truncation #8037

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 1 commit into from
Feb 4, 2020
Merged

KAFKA-9491; Increment high watermark after full log truncation #8037

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
8 changes: 4 additions & 4 deletions core/src/main/scala/kafka/cluster/Partition.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -490,8 +490,9 @@ class Partition(val topicPartition: TopicPartition,

val leaderLog = localLogOrException
val leaderEpochStartOffset = leaderLog.logEndOffset
info(s"$topicPartition starts at Leader Epoch ${partitionState.leaderEpoch} from " +
s"offset $leaderEpochStartOffset. Previous Leader Epoch was: $leaderEpoch")
info(s"$topicPartition starts at leader epoch ${partitionState.leaderEpoch} from " +
s"offset $leaderEpochStartOffset with high watermark ${leaderLog.highWatermark}. " +
s"Previous leader epoch was $leaderEpoch.")

//We cache the leader epoch here, persisting it only if it's local (hence having a log dir)
leaderEpoch = partitionState.leaderEpoch
Expand All @@ -506,12 +507,11 @@ class Partition(val topicPartition: TopicPartition,
leaderLog.maybeAssignEpochStartOffset(leaderEpoch, leaderEpochStartOffset)

val isNewLeader = !isLeader
val curLeaderLogEndOffset = leaderLog.logEndOffset
val curTimeMs = time.milliseconds
// initialize lastCaughtUpTime of replicas as well as their lastFetchTimeMs and lastFetchLeaderLogEndOffset.
remoteReplicas.foreach { replica =>
val lastCaughtUpTimeMs = if (inSyncReplicaIds.contains(replica.brokerId)) curTimeMs else 0L
replica.resetLastCaughtUpTime(curLeaderLogEndOffset, curTimeMs, lastCaughtUpTimeMs)
replica.resetLastCaughtUpTime(leaderEpochStartOffset, curTimeMs, lastCaughtUpTimeMs)

@ijuma ijuma Feb 4, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

resetLastCaughtUpTime takes curLeaderLogEndOffset, but we are now passing leaderEpochStartOffset. Do we need to update the parameter name of that method?

@hachikuji hachikuji Feb 4, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

They are the same thing. I just got rid of a redundant variable.

@ijuma ijuma Feb 4, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, I see.

}

if (isNewLeader) {
Expand Down
33 changes: 23 additions & 10 deletions core/src/main/scala/kafka/log/Log.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -301,7 +301,7 @@ class Log(@volatile var dir: File,

leaderEpochCache.foreach(_.truncateFromEnd(nextOffsetMetadata.messageOffset))

logStartOffset = math.max(logStartOffset, segments.firstEntry.getValue.baseOffset)
updateLogStartOffset(math.max(logStartOffset, segments.firstEntry.getValue.baseOffset))

// The earliest leader epoch may not be flushed during a hard failure. Recover it here.
leaderEpochCache.foreach(_.truncateFromStart(logStartOffset))
Expand Down Expand Up @@ -721,14 +721,30 @@ class Log(@volatile var dir: File,
}
}

private def updateLogEndOffset(messageOffset: Long): Unit = {
nextOffsetMetadata = LogOffsetMetadata(messageOffset, activeSegment.baseOffset, activeSegment.size)
private def updateLogEndOffset(offset: Long): Unit = {
nextOffsetMetadata = LogOffsetMetadata(offset, activeSegment.baseOffset, activeSegment.size)

// Update the high watermark in case it has gotten ahead of the log end offset following a truncation
// or if a new segment has been rolled and the offset metadata needs to be updated.
if (highWatermark >= messageOffset) {
if (highWatermark >= offset) {
updateHighWatermarkMetadata(nextOffsetMetadata)
}

if (this.recoveryPoint > offset) {
this.recoveryPoint = offset
}
}

private def updateLogStartOffset(offset: Long): Unit = {
logStartOffset = offset

if (highWatermark < offset) {
updateHighWatermark(offset)
}

if (this.recoveryPoint < offset) {

@chia7712 chia7712 Feb 4, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a case that recoveryPoint gets smaller than logStartOffset? If yes, is it ok to just move the recoveryPoint without flush? If not, is it worthy to throw exception or log this weird case?

@hachikuji hachikuji Feb 4, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is a recovery point lower than the log start offset useful? All data below the log start offset is subject to deletion.

@chia7712 chia7712 Feb 4, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

U are right :)

this.recoveryPoint = offset
}
}

/**
Expand Down Expand Up @@ -1242,7 +1258,7 @@ class Log(@volatile var dir: File,
checkIfMemoryMappedBufferClosed()
if (newLogStartOffset > logStartOffset) {
info(s"Incrementing log start offset to $newLogStartOffset")
logStartOffset = newLogStartOffset
updateLogStartOffset(newLogStartOffset)
leaderEpochCache.foreach(_.truncateFromStart(logStartOffset))
producerStateManager.truncateHead(newLogStartOffset)
maybeIncrementFirstUnstableOffset()
Expand Down Expand Up @@ -2046,8 +2062,7 @@ class Log(@volatile var dir: File,
removeAndDeleteSegments(deletable, asyncDelete = true)
activeSegment.truncateTo(targetOffset)
updateLogEndOffset(targetOffset)
this.recoveryPoint = math.min(targetOffset, this.recoveryPoint)
this.logStartOffset = math.min(targetOffset, this.logStartOffset)
updateLogStartOffset(math.min(targetOffset, this.logStartOffset))
leaderEpochCache.foreach(_.truncateFromEnd(targetOffset))
loadProducerState(targetOffset, reloadFromCleanShutdown = false)
}
Expand Down Expand Up @@ -2081,9 +2096,7 @@ class Log(@volatile var dir: File,
producerStateManager.truncate()
producerStateManager.updateMapEndOffset(newOffset)
maybeIncrementFirstUnstableOffset()

this.recoveryPoint = math.min(newOffset, this.recoveryPoint)
this.logStartOffset = newOffset
updateLogStartOffset(newOffset)

@chia7712 chia7712 Feb 4, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Previous behavior is recoveryPoint = math.min(newOffset, recoveryPoint)) but this patch changes it to

if (this.recoveryPoint < offset) {
  this.recoveryPoint = offset
}

which is equal to recoveryPoint = math.max(newOffset, recoveryPoint)). Is it a bug?

@hachikuji hachikuji Feb 4, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have also updated updateLogEndOffset to set the recovery point. In truncateFully where we delete all segments and set the log start offset to be equal to the log end offset, this ensures recovery point is also set consistently.

@chia7712 chia7712 Feb 4, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the explanation!

}
}
}
Expand Down
20 changes: 16 additions & 4 deletions core/src/test/scala/unit/kafka/log/LogTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,12 +117,13 @@ class LogTest {
def testHighWatermarkMaintenance(): Unit = {
val logConfig = LogTest.createLogConfig(segmentBytes = 1024 * 1024)
val log = createLog(logDir, logConfig)
val leaderEpoch = 0

val records = TestUtils.records(List(
def records(offset: Long): MemoryRecords = TestUtils.records(List(
new SimpleRecord(mockTime.milliseconds, "a".getBytes, "value".getBytes),
new SimpleRecord(mockTime.milliseconds, "b".getBytes, "value".getBytes),
new SimpleRecord(mockTime.milliseconds, "c".getBytes, "value".getBytes)
))
), baseOffset = offset, partitionLeaderEpoch= leaderEpoch)

def assertHighWatermark(offset: Long): Unit = {
assertEquals(offset, log.highWatermark)
Expand All @@ -133,7 +134,7 @@ class LogTest {
assertHighWatermark(0L)

// High watermark not changed by append
log.appendAsLeader(records, leaderEpoch = 0)
log.appendAsLeader(records(0), leaderEpoch)
assertHighWatermark(0L)

// Update high watermark as leader
Expand All @@ -145,13 +146,24 @@ class LogTest {
assertHighWatermark(3L)

// Update high watermark as follower
log.appendAsLeader(records, leaderEpoch = 0)
log.appendAsFollower(records(3L))
log.updateHighWatermark(6L)
assertHighWatermark(6L)

// High watermark should be adjusted by truncation
log.truncateTo(3L)
assertHighWatermark(3L)

log.appendAsLeader(records(0L), leaderEpoch = 0)
assertHighWatermark(3L)
assertEquals(6L, log.logEndOffset)
assertEquals(0L, log.logStartOffset)

// Full truncation should also reset high watermark
log.truncateFullyAndStartAt(4L)
assertEquals(4L, log.logEndOffset)
assertEquals(4L, log.logStartOffset)
assertHighWatermark(4L)
}

private def assertNonEmptyFetch(log: Log, offset: Long, isolation: FetchIsolation): Unit = {
Expand Down