rajinisivaram commented on a change in pull request #9382:
URL: https://github.com/apache/kafka/pull/9382#discussion_r518200860
##########
File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala
##########
@@ -454,15 +492,23 @@ abstract class AbstractFetcherThread(name: String,
* truncation completed if their offsetTruncationState indicates truncation
completed
*
* @param fetchOffsets the partitions to update fetch offset and maybe mark
truncation complete
+ * @param maySkipTruncation true if we can stay in Fetching mode and
perform truncation later based on
+ * diverging epochs from fetch responses.
*/
- private def updateFetchOffsetAndMaybeMarkTruncationComplete(fetchOffsets:
Map[TopicPartition, OffsetTruncationState]): Unit = {
+ private def updateFetchOffsetAndMaybeMarkTruncationComplete(fetchOffsets:
Map[TopicPartition, OffsetTruncationState],
+
maySkipTruncation: Boolean): Unit = {
val newStates: Map[TopicPartition, PartitionFetchState] =
partitionStates.partitionStateMap.asScala
.map { case (topicPartition, currentFetchState) =>
val maybeTruncationComplete = fetchOffsets.get(topicPartition) match {
case Some(offsetTruncationState) =>
- val state = if (offsetTruncationState.truncationCompleted)
Fetching else Truncating
+ val (state, lastFetchedEpoch) = if
(offsetTruncationState.truncationCompleted)
+ (Fetching, latestEpoch(topicPartition))
+ else if (maySkipTruncation &&
currentFetchState.lastFetchedEpoch.nonEmpty)
+ (Fetching, currentFetchState.lastFetchedEpoch)
Review comment:
I was trying to to optimize for the case where lastFetchEpoch didn't
need to be reset, but you are right, it would need resetting after truncation.
Updated to use `latestEpoch(topicPartition))`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
