satishd commented on a change in pull request #11390:
URL: https://github.com/apache/kafka/pull/11390#discussion_r767430230
##########
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)
+ previousLeaderEpoch -= 1
+ }
+ }
+ if (rlsMetadata.isPresent) {
+ val epochStream =
rlm.storageManager().fetchIndex(rlsMetadata.get(),
RemoteStorageManager.IndexType.LEADER_EPOCH)
+ val epochs = readLeaderEpochCheckpoint(epochStream, log.dir)
+
+ // Truncate the existing local log before restoring the leader
epoch cache and producer snapshots.
+ truncateFullyAndStartAt(partition, leaderLocalLogStartOffset)
+
+ log.maybeIncrementLogStartOffset(leaderLogStartOffset,
LeaderOffsetIncremented)
+ epochs.foreach(epochEntry => {
+ log.leaderEpochCache.map(cache => cache.assign(epochEntry.epoch,
epochEntry.startOffset))
+ })
+ info(s"Updated the epoch cache from remote tier till offset:
$leaderLocalLogStartOffset " +
+ s"with size: ${epochs.size} for $partition")
+
+ // Restore producer snapshot
+ val snapshotFile = UnifiedLog.producerSnapshotFile(log.dir,
leaderLocalLogStartOffset)
+ Files.copy(rlm.storageManager().fetchIndex(rlsMetadata.get(),
RemoteStorageManager.IndexType.PRODUCER_SNAPSHOT),
+ snapshotFile.toPath, StandardCopyOption.REPLACE_EXISTING)
+ log.producerStateManager.reloadSegments()
+ log.loadProducerState(leaderLocalLogStartOffset,
reloadFromCleanShutdown = false)
Review comment:
`rlsMetadata = rlm.fetchRemoteLogSegmentMetadata(partition, epoch.get,
leaderLocalLogStartOffset)` returns empty and this method throws an error back
to the caller here. It retries again until the required `rlsMetadata` is
available. `RemoteLogMetadataSnapshotFile` is loaded when assigning of
partitions are done as part of
`TopicBasedRemoteLogMetadataManager.assignPartitions()`.
--
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
