KAFKA-7415; Persist leader epoch and start offset on becoming a leader #5678
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
junrao
reviewed
Oct 1, 2018
There was a problem hiding this comment.
@hachikuji : Thanks for the patch. Looks good overall. Just a few minor comments below.
| } | ||
| epochs += entryToAppend | ||
|
|
||
| debug(s"Appended new epoch entry $entryToAppend. Cache now contains ${epochs.size} entries.") |
There was a problem hiding this comment.
Since this is not expected, should we log a WARN or error instead of debug?
| (requestedEpoch, leo().messageOffset) | ||
| // The latest epoch is always the current epoch that is still being written to. Followers | ||
| // should not have any reason to query the latest epoch since truncation always takes place | ||
| // after the epoch is incremented. |
There was a problem hiding this comment.
Is this comment accurate? When a follower is restarted and the leader isn't, the follower actually will ask for the end offset for the latest epoch.
There was a problem hiding this comment.
Let me try to clarify. The point of this patch is to ensure that the cache always contains the latest epoch following a leader election. Let's say there is a leader election and the new epoch is 5. The cache will reflect epoch 5 with a starting offset equal to the log end offset at the time the replica became a leader. Followers would not have any need to send OffsetsForLeaderEpoch queries for epoch 5, but the first epoch prior to 5.
| if (subsequentEpochs.isEmpty || requestedEpoch < epochs.head.epoch) | ||
| // no epochs recorded or requested epoch < the first epoch cached | ||
| if (subsequentEpochs.isEmpty) { | ||
| // The epoch is larger than any known epoch. This case should never be hit because the |
There was a problem hiding this comment.
The epoch => The requested epoch? Ditto in a few other places.
| inReadLock(lock) { | ||
| val epochAndOffset = | ||
| if (requestedEpoch == UNDEFINED_EPOCH) { | ||
| // this may happen if a bootstrapping follower sends a request with undefined epoch or | ||
| // This may happen if a bootstrapping follower sends a request with undefined epoch or |
There was a problem hiding this comment.
With this change, it seems that epochs should never be empty when used?
There was a problem hiding this comment.
Yeah, I kept this for upgrade compatibility.
core/src/test/scala/unit/kafka/server/epoch/LeaderEpochFileCacheTest.scala
Show resolved
Hide resolved
core/src/test/scala/unit/kafka/server/epoch/LeaderEpochFileCacheTest.scala
Show resolved
Hide resolved
| if (epoch >= 0 && epoch > latestEpoch && offset >= latestOffset) { | ||
| info(s"Updated PartitionLeaderEpoch. ${epochChangeMsg(epoch, offset)}. Cache now contains ${epochs.size} entries.") | ||
| epochs += EpochEntry(epoch, offset) | ||
| if (epoch != latestEpoch) { |
There was a problem hiding this comment.
In the case when epoch == latestEpoch, if the startOffset is lower than what's in the cache, should we update the epoch cache with the lower offset for consistency?
core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala
Show resolved
Hide resolved
| baseOffset = startOffset, partitionLeaderEpoch = epoch) | ||
| } | ||
|
|
||
| val logConfig = LogTest.createLogConfig(segmentBytes = 10 * createRecords(0, 0).sizeInBytes) |
There was a problem hiding this comment.
Not sure if this truly matters, but does createRecords(0, 0).sizeInBytes match the record size when records are created with different offsets?
There was a problem hiding this comment.
Hmm.. I think they should be the same. The batch base offset is a fixed size and the first record will always have a relative offset of 0.
junrao
approved these changes
Oct 4, 2018
hachikuji
added a commit
that referenced
this pull request
Oct 4, 2018
#5678) This patch ensures that the leader epoch cache is updated when a broker becomes leader with the latest epoch and the log end offset as its starting offset. This guarantees that the leader will be able to provide the right truncation point even if the follower has data from leader epochs which the leader itself does not have. This situation can occur when there are back to back leader elections. Additionally, we have made the following changes: 1. The leader epoch cache enforces monotonically increase epochs and starting offsets among its entry. Whenever a new entry is appended which violates requirement, we remove the conflicting entries from the cache. 2. Previously we returned an unknown epoch and offset if an epoch is queried which comes before the first entry in the cache. Now we return the smallest . For example, if the earliest entry in the cache is (epoch=5, startOffset=10), then a query for epoch 4 will return (epoch=4, endOffset=10). This ensures that followers (and consumers in KIP-320) can always determine where the correct starting point is for the active log range on the leader. Reviewers: Jun Rao <[email protected]>
hachikuji
added a commit
to hachikuji/kafka
that referenced
this pull request
Oct 5, 2018
apache#5678) This patch ensures that the leader epoch cache is updated when a broker becomes leader with the latest epoch and the log end offset as its starting offset. This guarantees that the leader will be able to provide the right truncation point even if the follower has data from leader epochs which the leader itself does not have. This situation can occur when there are back to back leader elections. Additionally, we have made the following changes: 1. The leader epoch cache enforces monotonically increase epochs and starting offsets among its entry. Whenever a new entry is appended which violates requirement, we remove the conflicting entries from the cache. 2. Previously we returned an unknown epoch and offset if an epoch is queried which comes before the first entry in the cache. Now we return the smallest . For example, if the earliest entry in the cache is (epoch=5, startOffset=10), then a query for epoch 4 will return (epoch=4, endOffset=10). This ensures that followers (and consumers in KIP-320) can always determine where the correct starting point is for the active log range on the leader. Reviewers: Jun Rao <[email protected]>
3 tasks
hachikuji
added a commit
that referenced
this pull request
Oct 5, 2018
#5678) (#5749) This patch ensures that the leader epoch cache is updated when a broker becomes leader with the latest epoch and the log end offset as its starting offset. This guarantees that the leader will be able to provide the right truncation point even if the follower has data from leader epochs which the leader itself does not have. This situation can occur when there are back to back leader elections. Additionally, we have made the following changes: 1. The leader epoch cache enforces monotonically increase epochs and starting offsets among its entry. Whenever a new entry is appended which violates requirement, we remove the conflicting entries from the cache. 2. Previously we returned an unknown epoch and offset if an epoch is queried which comes before the first entry in the cache. Now we return the smallest . For example, if the earliest entry in the cache is (epoch=5, startOffset=10), then a query for epoch 4 will return (epoch=4, endOffset=10). This ensures that followers (and consumers in KIP-320) can always determine where the correct starting point is for the active log range on the leader. Reviewers: Jun Rao <[email protected]>
pengxiaolong
pushed a commit
to pengxiaolong/kafka
that referenced
this pull request
Jun 14, 2019
apache#5678) This patch ensures that the leader epoch cache is updated when a broker becomes leader with the latest epoch and the log end offset as its starting offset. This guarantees that the leader will be able to provide the right truncation point even if the follower has data from leader epochs which the leader itself does not have. This situation can occur when there are back to back leader elections. Additionally, we have made the following changes: 1. The leader epoch cache enforces monotonically increase epochs and starting offsets among its entry. Whenever a new entry is appended which violates requirement, we remove the conflicting entries from the cache. 2. Previously we returned an unknown epoch and offset if an epoch is queried which comes before the first entry in the cache. Now we return the smallest . For example, if the earliest entry in the cache is (epoch=5, startOffset=10), then a query for epoch 4 will return (epoch=4, endOffset=10). This ensures that followers (and consumers in KIP-320) can always determine where the correct starting point is for the active log range on the leader. Reviewers: Jun Rao <[email protected]>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This patch ensures that the leader epoch cache is updated when a broker becomes leader with the latest epoch and the log end offset as its starting offset. This guarantees that the leader will be able to provide the right truncation point even if the follower has data from leader epochs which the leader itself does not have. This situation can occur when there are back to back leader elections.
Additionally, we have made the following changes:
Committer Checklist (excluded from commit message)