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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
