Add HostedPartition.Fenced to ReplicaManager

core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala

@@ -833,7 +833,7 @@ class GroupMetadataManager(brokerId: Int,
         val timestampType = TimestampType.CREATE_TIME
         val timestamp = time.milliseconds()
 
-          replicaManager.nonOfflinePartition(appendPartition).foreach { partition =>
+          replicaManager.onlinePartition(appendPartition).foreach { partition =>
             val tombstones = ArrayBuffer.empty[SimpleRecord]
             removedOffsets.forKeyValue { (topicPartition, offsetAndMetadata) =>
               trace(s"Removing expired/deleted offset and metadata for $groupId, $topicPartition: $offsetAndMetadata")

core/src/main/scala/kafka/log/LogManager.scala

@@ -346,7 +346,9 @@ class LogManager(logDirs: Seq[File],
         val numLogsLoaded = new AtomicInteger(0)
         numTotalLogs += logsToLoad.length
 
-        val jobsForDir = logsToLoad.map { logDir =>
+        val jobsForDir = logsToLoad
+          .filter(logDir => Log.parseTopicPartitionName(logDir).topic != "@metadata")
+          .map { logDir =>
           val runnable: Runnable = () => {
             try {
               debug(s"Loading log $logDir")

core/src/main/scala/kafka/server/BrokerServer.scala

@@ -187,7 +187,7 @@ class BrokerServer(
       this.replicaManager = new ReplicaManager(config, metrics, time, None,
         kafkaScheduler, logManager, isShuttingDown, quotaManagers,
         brokerTopicStats, metadataCache, logDirFailureChannel, alterIsrManager,
-        configRepository, threadNamePrefix)
+        configRepository, threadNamePrefix, true)
 
       /* start broker-to-controller channel managers */
       val controllerNodes =

core/src/main/scala/kafka/server/DelayedDeleteRecords.scala

@@ -83,6 +83,8 @@ class DelayedDeleteRecords(delayMs: Long,
               case None =>
                 (false, Errors.NOT_LEADER_OR_FOLLOWER, DeleteRecordsResponse.INVALID_LOW_WATERMARK)
             }
+          case HostedPartition.Deferred(_, _, _, _, _) =>
+            (false, Errors.NOT_LEADER_OR_FOLLOWER, DeleteRecordsResponse.INVALID_LOW_WATERMARK)
 
           case HostedPartition.Offline =>
             (false, Errors.KAFKA_STORAGE_ERROR, DeleteRecordsResponse.INVALID_LOW_WATERMARK)

core/src/main/scala/kafka/server/DelayedFetch.scala

@@ -90,7 +90,7 @@ class DelayedFetch(delayMs: Long,
         val fetchLeaderEpoch = fetchStatus.fetchInfo.currentLeaderEpoch
         try {
           if (fetchOffset != LogOffsetMetadata.UnknownOffsetMetadata) {
-            val partition = replicaManager.getPartitionOrException(topicPartition)
+            val partition = replicaManager.onlinePartitionOrException(topicPartition)
             val offsetSnapshot = partition.fetchOffsetSnapshot(fetchLeaderEpoch, fetchMetadata.fetchOnlyLeader)
 
             val endOffset = fetchMetadata.fetchIsolation match {

core/src/main/scala/kafka/server/DelayedProduce.scala

@@ -86,7 +86,7 @@ class DelayedProduce(delayMs: Long,
       trace(s"Checking produce satisfaction for $topicPartition, current status $status")
       // skip those partitions that have already been satisfied
       if (status.acksPending) {
-        val (hasEnough, error) = replicaManager.getPartitionOrError(topicPartition) match {
+        val (hasEnough, error) = replicaManager.onlinePartitionOrError(topicPartition) match {
           case Left(err) =>
             // Case A
             (false, err)

core/src/main/scala/kafka/server/KafkaServer.scala

@@ -388,7 +388,7 @@ class KafkaServer(
       time, config.brokerId, () => kafkaController.brokerEpoch)
     new ReplicaManager(config, metrics, time, Some(zkClient), kafkaScheduler, logManager,
       isShuttingDown, quotaManagers, brokerTopicStats, metadataCache, logDirFailureChannel,
-      alterIsrManager, new ZkConfigRepository(new AdminZkClient(zkClient)), None)
+      alterIsrManager, new ZkConfigRepository(new AdminZkClient(zkClient)), None, false)
   }
 
   private def initZkClient(time: Time): Unit = {

core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala

@@ -60,19 +60,19 @@ class ReplicaAlterLogDirsThread(name: String,
   private var inProgressPartition: Option[TopicPartition] = None
 
   override protected def latestEpoch(topicPartition: TopicPartition): Option[Int] = {
-    replicaMgr.futureLocalLogOrException(topicPartition).latestEpoch
+    replicaMgr.futureLocalOnlineLogOrException(topicPartition).latestEpoch
   }
 
   override protected def logStartOffset(topicPartition: TopicPartition): Long = {
-    replicaMgr.futureLocalLogOrException(topicPartition).logStartOffset
+    replicaMgr.futureLocalOnlineLogOrException(topicPartition).logStartOffset
   }
 
   override protected def logEndOffset(topicPartition: TopicPartition): Long = {
-    replicaMgr.futureLocalLogOrException(topicPartition).logEndOffset
+    replicaMgr.futureLocalOnlineLogOrException(topicPartition).logEndOffset
   }
 
   override protected def endOffsetForEpoch(topicPartition: TopicPartition, epoch: Int): Option[OffsetAndEpoch] = {
-    replicaMgr.futureLocalLogOrException(topicPartition).endOffsetForEpoch(epoch)
+    replicaMgr.futureLocalOnlineLogOrException(topicPartition).endOffsetForEpoch(epoch)
   }
 
   def fetchFromLeader(fetchRequest: FetchRequest.Builder): Map[TopicPartition, FetchData] = {
@@ -110,7 +110,7 @@ class ReplicaAlterLogDirsThread(name: String,
   override def processPartitionData(topicPartition: TopicPartition,
                                     fetchOffset: Long,
                                     partitionData: PartitionData[Records]): Option[LogAppendInfo] = {
-    val partition = replicaMgr.nonOfflinePartition(topicPartition).get
+    val partition = replicaMgr.onlinePartition(topicPartition).get
     val futureLog = partition.futureLocalLogOrException
     val records = toMemoryRecords(partitionData.records)
 
@@ -139,7 +139,7 @@ class ReplicaAlterLogDirsThread(name: String,
       // It is possible that the log dir fetcher completed just before this call, so we
       // filter only the partitions which still have a future log dir.
       val filteredFetchStates = initialFetchStates.filter { case (tp, _) =>
-        replicaMgr.futureLogExists(tp)
+        replicaMgr.futureLocalOnlineLogExists(tp)
       }
       super.addPartitions(filteredFetchStates)
     } finally {
@@ -148,12 +148,12 @@ class ReplicaAlterLogDirsThread(name: String,
   }
 
   override protected def fetchEarliestOffsetFromLeader(topicPartition: TopicPartition, leaderEpoch: Int): Long = {
-    val partition = replicaMgr.getPartitionOrException(topicPartition)
+    val partition = replicaMgr.onlinePartitionOrException(topicPartition)
     partition.localLogOrException.logStartOffset
   }
 
   override protected def fetchLatestOffsetFromLeader(topicPartition: TopicPartition, leaderEpoch: Int): Long = {
-    val partition = replicaMgr.getPartitionOrException(topicPartition)
+    val partition = replicaMgr.onlinePartitionOrException(topicPartition)
     partition.localLogOrException.logEndOffset
   }
 
@@ -170,7 +170,7 @@ class ReplicaAlterLogDirsThread(name: String,
             .setPartition(tp.partition)
             .setErrorCode(Errors.NONE.code)
         } else {
-          val partition = replicaMgr.getPartitionOrException(tp)
+          val partition = replicaMgr.onlinePartitionOrException(tp)
           partition.lastOffsetForLeaderEpoch(
             currentLeaderEpoch = epochData.currentLeaderEpoch,
             leaderEpoch = epochData.leaderEpoch,
@@ -206,12 +206,12 @@ class ReplicaAlterLogDirsThread(name: String,
    * exchange with the current replica to truncate to the largest common log prefix for the topic partition
    */
   override def truncate(topicPartition: TopicPartition, truncationState: OffsetTruncationState): Unit = {
-    val partition = replicaMgr.getPartitionOrException(topicPartition)
+    val partition = replicaMgr.onlinePartitionOrException(topicPartition)
     partition.truncateTo(truncationState.offset, isFuture = true)
   }
 
   override protected def truncateFullyAndStartAt(topicPartition: TopicPartition, offset: Long): Unit = {
-    val partition = replicaMgr.getPartitionOrException(topicPartition)
+    val partition = replicaMgr.onlinePartitionOrException(topicPartition)
     partition.truncateFullyAndStartAt(offset, isFuture = true)
   }
 
@@ -255,7 +255,7 @@ class ReplicaAlterLogDirsThread(name: String,
     val partitionsWithError = mutable.Set[TopicPartition]()
 
     try {
-      val logStartOffset = replicaMgr.futureLocalLogOrException(tp).logStartOffset
+      val logStartOffset = replicaMgr.futureLocalOnlineLogOrException(tp).logStartOffset
       val lastFetchedEpoch = if (isTruncationOnFetchSupported)
         fetchState.lastFetchedEpoch.map(_.asInstanceOf[Integer]).asJava
       else

core/src/main/scala/kafka/server/ReplicaFetcherThread.scala

@@ -109,19 +109,19 @@ class ReplicaFetcherThread(name: String,
   val fetchSessionHandler = new FetchSessionHandler(logContext, sourceBroker.id)
 
   override protected def latestEpoch(topicPartition: TopicPartition): Option[Int] = {
-    replicaMgr.localLogOrException(topicPartition).latestEpoch
+    replicaMgr.localOnlineLogOrException(topicPartition).latestEpoch
   }
 
   override protected def logStartOffset(topicPartition: TopicPartition): Long = {
-    replicaMgr.localLogOrException(topicPartition).logStartOffset
+    replicaMgr.localOnlineLogOrException(topicPartition).logStartOffset
   }
 
   override protected def logEndOffset(topicPartition: TopicPartition): Long = {
-    replicaMgr.localLogOrException(topicPartition).logEndOffset
+    replicaMgr.localOnlineLogOrException(topicPartition).logEndOffset
   }
 
   override protected def endOffsetForEpoch(topicPartition: TopicPartition, epoch: Int): Option[OffsetAndEpoch] = {
-    replicaMgr.localLogOrException(topicPartition).endOffsetForEpoch(epoch)
+    replicaMgr.localOnlineLogOrException(topicPartition).endOffsetForEpoch(epoch)
   }
 
   override def initiateShutdown(): Boolean = {
@@ -158,7 +158,7 @@ class ReplicaFetcherThread(name: String,
                                     fetchOffset: Long,
                                     partitionData: FetchData): Option[LogAppendInfo] = {
     val logTrace = isTraceEnabled
-    val partition = replicaMgr.nonOfflinePartition(topicPartition).get
+    val partition = replicaMgr.onlinePartition(topicPartition).get
     val log = partition.localLogOrException
     val records = toMemoryRecords(partitionData.records)
 
@@ -308,7 +308,7 @@ class ReplicaFetcherThread(name: String,
    * The logic for finding the truncation offset is implemented in AbstractFetcherThread.getOffsetTruncationState
    */
   override def truncate(tp: TopicPartition, offsetTruncationState: OffsetTruncationState): Unit = {
-    val partition = replicaMgr.nonOfflinePartition(tp).get
+    val partition = replicaMgr.onlinePartition(tp).get
     val log = partition.localLogOrException
 
     partition.truncateTo(offsetTruncationState.offset, isFuture = false)
@@ -324,7 +324,7 @@ class ReplicaFetcherThread(name: String,
   }
 
   override protected def truncateFullyAndStartAt(topicPartition: TopicPartition, offset: Long): Unit = {
-    val partition = replicaMgr.nonOfflinePartition(topicPartition).get
+    val partition = replicaMgr.onlinePartition(topicPartition).get
     partition.truncateFullyAndStartAt(offset, isFuture = false)
   }