You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "m1a2st (via GitHub)" <gi...@apache.org> on 2026/08/31 15:07:31 UTC
[PR] KAFKA-18477 remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread (WIP) [kafka]
m1a2st opened a new pull request, #23318:
URL: https://github.com/apache/kafka/pull/23318
Since the 4.0 baseline `MetadataVersion` is 3.0, all brokers support truncation on fetch (`Fetch` v12+ with `LastFetchedEpoch` and `DivergingEpoch`). Fetcher threads can therefore skip the separate `OffsetsForLeaderEpochRequest` and detect divergence directly from fetch responses.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4110608450
##########
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:
Yes, those cases now surface as `OFFSET_OUT_OF_RANGE` and are handled by `fetchOffsetAndTruncate()` instead. I've removed both branches.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #23318:
URL: https://github.com/apache/kafka/pull/23318
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4155243763
##########
core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala:
##########
@@ -179,22 +179,18 @@ class ReplicaAlterLogDirsThread(name: String,
}
/**
- * Truncate the log for each partition based on current replica's returned epoch and offset.
+ * Truncate the future replica's log for each partition based on the diverging epoch and offset
Review Comment:
Thanks, Addressed it.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #23318:
URL: https://github.com/apache/kafka/pull/23318#issuecomment-5892038021
Thanks for @junrao review, addressed all comments
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4186361729
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -345,6 +346,64 @@ class ReplicaManagerTest {
}
}
+ @Test
+ def testMaybeAddLogDirFetchersWhenFutureLogIsAheadOfHighWatermark(): Unit = {
+ val dir1 = TestUtils.tempDir()
+ val dir2 = TestUtils.tempDir()
+ val props = TestUtils.createBrokerConfig(0)
+ props.put("log.dirs", dir1.getAbsolutePath + "," + dir2.getAbsolutePath)
+ val config = KafkaConfig.fromProps(props)
+ val logManager = TestUtils.createLogManager(config.logDirs.asScala.map(new File(_)), new LogConfig(new Properties()))
+ mockGetAliveBrokerFunctions(metadataCache, Seq(new Node(0, "host0", 0)))
+ when(metadataCache.metadataVersion()).thenReturn(MetadataVersion.MINIMUM_VERSION)
+ val rm = new ReplicaManager(
+ metrics = metrics,
+ config = config,
+ time = time,
+ scheduler = new MockScheduler(time),
+ logManager = logManager,
+ quotaManagers = quotaManager,
+ metadataCache = metadataCache,
+ logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
+ alterPartitionManager = alterPartitionManager)
+
+ try {
+ val delta = topicsCreateDelta(0, isStartIdLeader = true, partitions = List(0), topicName = topic, topicId = topicIds(topic))
+ val image = imageFromTopics(delta.apply())
+ rm.applyDelta(delta, image)
+ val partition = rm.getPartitionOrException(topicPartition)
+ val leaderEpoch = partition.getLeaderEpoch
+
+ // Append one record per batch, so that the future log can be at a batch boundary below the log
+ // end offset of the current replica. Fetches are batch aligned, so a future log in the middle
+ // of a batch is not a state the fetcher can be in.
+ appendRecords(rm, topicPartition, MemoryRecords.withRecords(Compression.NONE, new SimpleRecord("first message".getBytes())))
+ appendRecords(rm, topicPartition, MemoryRecords.withRecords(Compression.NONE, new SimpleRecord("second message".getBytes())))
+ logManager.maybeUpdatePreferredLogDir(topicPartition, dir2.getAbsolutePath)
+
+ partition.createLogIfNotExists(isNew = true, isFutureReplica = true,
+ new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints.asJava), None)
+
+ // Simulate a future log which has already replicated a record that is not committed on the
+ // current replica yet, so that its log end offset is ahead of its high watermark.
+ partition.appendRecordsToFollowerOrFutureReplica(
+ MemoryRecords.withRecords(0L, Compression.NONE, leaderEpoch, new SimpleRecord("first message".getBytes())),
+ isFuture = true, leaderEpoch)
+ val futureLog = partition.futureLocalLogOrException
+ assertTrue(futureLog.logEndOffset > futureLog.highWatermark)
+ assertTrue(futureLog.latestEpoch.isPresent)
+
+ // The fetcher has to start from the log end offset of the future log. Starting from its high
+ // watermark makes the first fetch fail with an offset mismatch in
+ // ReplicaAlterLogDirsThread.processPartitionData, which marks the partition as failed.
+ rm.maybeAddLogDirFetchers(Set(partition), new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints.asJava), _ => None)
+ rm.replicaAlterLogDirsManager.fetcherThreadMap.values.foreach(t => t.doWork())
Review Comment:
Should we verify the initial fetch offset for the future replica before calling doWork()?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4125135876
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -110,7 +108,6 @@ abstract class AbstractFetcherThread(name: String,
}
override def doWork(): Unit = {
- maybeTruncate()
Review Comment:
Could we just fold maybeFetch() into doWork()?
##########
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:
I overlooked one case. It's possible that there are existing pre-V2 records in the follower. In that case, when the follower starts up, it could be in a state that lastFetchedEpoch is not present and HWM < LEO. In this case, the follower replica will initialize the fetch offset with HWM and we still want to do the truncation to HWM. So, we probably can't get rid of the truncating state. But we can simply maybeTruncate() to just truncating to the fetch offset.
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -603,54 +464,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)
+
+ // 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
Review Comment:
This is an existing issue, but followerEndOffset is better named as followerEpochEndOffset. Otherwise, it's hard to differentiating between followerEndOffset and replicaEndOffset.
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -197,91 +158,22 @@ abstract class AbstractFetcherThread(name: String,
}
}
- /**
- * - Build a leader epoch fetch based on partitions that are in the Truncating phase
- * - Send OffsetsForLeaderEpochRequest, retrieving the latest offset for each partition's
- * leader epoch. This is the offset the follower should truncate to ensure
- * accurate log replication.
- * - Finally truncate the logs for partitions in the truncating phase and mark the
- * truncation complete. Do this within a lock to ensure no leadership changes can
- * occur during truncation.
- */
- private def truncateToEpochEndOffsets(latestEpochsForPartitions: Map[TopicPartition, EpochData]): Unit = {
- val endOffsets = leader.fetchEpochEndOffsets(latestEpochsForPartitions.asJava)
- // Ensure we hold a lock during truncation
-
- LockUtils.inLock[Exception](partitionMapLock, () => {
- //Check no leadership and no leader epoch changes happened whilst we were unlocked, fetching epochs
-
- val epochEndOffsets = endOffsets.asScala.filter { case (tp, _) =>
- val curPartitionState = partitionStates.stateValue(tp)
- val partitionEpochRequest = latestEpochsForPartitions.getOrElse(tp, {
- throw new IllegalStateException(
- s"Leader replied with partition $tp not requested in OffsetsForLeaderEpoch request")
- })
- val leaderEpochInRequest = partitionEpochRequest.currentLeaderEpoch
- curPartitionState != null && leaderEpochInRequest == curPartitionState.currentLeaderEpoch
- }
-
- val result = maybeTruncateToEpochEndOffsets(epochEndOffsets, latestEpochsForPartitions)
- handlePartitionsWithErrors(result.partitionsWithError.asScala, "truncateToEpochEndOffsets")
- updateFetchOffsetAndMaybeMarkTruncationComplete(result.result)
- })
- }
-
// Visibility for unit tests
protected[server] def truncateOnFetchResponse(epochEndOffsets: Map[TopicPartition, EpochEndOffset]): Unit = {
LockUtils.inLock[Exception](partitionMapLock, () => {
- val result = maybeTruncateToEpochEndOffsets(epochEndOffsets, Map.empty)
- handlePartitionsWithErrors(result.partitionsWithError.asScala, "truncateOnFetchResponse")
- updateFetchOffsetAndMaybeMarkTruncationComplete(result.result)
+ updateFetchOffsetAndMaybeMarkTruncationComplete(maybeTruncateToEpochEndOffsets(epochEndOffsets))
})
}
- // Visible for testing
- private[server] def truncateToHighWatermark(partitions: Set[TopicPartition]): Unit =
- LockUtils.inLock[Exception](partitionMapLock, () => {
+ private def maybeTruncateToEpochEndOffsets(fetchedEpochs: Map[TopicPartition, EpochEndOffset]): Map[TopicPartition, OffsetTruncationState] = {
val fetchOffsets = mutable.HashMap.empty[TopicPartition, OffsetTruncationState]
- for (tp <- partitions) {
- val partitionState = partitionStates.stateValue(tp)
- if (partitionState != null) {
- val highWatermark = partitionState.fetchOffset
- val truncationState = OffsetTruncationState(highWatermark, truncationCompleted = true)
Review Comment:
truncationCompleted is never read now. Could we remove it?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #23318:
URL: https://github.com/apache/kafka/pull/23318#issuecomment-5912968875
Thanks for the review, @junrao. PTAL
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4162879220
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -501,25 +408,29 @@ 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 from `initOffset`.
+ *
+ * When `latestEpoch` is present, the partition starts in the Fetching state. Any log divergence is
+ * reported by the leader as a diverging epoch in the fetch response and handled by
+ * `truncateOnFetchResponse`.
+ *
+ * If `latestEpoch` is empty (for example, when the log contains only records in a message format
+ * older than v2), the fetch request carries no last fetched epoch and the leader cannot report a
+ * diverging epoch. In this case, the partition starts in the Truncating state, so that
+ * `maybeTruncate()` truncates its log to `initOffset`, which is the high watermark.
*/
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) {
- // With old message format, `latestEpoch` will be empty and we use Truncating state
- // to truncate to high watermark.
+ } else {
val lastFetchedEpoch = latestEpoch(tp)
val state = if (lastFetchedEpoch.isPresent) ReplicaState.FETCHING else ReplicaState.TRUNCATING
- new PartitionFetchState(initialFetchState.topicId.toJava, initialFetchState.initOffset, Optional.empty(), initialFetchState.currentLeaderEpoch,
- state, lastFetchedEpoch)
- } else {
Review Comment:
It seems to be a bug for a future log with LEO > HWM, which is common when the future log has caught up but the current replica still has uncommitted data. The first fetch will cause the exception shown below.
```java
if (fetchOffset != futureLog.logEndOffset)
throw new IllegalStateException("Offset mismatch for the future replica %s: fetched offset = %d, log end offset = %d.".format(
topicPartition, fetchOffset, futureLog.logEndOffset))
```
It was handled by the truncating phase (`getOffsetTruncationState` overrides the initial offset), but that phase is now removed by this PR. I think we need to update `ReplicaManager` to let `InitialFetchState` use `initialFetchOffset(futureLog)` instead of HWM, same as the follower path.
```java
futureReplicasAndInitialOffset.put(topicPartition, InitialFetchState(topicIds(topicPartition.topic), leader,
partition.getLeaderEpoch, futureLog.highWatermark))
val initialFetchState = InitialFetchState(topicId.toScala, new BrokerEndPoint(config.brokerId, "localhost", -1),
partition.getLeaderEpoch, futureLog.highWatermark)
```
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -144,144 +142,71 @@ 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 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.
Review Comment:
The empty log also enters this state, right?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4166908724
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -501,25 +408,29 @@ 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 from `initOffset`.
+ *
+ * When `latestEpoch` is present, the partition starts in the Fetching state. Any log divergence is
+ * reported by the leader as a diverging epoch in the fetch response and handled by
+ * `truncateOnFetchResponse`.
+ *
+ * If `latestEpoch` is empty (for example, when the log contains only records in a message format
+ * older than v2), the fetch request carries no last fetched epoch and the leader cannot report a
+ * diverging epoch. In this case, the partition starts in the Truncating state, so that
+ * `maybeTruncate()` truncates its log to `initOffset`, which is the high watermark.
*/
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) {
- // With old message format, `latestEpoch` will be empty and we use Truncating state
- // to truncate to high watermark.
+ } else {
val lastFetchedEpoch = latestEpoch(tp)
val state = if (lastFetchedEpoch.isPresent) ReplicaState.FETCHING else ReplicaState.TRUNCATING
- new PartitionFetchState(initialFetchState.topicId.toJava, initialFetchState.initOffset, Optional.empty(), initialFetchState.currentLeaderEpoch,
- state, lastFetchedEpoch)
- } else {
Review Comment:
Good point, I have addressed it!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4191523403
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -345,6 +346,64 @@ class ReplicaManagerTest {
}
}
+ @Test
+ def testMaybeAddLogDirFetchersWhenFutureLogIsAheadOfHighWatermark(): Unit = {
+ val dir1 = TestUtils.tempDir()
+ val dir2 = TestUtils.tempDir()
+ val props = TestUtils.createBrokerConfig(0)
+ props.put("log.dirs", dir1.getAbsolutePath + "," + dir2.getAbsolutePath)
+ val config = KafkaConfig.fromProps(props)
+ val logManager = TestUtils.createLogManager(config.logDirs.asScala.map(new File(_)), new LogConfig(new Properties()))
+ mockGetAliveBrokerFunctions(metadataCache, Seq(new Node(0, "host0", 0)))
+ when(metadataCache.metadataVersion()).thenReturn(MetadataVersion.MINIMUM_VERSION)
+ val rm = new ReplicaManager(
+ metrics = metrics,
+ config = config,
+ time = time,
+ scheduler = new MockScheduler(time),
+ logManager = logManager,
+ quotaManagers = quotaManager,
+ metadataCache = metadataCache,
+ logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
+ alterPartitionManager = alterPartitionManager)
+
+ try {
+ val delta = topicsCreateDelta(0, isStartIdLeader = true, partitions = List(0), topicName = topic, topicId = topicIds(topic))
+ val image = imageFromTopics(delta.apply())
+ rm.applyDelta(delta, image)
+ val partition = rm.getPartitionOrException(topicPartition)
+ val leaderEpoch = partition.getLeaderEpoch
+
+ // Append one record per batch, so that the future log can be at a batch boundary below the log
+ // end offset of the current replica. Fetches are batch aligned, so a future log in the middle
+ // of a batch is not a state the fetcher can be in.
+ appendRecords(rm, topicPartition, MemoryRecords.withRecords(Compression.NONE, new SimpleRecord("first message".getBytes())))
+ appendRecords(rm, topicPartition, MemoryRecords.withRecords(Compression.NONE, new SimpleRecord("second message".getBytes())))
+ logManager.maybeUpdatePreferredLogDir(topicPartition, dir2.getAbsolutePath)
+
+ partition.createLogIfNotExists(isNew = true, isFutureReplica = true,
+ new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints.asJava), None)
+
+ // Simulate a future log which has already replicated a record that is not committed on the
+ // current replica yet, so that its log end offset is ahead of its high watermark.
+ partition.appendRecordsToFollowerOrFutureReplica(
+ MemoryRecords.withRecords(0L, Compression.NONE, leaderEpoch, new SimpleRecord("first message".getBytes())),
+ isFuture = true, leaderEpoch)
+ val futureLog = partition.futureLocalLogOrException
+ assertTrue(futureLog.logEndOffset > futureLog.highWatermark)
+ assertTrue(futureLog.latestEpoch.isPresent)
+
+ // The fetcher has to start from the log end offset of the future log. Starting from its high
+ // watermark makes the first fetch fail with an offset mismatch in
+ // ReplicaAlterLogDirsThread.processPartitionData, which marks the partition as failed.
+ rm.maybeAddLogDirFetchers(Set(partition), new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints.asJava), _ => None)
+ rm.replicaAlterLogDirsManager.fetcherThreadMap.values.foreach(t => t.doWork())
Review Comment:
I have updated the test
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #23318:
URL: https://github.com/apache/kafka/pull/23318#issuecomment-5844369466
Thanks for the review, @junrao. I've addressed all your comments. PTAL.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4134457393
##########
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:
Addressed it.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #23318:
URL: https://github.com/apache/kafka/pull/23318#issuecomment-5578608007
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4110596200
##########
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:
Thanks for the explanation, that makes sense. Since `initOffset` is already the HWM when there is no leader epoch, truncation becomes a no-op.
With this change, nothing transitions a replica into `ReplicaState.TRUNCATING` anymore. If that sounds reasonable to you, I'll file a follow-up Jira to remove `ReplicaState.TRUNCATING` and `PartitionFetchState.isTruncating()`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4148985416
##########
core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala:
##########
@@ -179,22 +179,18 @@ class ReplicaAlterLogDirsThread(name: String,
}
/**
- * Truncate the log for each partition based on current replica's returned epoch and offset.
+ * Truncate the future replica's log for each partition based on the diverging epoch and offset
Review Comment:
This comment is not very accurate since truncate() can be called when truncating to the initial fetch offset too, not just from diverging epoch.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org