Add HostedPartition.Fenced to ReplicaManager

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/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 {
@@ -184,7 +184,7 @@ class DelayedFetch(delayMs: Long,
 
     val fetchPartitionData = logReadResults.map { case (tp, result) =>
       val isReassignmentFetch = fetchMetadata.isFromFollower &&
-        replicaManager.isAddingReplica(tp, fetchMetadata.replicaId)
+        replicaManager.isAddingReplica(tp, fetchMetadata.replicaId, false)
 
       tp -> FetchPartitionData(
         result.error,

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/ReplicaAlterLogDirsThread.scala

@@ -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)
   }
 

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 = {