satishd commented on a change in pull request #11390:
URL: https://github.com/apache/kafka/pull/11390#discussion_r773123927
##########
File path: core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
##########
@@ -386,11 +396,76 @@ class ReplicaFetcherThread(name: String,
}
/**
- * To avoid ISR thrashing, we only throttle a replica on the follower if
it's in the throttled replica list,
- * the quota is exceeded and the replica is not in sync.
+ * To avoid ISR thrashing, we only throttle a replica on the follower if
it's in the throttled replica list,
+ * the quota is exceeded and the replica is not in sync.
*/
private def shouldFollowerThrottle(quota: ReplicaQuota, fetchState:
PartitionFetchState, topicPartition: TopicPartition): Boolean = {
!fetchState.isReplicaInSync && quota.isThrottled(topicPartition) &&
quota.isQuotaExceeded
}
+ override protected def buildRemoteLogAuxState(partition: TopicPartition,
+ currentLeaderEpoch: Int,
+ leaderLocalLogStartOffset:
Long,
+ leaderLogStartOffset: Long):
Unit = {
+ replicaMgr.localLog(partition).foreach(log =>
+ if (log.remoteStorageSystemEnable &&
log.config.remoteLogConfig.remoteStorageEnable) {
+ replicaMgr.remoteLogManager.foreach(rlm => {
+ var rlsMetadata: Optional[RemoteLogSegmentMetadata] =
Optional.empty()
+ val epoch = log.leaderEpochCache.flatMap(cache =>
cache.epochForOffset(leaderLocalLogStartOffset))
+ if (epoch.isDefined) {
+ rlsMetadata = rlm.fetchRemoteLogSegmentMetadata(partition,
epoch.get, leaderLocalLogStartOffset)
+ } else {
+ // If epoch is not available, then it might be possible that this
broker might lost its entire local storage.
+ // We may also have to build the leader epoch cache. To find out
the remote log segment metadata for the
+ // leaderLocalLogStartOffset-1, start from the current leader
epoch and subtract one to the epoch till
+ // finding the metadata.
+ var previousLeaderEpoch = currentLeaderEpoch
+ while (!rlsMetadata.isPresent && previousLeaderEpoch >= 0) {
+ rlsMetadata = rlm.fetchRemoteLogSegmentMetadata(partition,
previousLeaderEpoch, leaderLocalLogStartOffset - 1)
Review comment:
Changed it to use the leader epoch of leader's local-log-start-offset
and find the respective earlier epoch for (leader's local-log-start-offset -1)
from the leader.
--
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
