satishd commented on a change in pull request #11390:
URL: https://github.com/apache/kafka/pull/11390#discussion_r773121903
##########
File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala
##########
@@ -715,6 +726,57 @@ abstract class AbstractFetcherThread(name: String,
}
}
+ /**
+ * Handle a partition whose offset is out of range and return a new fetch
offset.
+ */
+ protected def fetchOffsetAndTruncate(topicPartition: TopicPartition,
topicId: Option[Uuid], currentLeaderEpoch: Int): PartitionFetchState = {
+ fetchOffsetAndApplyFun(topicPartition, topicId, currentLeaderEpoch,
+ leaderLogStartOffset => truncateFullyAndStartAt(topicPartition,
leaderLogStartOffset))
+ }
+
+ /**
+ * Handle a partition whose offset is moved to tiered storage and return a
new fetch offset.
+ */
+ protected def fetchOffsetAndBuildRemoteLogAuxState(topicPartition:
TopicPartition, topicId: Option[Uuid],
+ currentLeaderEpoch: Int,
+ leaderLogStartOffset:
Long): PartitionFetchState = {
+ fetchOffsetAndApplyFun(topicPartition, topicId, currentLeaderEpoch,
+ leaderLocalLogStartOffset =>
+ buildRemoteLogAuxState(topicPartition, currentLeaderEpoch,
leaderLocalLogStartOffset, leaderLogStartOffset))
+ }
+
+ /**
+ * Handle the offset moved to tiered storage error.
+ *
+ * Return false if
+ * 1) it is able to build the required remote log auxiliary state or
+ * 2) was fenced and this thread haven't received new epoch,
+ * which means we need not backoff and retry. True if there was a retriable
error.
+ */
+ private def handleOffsetMovedToTieredStorage(topicPartition: TopicPartition,
Review comment:
This is inline with `handleOutOfRangeError` contract. I am fine with the
suggested change but it is good to have similar semantics to
`handleOutOfRangeError` method too for uniformity.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
