satishd commented on a change in pull request #11390:
URL: https://github.com/apache/kafka/pull/11390#discussion_r773121400
##########
File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala
##########
@@ -715,6 +725,58 @@ 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) the request succeeded 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,
+ topicId: Option[Uuid],
+ fetchState: PartitionFetchState,
+ requestEpoch: Optional[Integer],
+ leaderLogStartOffset: Long):
Boolean = {
+ try {
+ val newFetchState = fetchOffsetAndBuildRemoteLogAuxState(topicPartition,
topicId, fetchState.currentLeaderEpoch, leaderLogStartOffset)
Review comment:
Addressed it in the latest commit.
--
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
