junrao commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4067037745
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -502,24 +433,20 @@ abstract class AbstractFetcherThread(name: String,
/**
* Returns initial partition fetch state based on current state and the
provided `initialFetchState`.
- * From IBP 2.7 onwards, we can rely on truncation based on diverging data
returned in fetch responses.
- * For older versions, we can skip the truncation step iff the leader epoch
matches the existing epoch.
+ * We rely on truncation based on diverging data returned in fetch responses.
*/
private def partitionFetchState(tp: TopicPartition, initialFetchState:
InitialFetchState, currentState: PartitionFetchState): PartitionFetchState = {
if (currentState != null && currentState.currentLeaderEpoch ==
initialFetchState.currentLeaderEpoch) {
currentState
} else if (initialFetchState.initOffset < 0) {
fetchOffsetAndTruncate(tp, initialFetchState.topicId,
initialFetchState.currentLeaderEpoch)
- } else if (leader.isTruncationOnFetchSupported) {
+ } else {
Review Comment:
It seems that we can simplify partitionFetchState() to always set the state
to FETCHING. Currently, this method sets the state to TRUNCATING only when
lastFetchedEpoch.isPresent is false. In that case, when exercising
maybeTruncate(), it degenerates to truncating to the HWM. However, In
ReplicaManager.initialFetchOffset(), if log.latestEpoch.isPresent is false, the
initial fetch offset is set to HWM. So, truncating to HMW to a no-op.
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -502,24 +433,20 @@ abstract class AbstractFetcherThread(name: String,
/**
* Returns initial partition fetch state based on current state and the
provided `initialFetchState`.
- * From IBP 2.7 onwards, we can rely on truncation based on diverging data
returned in fetch responses.
- * For older versions, we can skip the truncation step iff the leader epoch
matches the existing epoch.
+ * We rely on truncation based on diverging data returned in fetch responses.
*/
private def partitionFetchState(tp: TopicPartition, initialFetchState:
InitialFetchState, currentState: PartitionFetchState): PartitionFetchState = {
if (currentState != null && currentState.currentLeaderEpoch ==
initialFetchState.currentLeaderEpoch) {
currentState
} else if (initialFetchState.initOffset < 0) {
Review Comment:
This is an existing issue. It seems that initialFetchState.initOffset is
always >=0. We can just remove this check.
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -502,24 +433,20 @@ abstract class AbstractFetcherThread(name: String,
/**
* Returns initial partition fetch state based on current state and the
provided `initialFetchState`.
- * From IBP 2.7 onwards, we can rely on truncation based on diverging data
returned in fetch responses.
- * For older versions, we can skip the truncation step iff the leader epoch
matches the existing epoch.
+ * We rely on truncation based on diverging data returned in fetch responses.
Review Comment:
diverging data => diverging epoch
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -604,10 +525,7 @@ abstract class AbstractFetcherThread(name: String,
private def getOffsetTruncationState(tp: TopicPartition, leaderEpochOffset:
EpochEndOffset): OffsetTruncationState =
LockUtils.inLock(partitionMapLock , () => {
if (leaderEpochOffset.endOffset == UNDEFINED_EPOCH_OFFSET) {
- // truncate to initial offset which is the high watermark for follower
replica. For
- // future replica, it is either high watermark of the future replica or
current
- // replica's truncation offset (when the current replica truncates, it
forces future
- // replica's partition state to 'truncating' and sets initial offset to
its truncation offset)
+ // truncate to initial offset which is the high watermark of the replica
Review Comment:
It seems that leaderEpochOffset.endOffset == UNDEFINED_EPOCH_OFFSET can
never be true in production?
Also, it seems that the branch leaderEpochOffset.leaderEpoch ==
UNDEFINED_EPOCH) can never be true and the comment "inter-broker protocol
version < IBP_2_0_IV0" is no longer valid.
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -273,9 +223,7 @@ abstract class AbstractFetcherThread(name: String,
fetchOffsets.put(tp, offsetTruncationState)
case Errors.FENCED_LEADER_EPOCH =>
- val currentLeaderEpoch = latestEpochsForPartitions.get(tp)
- .map(epochEndOffset =>
Int.box(epochEndOffset.currentLeaderEpoch)).toJava
- if (onPartitionFenced(tp, currentLeaderEpoch))
+ if (onPartitionFenced(tp, Optional.empty()))
Review Comment:
If requestEpoch is empty, onPartitionFenced() always returns true. So we can
skip this test.
Also, maybeTruncateToEpochEndOffsets() is only called when there is
diverging epoch in the response. In that case, the response error is None. So,
the Errors.FENCED_LEADER_EPOCH path seems unnecessary.
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -144,38 +143,22 @@ abstract class AbstractFetcherThread(name: String,
}
/**
- * Builds offset for leader epoch requests for partitions that are in the
truncating phase based
- * on latest epochs of the future replicas (the one that is fetching)
+ * Truncate partitions that are in the truncating phase to their local high
watermark. A partition
+ * only enters the truncating phase when the replica has no leader epoch
information (e.g. it only
+ * contains messages with the old format); replicas with leader epochs rely
on the diverging epoch
+ * returned in fetch responses for truncation.
*/
- private def fetchTruncatingPartitions(): (Map[TopicPartition, EpochData],
Set[TopicPartition]) =
- LockUtils.inLock(partitionMapLock, () => {
- val partitionsWithEpochs = mutable.Map.empty[TopicPartition, EpochData]
- val partitionsWithoutEpochs = mutable.Set.empty[TopicPartition]
-
- partitionStates.partitionStateMap.forEach { (tp, state) =>
- if (state.isTruncating) {
- latestEpoch(tp).toScala match {
- case Some(epoch) =>
- partitionsWithEpochs += tp -> new EpochData()
- .setPartition(tp.partition)
- .setCurrentLeaderEpoch(state.currentLeaderEpoch)
- .setLeaderEpoch(epoch)
- case _ =>
- partitionsWithoutEpochs += tp
- }
- }
- }
-
- (partitionsWithEpochs, partitionsWithoutEpochs)
- })
-
private def maybeTruncate(): Unit = {
- val (partitionsWithEpochs, partitionsWithoutEpochs) =
fetchTruncatingPartitions()
- if (partitionsWithEpochs.nonEmpty) {
- truncateToEpochEndOffsets(partitionsWithEpochs)
- }
- if (partitionsWithoutEpochs.nonEmpty) {
- truncateToHighWatermark(partitionsWithoutEpochs)
+ val truncatingPartitions = LockUtils.inLock(partitionMapLock, () => {
+ val partitions = mutable.Set.empty[TopicPartition]
+ partitionStates.partitionStateMap.forEach { (tp, state) =>
+ if (state.isTruncating)
+ partitions += tp
+ }
+ partitions
+ })
+ if (truncatingPartitions.nonEmpty) {
+ truncateToHighWatermark(truncatingPartitions)
Review Comment:
This is an existing issue, but truncateToHighWatermark() is probably better
named as truncateToFetchOffset().
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -630,11 +548,12 @@ abstract class AbstractFetcherThread(name: String,
if (followerEpoch != leaderEpochOffset.leaderEpoch) {
// the follower does not know about the epoch that leader replied
with
// we truncate to the end offset of the largest epoch that is
smaller than the
- // epoch the leader replied with, and send another offset for leader
epoch request
+ // epoch the leader replied with; the next fetch request carries
that epoch as
+ // the last fetched epoch and the leader may reply with another
diverging epoch
val intermediateOffsetToTruncateTo = min(followerEndOffset,
replicaEndOffset)
info(s"Based on replica's leader epoch, leader replied with epoch
${leaderEpochOffset.leaderEpoch} " +
s"unknown to the replica for $tp. " +
- s"Will truncate to $intermediateOffsetToTruncateTo and send
another leader epoch request to the leader.")
+ s"Will truncate to $intermediateOffsetToTruncateTo and continue
fetching with the last known epoch.")
Review Comment:
In the else branch in line 563, we probably want to adjust the following
comment. If the follower is not tracking the epoch, the leader won't reply with
diverging epoch and won't reach there.
> // This can happen if the follower was not tracking leader epochs
at that point (before the
> // upgrade, or if this broker is new).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
