junrao commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4136497787
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -501,25 +408,25 @@ 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.
+ * Returns the initial partition fetch state based on the current state and
the provided
+ * `initialFetchState`.
+ *
+ * If the current state already has the same leader epoch, it is preserved.
Otherwise, the
+ * partition starts fetching from `initOffset` immediately. Any divergence
is detected via the
Review Comment:
This needs to be adjusted a bit since we don't always start with the
fetching state.
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -565,12 +472,8 @@ abstract class AbstractFetcherThread(name: String,
val maybeTruncationComplete = fetchOffsets.get(topicPartition) match {
case Some(offsetTruncationState) =>
val lastFetchedEpoch = latestEpoch(topicPartition)
- val state = if (leader.isTruncationOnFetchSupported ||
offsetTruncationState.truncationCompleted)
- ReplicaState.FETCHING
- else
- ReplicaState.TRUNCATING
new PartitionFetchState(currentFetchState.topicId,
offsetTruncationState.offset, currentFetchState.lag,
- currentFetchState.currentLeaderEpoch, currentFetchState.delay,
state, lastFetchedEpoch)
+ currentFetchState.currentLeaderEpoch, currentFetchState.delay,
ReplicaState.FETCHING, lastFetchedEpoch)
Review Comment:
> updating their fetch offset and maybe marking them as truncation completed
if their offsetTruncationState indicates truncation completed
This comment above is no longer valid since truncationCompleted is no longer
present in OffsetTruncationState.
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -478,9 +477,12 @@ class AbstractFetcherThreadTest {
}
}
- val replicaState = PartitionState(leaderEpoch = 5)
+ // The replica has a batch with an epoch unknown to the leader, so the
first fetch diverges
+ val replicaLog = Seq(
+ mkBatch(baseOffset = 0, leaderEpoch = 2, new SimpleRecord("x".getBytes)))
Review Comment:
Could this be merged to the previous line?
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -140,6 +141,38 @@ abstract class AbstractFetcherThread(name: String,
}
}
+ /**
+ * Truncate the log of every partition in the truncating state to its fetch
offset, which is the
+ * high watermark. A partition only enters that state when the follower has
no leader epoch, i.e.
+ * its log only contains records in a message format older than v2.
Divergence cannot be detected
+ * from the fetch response in that case, so the records above the high
watermark, which may not
+ * have been committed, are truncated before fetching resumes.
+ */
+ private[server] def maybeTruncate(): Unit =
LockUtils.inLock[Exception](partitionMapLock, () => {
+ // Collect the partitions first since truncating one of them may fail and
remove it from `partitionStates`.
+ val truncatingPartitions = mutable.Set.empty[TopicPartition]
+ partitionStates.partitionStateMap.forEach { (tp, state) =>
+ if (state.isTruncating)
+ truncatingPartitions += tp
+ }
+
+ if (truncatingPartitions.nonEmpty) {
+ val fetchOffsets = mutable.HashMap.empty[TopicPartition,
OffsetTruncationState]
+
+ for (tp <- truncatingPartitions) {
+ val partitionState = partitionStates.stateValue(tp)
+ if (partitionState != null) {
+ val truncationState =
OffsetTruncationState(partitionState.fetchOffset)
+ info(s"Truncating partition $tp with $truncationState due to local
high watermark ${partitionState.fetchOffset}")
+ if (doTruncate(tp, truncationState))
+ fetchOffsets.put(tp, truncationState)
+ }
+ }
+
+ updateFetchOffsetAndMaybeMarkTruncationComplete(fetchOffsets)
Review Comment:
Could we give updateFetchOffsetAndMaybeMarkTruncationComplete() a better
name since it always sets the state to Fetch?
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -400,21 +397,20 @@ class AbstractFetcherThreadTest {
def testTruncateToHighWatermarkIfLeaderEpochInfoNotAvailable(): Unit = {
val highWatermark = 2L
val partition = new TopicPartition("topic", 0)
- val mockLeaderEndPoint = new MockLeaderEndPoint(version = version) {
- override def fetchEpochEndOffsets(partitions:
java.util.Map[TopicPartition,
OffsetForLeaderEpochRequestData.OffsetForLeaderPartition]):
java.util.Map[TopicPartition, EpochEndOffset] =
- throw new UnsupportedOperationException
- }
+
+ val mockLeaderEndPoint = new MockLeaderEndPoint(version = version)
val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndPoint)
val fetcher = new MockFetcherThread(mockLeaderEndPoint,
mockTierStateMachine) {
- override def truncate(topicPartition: TopicPartition, truncationState:
OffsetTruncationState): Unit = {
- assertEquals(highWatermark, truncationState.offset)
- assertTrue(truncationState.truncationCompleted)
- super.truncate(topicPartition, truncationState)
- }
-
- override def latestEpoch(topicPartition: TopicPartition):
Optional[Integer] = Optional.empty
+ override def truncate(topicPartition: TopicPartition, truncationState:
OffsetTruncationState): Unit = {
+ assertEquals(highWatermark, truncationState.offset)
+ super.truncate(topicPartition, truncationState)
}
+ override def latestEpoch(topicPartition: TopicPartition):
Optional[Integer] = Optional.empty
+ }
+
+ // The replica has records above the high watermark which may not have
been committed. It has no
+ // leader epoch, so the leader cannot report a diverging epoch and they
are truncated beforehand.
Review Comment:
The comment is a bit misleading. The records actually contain the leader
epoch. It's latestEpoch() that doesn't contain a leader epoch.
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -603,54 +500,39 @@ 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)
- warn(s"Based on replica's leader epoch, leader replied with an unknown
offset in $tp. " +
- s"The initial fetch offset
${partitionStates.stateValue(tp).fetchOffset} will be used for truncation.")
- OffsetTruncationState(partitionStates.stateValue(tp).fetchOffset,
truncationCompleted = true)
- } else if (leaderEpochOffset.leaderEpoch == UNDEFINED_EPOCH) {
- // either leader or follower or both use inter-broker protocol version <
IBP_2_0_IV0
- // (version 0 of OffsetForLeaderEpoch request/response)
- warn(s"Leader or replica is on protocol version where leader epoch is
not considered in the OffsetsForLeaderEpoch response. " +
- s"The leader's offset ${leaderEpochOffset.endOffset} will be used for
truncation in $tp.")
- OffsetTruncationState(min(leaderEpochOffset.endOffset,
logEndOffset(tp)), truncationCompleted = true)
- } else {
- val replicaEndOffset = logEndOffset(tp)
-
- // get (leader epoch, end offset) pair that corresponds to the largest
leader epoch
- // less than or equal to the requested epoch.
- val endOffsetForEpochOpt = endOffsetForEpoch(tp,
leaderEpochOffset.leaderEpoch)
- if (endOffsetForEpochOpt.isPresent) {
- val offsetAndEpoch = endOffsetForEpochOpt.get
- val followerEndOffset = offsetAndEpoch.offset
- val followerEpoch = offsetAndEpoch.epoch()
- 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
- 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.")
- OffsetTruncationState(intermediateOffsetToTruncateTo,
truncationCompleted = false)
- } else {
- val offsetToTruncateTo = min(followerEndOffset,
leaderEpochOffset.endOffset)
- OffsetTruncationState(min(offsetToTruncateTo, replicaEndOffset),
truncationCompleted = true)
- }
+ val replicaEndOffset = logEndOffset(tp)
Review Comment:
Since OffsetTruncationState has only 1 field, we can probably change
getOffsetTruncationState() to return a long and give it a better name
accordinly.
--
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]
