diff --git a/core/src/main/scala/kafka/server/DelayedDeleteRecords.scala b/core/src/main/scala/kafka/server/DelayedDeleteRecords.scala index 164cab5ad058d..dae17c60aaa49 100644 --- a/core/src/main/scala/kafka/server/DelayedDeleteRecords.scala +++ b/core/src/main/scala/kafka/server/DelayedDeleteRecords.scala @@ -84,7 +84,7 @@ class DelayedDeleteRecords(delayMs: Long, (false, Errors.NOT_LEADER_OR_FOLLOWER, DeleteRecordsResponse.INVALID_LOW_WATERMARK) } - case HostedPartition.Deferred(_) => + case _: HostedPartition.Deferred => (false, Errors.UNKNOWN_TOPIC_OR_PARTITION, DeleteRecordsResponse.INVALID_LOW_WATERMARK) case HostedPartition.Offline => diff --git a/core/src/main/scala/kafka/server/RaftReplicaChangeDelegate.scala b/core/src/main/scala/kafka/server/RaftReplicaChangeDelegate.scala new file mode 100644 index 0000000000000..0b7ee4264ae26 --- /dev/null +++ b/core/src/main/scala/kafka/server/RaftReplicaChangeDelegate.scala @@ -0,0 +1,246 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package kafka.server + +import kafka.cluster.Partition +import kafka.controller.StateChangeLogger +import kafka.log.Log +import kafka.server.checkpoints.OffsetCheckpoints +import kafka.server.metadata.{MetadataBrokers, MetadataPartition} +import kafka.utils.Implicits.MapExtensionMethods +import org.apache.kafka.common.TopicPartition +import org.apache.kafka.common.errors.KafkaStorageException + +import scala.collection.{Map, Set, mutable} + +trait RaftReplicaChangeDelegateHelper { + def stateChangeLogger: StateChangeLogger + def replicaFetcherManager: ReplicaFetcherManager + def replicaAlterLogDirsManager: ReplicaAlterLogDirsManager + def markDeferred(state: HostedPartition.Deferred): Unit + def getLogDir(topicPartition: TopicPartition): Option[String] + def error(msg: => String, e: => Throwable): Unit + def markOffline(topicPartition: TopicPartition): Unit + def completeDelayedFetchOrProduceRequests(topicPartition: TopicPartition): Unit + def isShuttingDown: Boolean + def initialFetchOffset(log: Log): Long + def config: KafkaConfig +} + +class RaftReplicaChangeDelegate(helper: RaftReplicaChangeDelegateHelper) { + def makeDeferred(partitionsNewMap: Map[Partition, Boolean], + metadataOffset: Long): Unit = { + val traceLoggingEnabled = helper.stateChangeLogger.isTraceEnabled + if (traceLoggingEnabled) + partitionsNewMap.forKeyValue { (partition, isNew) => + helper.stateChangeLogger.trace(s"Metadata batch $metadataOffset: starting the " + + s"become-deferred transition for partition ${partition.topicPartition} isNew=$isNew") + } + + // Stop fetchers for all the partitions + helper.replicaFetcherManager.removeFetcherForPartitions(partitionsNewMap.keySet.map(_.topicPartition)) + helper.stateChangeLogger.info(s"Metadata batch $metadataOffset: as part of become-deferred request, " + + s"stopped any fetchers for ${partitionsNewMap.size} partitions") + // mark all the partitions as deferred + partitionsNewMap.forKeyValue((partition, isNew) => helper.markDeferred(HostedPartition.Deferred(partition, isNew))) + + helper.replicaFetcherManager.shutdownIdleFetcherThreads() + helper.replicaAlterLogDirsManager.shutdownIdleFetcherThreads() + + if (traceLoggingEnabled) + partitionsNewMap.keys.foreach { partition => + helper.stateChangeLogger.trace(s"Completed batch $metadataOffset become-deferred " + + s"transition for partition ${partition.topicPartition}") + } + } + + def makeLeaders(prevPartitionsAlreadyExisting: Set[MetadataPartition], + partitionStates: Map[Partition, MetadataPartition], + highWatermarkCheckpoints: OffsetCheckpoints, + metadataOffset: Option[Long]): Set[Partition] = { + val partitionsMadeLeaders = mutable.Set[Partition]() + val traceLoggingEnabled = helper.stateChangeLogger.isTraceEnabled + val deferredBatches = metadataOffset.isEmpty + val topLevelLogPrefix = if (deferredBatches) + "Metadata batch " + else + s"Metadata batch ${metadataOffset.get}" + try { + // First stop fetchers for all the partitions + helper.replicaFetcherManager.removeFetcherForPartitions(partitionStates.keySet.map(_.topicPartition)) + helper.stateChangeLogger.info(s"$topLevelLogPrefix: stopped ${partitionStates.size} fetcher(s)") + // Update the partition information to be the leader + partitionStates.forKeyValue { (partition, state) => + val topicPartition = partition.topicPartition + val partitionLogMsgPrefix = if (deferredBatches) + s"Apply deferred leader partition $topicPartition" + else + s"Metadata batch ${metadataOffset.get} $topicPartition" + try { + val isrState = state.toLeaderAndIsrPartitionState( + !prevPartitionsAlreadyExisting(state)) + if (partition.makeLeader(isrState, highWatermarkCheckpoints)) { + partitionsMadeLeaders += partition + if (traceLoggingEnabled) { + helper.stateChangeLogger.trace(s"$partitionLogMsgPrefix: completed the become-leader state change.") + } + } else { + helper.stateChangeLogger.info(s"$partitionLogMsgPrefix: skipped the " + + "become-leader state change since it is already the leader.") + } + } catch { + case e: KafkaStorageException => + helper.stateChangeLogger.error(s"$partitionLogMsgPrefix: unable to make " + + s"leader because the replica for the partition is offline due to disk error $e") + val dirOpt = helper.getLogDir(topicPartition) + helper.error(s"Error while making broker the leader for partition $partition in dir $dirOpt", e) + helper.markOffline(topicPartition) + } + } + } catch { + case e: Throwable => + helper.stateChangeLogger.error(s"$topLevelLogPrefix: error while processing batch.", e) + // Re-throw the exception for it to be caught in BrokerMetadataListener + throw e + } + partitionsMadeLeaders + } + + def makeFollowers(prevPartitionsAlreadyExisting: Set[MetadataPartition], + currentBrokers: MetadataBrokers, + partitionStates: Map[Partition, MetadataPartition], + highWatermarkCheckpoints: OffsetCheckpoints, + metadataOffset: Option[Long]): Set[Partition] = { + val traceLoggingEnabled = helper.stateChangeLogger.isTraceEnabled + val deferredBatches = metadataOffset.isEmpty + val topLevelLogPrefix = if (deferredBatches) + "Metadata batch " + else + s"Metadata batch ${metadataOffset.get}" + if (traceLoggingEnabled) { + partitionStates.forKeyValue { (partition, state) => + val topicPartition = partition.topicPartition + val partitionLogMsgPrefix = if (deferredBatches) + s"Apply deferred follower partition $topicPartition" + else + s"Metadata batch ${metadataOffset.get} $topicPartition" + helper.stateChangeLogger.trace(s"$partitionLogMsgPrefix: starting the " + + s"become-follower transition with leader ${state.leaderId}") + } + } + + val partitionsMadeFollower: mutable.Set[Partition] = mutable.Set() + // all brokers, including both alive and not + val acceptableLeaderBrokerIds = currentBrokers.iterator().map(broker => broker.id).toSet + val allBrokersByIdMap = currentBrokers.iterator().map(broker => broker.id -> broker).toMap + try { + partitionStates.forKeyValue { (partition, state) => + val topicPartition = partition.topicPartition + val partitionLogMsgPrefix = if (deferredBatches) + s"Apply deferred follower partition $topicPartition" + else + s"Metadata batch ${metadataOffset.get} $topicPartition" + try { + val isNew = !prevPartitionsAlreadyExisting(state) + if (!acceptableLeaderBrokerIds.contains(state.leaderId)) { + // The leader broker should always be present in the metadata cache. + // If not, we should record the error message and abort the transition process for this partition + helper.stateChangeLogger.error(s"$partitionLogMsgPrefix: cannot become follower " + + s"since the new leader ${state.leaderId} is unavailable.") + // Create the local replica even if the leader is unavailable. This is required to ensure that we include + // the partition's high watermark in the checkpoint file (see KAFKA-1647) + partition.createLogIfNotExists(isNew, isFutureReplica = false, highWatermarkCheckpoints) + } else { + val isrState = state.toLeaderAndIsrPartitionState(isNew) + if (partition.makeFollower(isrState, highWatermarkCheckpoints)) { + partitionsMadeFollower += partition + if (traceLoggingEnabled) { + helper.stateChangeLogger.trace(s"$partitionLogMsgPrefix: completed the " + + s"become-follower state change with new leader ${state.leaderId}.") + } + } else { + helper.stateChangeLogger.info(s"$partitionLogMsgPrefix: skipped the " + + s"become-follower state change since " + + s"the new leader ${state.leaderId} is the same as the old leader.") + } + } + } catch { + case e: KafkaStorageException => + helper.stateChangeLogger.error(s"$partitionLogMsgPrefix: unable to complete the " + + s"become-follower state change since the " + + s"replica for the partition is offline due to disk error $e") + val dirOpt = helper.getLogDir(partition.topicPartition) + helper.error(s"Error while making broker the follower with leader ${state.leaderId} in dir $dirOpt", e) + helper.markOffline(topicPartition) + } + } + + if (partitionsMadeFollower.nonEmpty) { + helper.replicaFetcherManager.removeFetcherForPartitions(partitionsMadeFollower.map(_.topicPartition)) + helper.stateChangeLogger.info(s"$topLevelLogPrefix: stopped followers for ${partitionsMadeFollower.size} partitions") + + partitionsMadeFollower.foreach { partition => + helper.completeDelayedFetchOrProduceRequests(partition.topicPartition) + } + + if (helper.isShuttingDown) { + if (traceLoggingEnabled) { + partitionsMadeFollower.foreach { partition => + val topicPartition = partition.topicPartition + val partitionLogMsgPrefix = if (deferredBatches) + s"Apply deferred follower partition $topicPartition" + else + s"Metadata batch ${metadataOffset.get} $topicPartition" + helper.stateChangeLogger.trace(s"$partitionLogMsgPrefix: skipped the " + + s"adding-fetcher step of the become-follower state for " + + s"$topicPartition since we are shutting down.") + } + } + } else { + // we do not need to check if the leader exists again since this has been done at the beginning of this process + val partitionsToMakeFollowerWithLeaderAndOffset = partitionsMadeFollower.map { partition => + val leader = allBrokersByIdMap(partition.leaderReplicaIdOpt.get).brokerEndPoint(helper.config.interBrokerListenerName) + val log = partition.localLogOrException + val fetchOffset = helper.initialFetchOffset(log) + partition.topicPartition -> InitialFetchState(leader, partition.getLeaderEpoch, fetchOffset) + }.toMap + + helper.replicaFetcherManager.addFetcherForPartitions(partitionsToMakeFollowerWithLeaderAndOffset) + } + } + } catch { + case e: Throwable => + helper.stateChangeLogger.error(s"$topLevelLogPrefix: error while processing batch", e) + // Re-throw the exception for it to be caught in BrokerMetadataListener + throw e + } + + if (traceLoggingEnabled) + partitionsMadeFollower.foreach { partition => + val topicPartition = partition.topicPartition + val state = partitionStates(partition) + val partitionLogMsgPrefix = if (deferredBatches) + s"Apply deferred follower partition $topicPartition" + else + s"Metadata batch ${metadataOffset.get} $topicPartition" + helper.stateChangeLogger.trace(s"$partitionLogMsgPrefix: completed become-follower " + + s"transition for partition $topicPartition with new leader ${state.leaderId}") + } + + partitionsMadeFollower + } +} diff --git a/core/src/main/scala/kafka/server/RaftReplicaManager.scala b/core/src/main/scala/kafka/server/RaftReplicaManager.scala new file mode 100644 index 0000000000000..255b34974d3b6 --- /dev/null +++ b/core/src/main/scala/kafka/server/RaftReplicaManager.scala @@ -0,0 +1,377 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package kafka.server + +import java.util.concurrent.atomic.AtomicBoolean + +import kafka.cluster.Partition +import kafka.controller.StateChangeLogger +import kafka.log.{Log, LogManager} +import kafka.server.QuotaFactory.QuotaManagers +import kafka.server.checkpoints.LazyOffsetCheckpoints +import kafka.server.metadata.{ConfigRepository, MetadataImage, MetadataImageBuilder, MetadataPartition, RaftMetadataCache} +import kafka.utils.Scheduler +import org.apache.kafka.common.TopicPartition +import org.apache.kafka.common.errors.KafkaStorageException +import org.apache.kafka.common.metrics.Metrics +import org.apache.kafka.common.utils.Time + +import scala.collection.{Set, mutable} + +class RaftReplicaManager(config: KafkaConfig, + metrics: Metrics, + time: Time, + scheduler: Scheduler, + logManager: LogManager, + isShuttingDown: AtomicBoolean, + quotaManagers: QuotaManagers, + brokerTopicStats: BrokerTopicStats, + metadataCache: RaftMetadataCache, + logDirFailureChannel: LogDirFailureChannel, + delayedProducePurgatory: DelayedOperationPurgatory[DelayedProduce], + delayedFetchPurgatory: DelayedOperationPurgatory[DelayedFetch], + delayedDeleteRecordsPurgatory: DelayedOperationPurgatory[DelayedDeleteRecords], + delayedElectLeaderPurgatory: DelayedOperationPurgatory[DelayedElectLeader], + threadNamePrefix: Option[String], + configRepository: ConfigRepository, + alterIsrManager: AlterIsrManager) extends ReplicaManager( + config, metrics, time, None, scheduler, logManager, isShuttingDown, quotaManagers, + brokerTopicStats, metadataCache, logDirFailureChannel, delayedProducePurgatory, delayedFetchPurgatory, + delayedDeleteRecordsPurgatory, delayedElectLeaderPurgatory, threadNamePrefix, configRepository, alterIsrManager) { + + def this(config: KafkaConfig, + metrics: Metrics, + time: Time, + scheduler: Scheduler, + logManager: LogManager, + isShuttingDown: AtomicBoolean, + quotaManagers: QuotaManagers, + brokerTopicStats: BrokerTopicStats, + metadataCache: RaftMetadataCache, + logDirFailureChannel: LogDirFailureChannel, + alterIsrManager: AlterIsrManager, + configRepository: ConfigRepository, + threadNamePrefix: Option[String] = None) = { + this(config, metrics, time, scheduler, logManager, isShuttingDown, + quotaManagers, brokerTopicStats, metadataCache, logDirFailureChannel, + DelayedOperationPurgatory[DelayedProduce]( + purgatoryName = "Produce", brokerId = config.brokerId, + purgeInterval = config.producerPurgatoryPurgeIntervalRequests), + DelayedOperationPurgatory[DelayedFetch]( + purgatoryName = "Fetch", brokerId = config.brokerId, + purgeInterval = config.fetchPurgatoryPurgeIntervalRequests), + DelayedOperationPurgatory[DelayedDeleteRecords]( + purgatoryName = "DeleteRecords", brokerId = config.brokerId, + purgeInterval = config.deleteRecordsPurgatoryPurgeIntervalRequests), + DelayedOperationPurgatory[DelayedElectLeader]( + purgatoryName = "ElectLeader", brokerId = config.brokerId), + threadNamePrefix, configRepository, alterIsrManager) + } + + if (config.requiresZookeeper) { + throw new IllegalStateException(s"Cannot use ${getClass.getSimpleName} when using ZooKeeper") + } + + class RaftReplicaManagerChangeDelegateHelper(raftReplicaManager: RaftReplicaManager) extends RaftReplicaChangeDelegateHelper { + override def completeDelayedFetchOrProduceRequests(topicPartition: TopicPartition): Unit = raftReplicaManager.completeDelayedFetchOrProduceRequests(topicPartition) + + override def config: KafkaConfig = raftReplicaManager.config + + override def error(msg: => String, e: => Throwable): Unit = raftReplicaManager.error(msg, e) + + override def getLogDir(topicPartition: TopicPartition): Option[String] = raftReplicaManager.getLogDir(topicPartition) + + override def initialFetchOffset(log: Log): Long = raftReplicaManager.initialFetchOffset(log) + + override def isShuttingDown: Boolean = raftReplicaManager.isShuttingDown.get + + override def markDeferred(state: HostedPartition.Deferred): Unit = raftReplicaManager.markPartitionDeferred(state) + + override def markOffline(topicPartition: TopicPartition): Unit = raftReplicaManager.markPartitionOffline(topicPartition) + + override def replicaAlterLogDirsManager: ReplicaAlterLogDirsManager = raftReplicaManager.replicaAlterLogDirsManager + + override def replicaFetcherManager: ReplicaFetcherManager = raftReplicaManager.replicaFetcherManager + + override def stateChangeLogger: StateChangeLogger = raftReplicaManager.stateChangeLogger + } + + // visible/overwriteable for testing, generally will not change otherwise + private[server] var delegate = new RaftReplicaChangeDelegate(new RaftReplicaManagerChangeDelegateHelper(this)) + + // Changes are initially deferred when using a Raft-based metadata quorum, and they may flip-flop to not + // being deferred and being deferred again thereafter as the broker (re)acquires/loses its lease. + // Changes are never deferred when using ZooKeeper. When true, this indicates that we should transition + // online partitions to the deferred state if we see a metadata update for that partition. + private var deferringMetadataChanges: Boolean = true + stateChangeLogger.debug(s"Metadata changes are initially being deferred") + + def beginMetadataChangeDeferral(): Unit = { + replicaStateChangeLock synchronized { + deferringMetadataChanges = true + stateChangeLogger.info(s"Metadata changes are now being deferred") + } + } + + def endMetadataChangeDeferral(onLeadershipChange: (Iterable[Partition], Iterable[Partition]) => Unit): Unit = { + val startMs = time.milliseconds() + val partitionsMadeFollower = mutable.Set[Partition]() + val partitionsMadeLeader = mutable.Set[Partition]() + replicaStateChangeLock synchronized { + stateChangeLogger.info(s"Applying deferred metadata changes") + val highWatermarkCheckpoints = new LazyOffsetCheckpoints(this.highWatermarkCheckpoints) + val metadataImage = metadataCache.currentImage() + val brokers = metadataImage.brokers + try { + val leaderPartitionStates = mutable.Map[Partition, MetadataPartition]() + val followerPartitionStates = mutable.Map[Partition, MetadataPartition]() + val partitionsAlreadyExisting = mutable.Set[MetadataPartition]() + deferredPartitionsIterator.foreach { deferredPartition => + val partition = deferredPartition.partition + val state = cachedState(metadataImage, partition) + if (state.leaderId == localBrokerId) { + leaderPartitionStates.put(partition, state) + } else { + followerPartitionStates.put(partition, state) + } + if (!deferredPartition.isNew) { + partitionsAlreadyExisting += state + } + } + + val partitionsMadeLeader = if (leaderPartitionStates.nonEmpty) + delegate.makeLeaders(partitionsAlreadyExisting, leaderPartitionStates, highWatermarkCheckpoints, None) + else + Set.empty[Partition] + val partitionsMadeFollower = if (followerPartitionStates.nonEmpty) + delegate.makeFollowers(partitionsAlreadyExisting, brokers, followerPartitionStates, highWatermarkCheckpoints, None) + else + Set.empty[Partition] + + // We need to transition anything that hasn't transitioned from Deferred to Offline to the Online state. + deferredPartitionsIterator.foreach { deferredPartition => + val partition = deferredPartition.partition + allPartitions.put(partition.topicPartition, HostedPartition.Online(partition)) + } + + updateLeaderAndFollowerMetrics(partitionsMadeFollower.map(_.topic).toSet) + + maybeAddLogDirFetchers(partitionsMadeFollower, highWatermarkCheckpoints) + + replicaFetcherManager.shutdownIdleFetcherThreads() + replicaAlterLogDirsManager.shutdownIdleFetcherThreads() + if (partitionsMadeLeader.nonEmpty || partitionsMadeFollower.nonEmpty) { + onLeadershipChange(partitionsMadeLeader, partitionsMadeFollower) + } + } catch { + case e: Throwable => + deferredPartitionsIterator.foreach { metadata => + val partition = metadata.partition + val state = cachedState(metadataImage, partition) + val topicPartition = partition.topicPartition + val leader = state.leaderId == localBrokerId + val leaderOrFollower = if (leader) "leader" else "follower" + val partitionLogMsgPrefix = s"Apply deferred $leaderOrFollower partition $topicPartition" + stateChangeLogger.error(s"$partitionLogMsgPrefix: error while applying deferred metadata.", e) + } + stateChangeLogger.info(s"Applied ${partitionsMadeLeader.size + partitionsMadeFollower.size} deferred partitions prior to the error: " + + s"${partitionsMadeLeader.size} leader(s) and ${partitionsMadeFollower.size} follower(s)") + // Re-throw the exception for it to be caught in BrokerMetadataListener + throw e + } + deferringMetadataChanges = false + } + val endMs = time.milliseconds() + val elapsedMs = endMs - startMs + stateChangeLogger.info(s"Applied ${partitionsMadeLeader.size + partitionsMadeFollower.size} deferred partitions: " + + s"${partitionsMadeLeader.size} leader(s) and ${partitionsMadeFollower.size} follower(s)" + + s"in $elapsedMs ms") + stateChangeLogger.info("Metadata changes are no longer being deferred") + } + + /** + * Handle changes made by a batch of metadata log records. + * + * @param imageBuilder The MetadataImage builder. + * @param metadataOffset The last offset in the batch of records. + * @param onLeadershipChange The callbacks to invoke when leadership changes. + */ + def handleMetadataRecords(imageBuilder: MetadataImageBuilder, + metadataOffset: Long, + onLeadershipChange: (Iterable[Partition], Iterable[Partition]) => Unit): Unit = { + val startMs = time.milliseconds() + val builder = imageBuilder.partitionsBuilder() + replicaStateChangeLock synchronized { + stateChangeLogger.info(("Metadata batch %d: %d local partition(s) changed, %d " + + "local partition(s) removed.").format(metadataOffset, builder.localChanged().size, + builder.localRemoved().size)) + if (stateChangeLogger.isTraceEnabled) { + builder.localChanged().foreach { state => + stateChangeLogger.trace(s"Metadata batch $metadataOffset: locally changed: ${state}") + } + builder.localRemoved().foreach { state => + stateChangeLogger.trace(s"Metadata batch $metadataOffset: locally removed: ${state}") + } + } + if (deferringMetadataChanges) { + val prevPartitions = imageBuilder.prevImage.partitions + // partitionChangesToBeDeferred maps each partition to be deferred to whether it is new (i.e. existed before deferral began) + val partitionChangesToBeDeferred = mutable.HashMap[Partition, Boolean]() + builder.localChanged().foreach { currentState => + val topicPartition = currentState.toTopicPartition + val (partition, priorDeferredMetadata) = getPartition(topicPartition) match { + case HostedPartition.Offline => + stateChangeLogger.warn(s"Ignoring handlePartitionChanges at $metadataOffset " + + s"for partition $topicPartition as the local replica for the partition is " + + "in an offline log directory") + (None, None) + + case HostedPartition.Online(partition) => (Some(partition), None) + case deferred@HostedPartition.Deferred(partition, _) => (Some(partition), Some(deferred)) + + case HostedPartition.None => + // Create the partition instance since it does not yet exist + (Some(Partition(topicPartition, time, configRepository, this)), None) + } + partition.foreach { partition => + val isNew = priorDeferredMetadata match { + case Some(alreadyDeferred) => alreadyDeferred.isNew + case _ => prevPartitions.topicPartition(topicPartition.topic(), topicPartition.partition()).isEmpty + } + partitionChangesToBeDeferred.put(partition, isNew) + } + } + + stateChangeLogger.info(s"Deferring metadata changes for ${partitionChangesToBeDeferred.size} partition(s)") + if (partitionChangesToBeDeferred.nonEmpty) { + delegate.makeDeferred(partitionChangesToBeDeferred, metadataOffset) + } + } else { // not deferring changes, so make leaders/followers accordingly + val partitionsToBeLeader = mutable.HashMap[Partition, MetadataPartition]() + val partitionsToBeFollower = mutable.HashMap[Partition, MetadataPartition]() + builder.localChanged().foreach { currentState => + val topicPartition = currentState.toTopicPartition + val partition = getPartition(topicPartition) match { + case HostedPartition.Offline => + stateChangeLogger.warn(s"Ignoring handlePartitionChanges at $metadataOffset " + + s"for partition $topicPartition as the local replica for the partition is " + + "in an offline log directory") + None + + case HostedPartition.Online(partition) => Some(partition) + case _: HostedPartition.Deferred => throw new IllegalStateException( + s"There should never be deferred partition metadata when we aren't deferring changes: $topicPartition") + + case HostedPartition.None => + // it's a partition that we don't know about yet, so create it and mark it online + val partition = Partition(topicPartition, time, configRepository, this) + allPartitions.putIfNotExists(topicPartition, HostedPartition.Online(partition)) + Some(partition) + } + partition.foreach { partition => + if (currentState.leaderId == localBrokerId) { + partitionsToBeLeader.put(partition, currentState) + } else { + partitionsToBeFollower.put(partition, currentState) + } + } + } + + val prevPartitions = imageBuilder.prevImage.partitions + val changedPartitionsPreviouslyExisting = mutable.Set[MetadataPartition]() + builder.localChanged().foreach(metadataPartition => + prevPartitions.topicPartition(metadataPartition.topicName, metadataPartition.partitionIndex).foreach( + changedPartitionsPreviouslyExisting.add)) + val nextBrokers = imageBuilder.brokers() + val highWatermarkCheckpoints = new LazyOffsetCheckpoints(this.highWatermarkCheckpoints) + val partitionsBecomeLeader = if (partitionsToBeLeader.nonEmpty) + delegate.makeLeaders(changedPartitionsPreviouslyExisting, partitionsToBeLeader, highWatermarkCheckpoints, + Some(metadataOffset)) + else + Set.empty[Partition] + val partitionsBecomeFollower = if (partitionsToBeFollower.nonEmpty) + delegate.makeFollowers(changedPartitionsPreviouslyExisting, nextBrokers, partitionsToBeFollower, highWatermarkCheckpoints, + Some(metadataOffset)) + else + Set.empty[Partition] + updateLeaderAndFollowerMetrics(partitionsBecomeFollower.map(_.topic).toSet) + + builder.localChanged().foreach { state => + val topicPartition = state.toTopicPartition + /* + * If there is offline log directory, a Partition object may have been created by getOrCreatePartition() + * before getOrCreateReplica() failed to create local replica due to KafkaStorageException. + * In this case ReplicaManager.allPartitions will map this topic-partition to an empty Partition object. + * we need to map this topic-partition to OfflinePartition instead. + */ + if (localLog(topicPartition).isEmpty) { + markPartitionOffline(topicPartition) + } + } + + maybeAddLogDirFetchers(partitionsBecomeFollower, highWatermarkCheckpoints) + + replicaFetcherManager.shutdownIdleFetcherThreads() + replicaAlterLogDirsManager.shutdownIdleFetcherThreads() + onLeadershipChange(partitionsBecomeLeader, partitionsBecomeFollower) + } + // TODO: we should move aside log directories which have been deleted rather than + // purging them from the disk immediately. + if (builder.localRemoved().nonEmpty) { + // we schedule removal immediately even if we are deferring changes + stopPartitions(builder.localRemoved().map(_.toTopicPartition -> true).toMap).foreach { case (topicPartition, e) => + if (e.isInstanceOf[KafkaStorageException]) { + stateChangeLogger.error(s"Metadata batch $metadataOffset: unable to delete " + + s"${topicPartition} as the local replica for the partition is in an offline " + + "log directory") + } else { + stateChangeLogger.error(s"Metadata batch $metadataOffset: unable to delete " + + s"${topicPartition} due to an unexpected ${e.getClass.getName} exception: " + + s"${e.getMessage}") + } + } + } + } + val endMs = time.milliseconds() + val elapsedMs = endMs - startMs + stateChangeLogger.info(s"Metadata batch $metadataOffset: handled replica changes " + + s"in ${elapsedMs} ms") + } + + def markPartitionDeferred(partition: Partition, isNew: Boolean): Unit = { + markPartitionDeferred(HostedPartition.Deferred(partition, isNew)) + } + + private def markPartitionDeferred(state: HostedPartition.Deferred): Unit = replicaStateChangeLock synchronized { + allPartitions.put(state.partition.topicPartition, state) + } + + // An iterator over all deferred partitions. This is a weakly consistent iterator; a partition made off/online + // after the iterator has been constructed could still be returned by this iterator. + private def deferredPartitionsIterator: Iterator[HostedPartition.Deferred] = { + allPartitions.values.iterator.flatMap { + case deferred: HostedPartition.Deferred => Some(deferred) + case _ => None + } + } + + private def cachedState(metadataImage: MetadataImage, partition: Partition): MetadataPartition = { + metadataImage.partitions.topicPartition(partition.topic, partition.partitionId).getOrElse( + throw new IllegalStateException(s"Partition has metadata changes but does not exist in the metadata cache: ${partition.topicPartition}")) + } +} diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index e6397274e58a9..3f52e22507a78 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -163,7 +163,7 @@ case class FetchPartitionData(error: Errors = Errors.NONE, sealed trait HostedPartition /** - * Trait to represent a partition that isn't Offline -- i.e. it is either Online or it is Deferred + * Trait to represent a partition that isn't Offline -- i.e. it is either Online or it is Deferred. */ sealed trait NonOffline extends HostedPartition { val partition: Partition @@ -180,12 +180,13 @@ object HostedPartition { final case class Online(partition: Partition) extends NonOffline /** - * This broker hosted the partition but it is deferring metadata changes - * until it catches up on the Raft-based metadata log. - * This state only applies to brokers that are using a Raft-based metadata - * quorum; it never happens when using ZooKeeper. + * This broker hosted the partition (or will soon host it if it is new) but + * it is deferring metadata changes until it catches up on the Raft-based metadata + * log. This state only applies to brokers that are using a Raft-based metadata + * quorum; it never happens when using ZooKeeper. The isNew value indicates + * if the partition needs to be created when we apply the deferred changes. */ - final case class Deferred(partition: Partition) extends NonOffline + final case class Deferred(partition: Partition, isNew: Boolean) extends NonOffline /** * This broker hosts the partition, but it is in an offline log directory. @@ -247,20 +248,20 @@ class ReplicaManager(val config: KafkaConfig, } /* epoch of the controller that last changed the leader */ - @volatile var controllerEpoch: Int = KafkaController.InitialControllerEpoch - private val localBrokerId = config.brokerId - private val allPartitions = new Pool[TopicPartition, HostedPartition]( + @volatile private[server] var controllerEpoch: Int = KafkaController.InitialControllerEpoch + protected val localBrokerId = config.brokerId + protected val allPartitions = new Pool[TopicPartition, HostedPartition]( valueFactory = Some(tp => HostedPartition.Online(Partition(tp, time, configRepository, this))) ) - private val replicaStateChangeLock = new Object + protected val replicaStateChangeLock = new Object val replicaFetcherManager = createReplicaFetcherManager(metrics, time, threadNamePrefix, quotaManagers.follower) - val replicaAlterLogDirsManager = createReplicaAlterLogDirsManager(quotaManagers.alterLogDirs, brokerTopicStats) + private[server] val replicaAlterLogDirsManager = createReplicaAlterLogDirsManager(quotaManagers.alterLogDirs, brokerTopicStats) private val highWatermarkCheckPointThreadStarted = new AtomicBoolean(false) - @volatile var highWatermarkCheckpoints: Map[String, OffsetCheckpointFile] = logManager.liveLogDirs.map(dir => + @volatile private[server] var highWatermarkCheckpoints: Map[String, OffsetCheckpointFile] = logManager.liveLogDirs.map(dir => (dir.getAbsolutePath, new OffsetCheckpointFile(new File(dir, ReplicaManager.HighWatermarkFilename), logDirFailureChannel))).toMap this.logIdent = s"[ReplicaManager broker=$localBrokerId] " - private val stateChangeLogger = new StateChangeLogger(localBrokerId, inControllerContext = false, None) + protected val stateChangeLogger = new StateChangeLogger(localBrokerId, inControllerContext = false, None) private var logDirFailureHandler: LogDirFailureHandler = null @@ -275,33 +276,6 @@ class ReplicaManager(val config: KafkaConfig, } } - // Changes are initially deferred when using a Raft-based metadata quorum, and they may flip-flop to not - // being deferred and being deferred again thereafter as the broker (re)acquires/loses its lease. - // Changes are never deferred when using ZooKeeper. When true, this indicates that we should transition - // online partitions to the deferred state if we see a metadata update for that partition. - private var deferringMetadataChanges: Boolean = !config.requiresZookeeper - stateChangeLogger.debug(s"Metadata changes deferred=$deferringMetadataChanges") - - def beginMetadataChangeDeferral(): Unit = { - if (config.requiresZookeeper) { - throw new IllegalStateException("Partition metadata changes can never be deferred when using ZooKeeper") - } - replicaStateChangeLock synchronized { - deferringMetadataChanges = true - stateChangeLogger.info(s"Metadata changes are now being deferred") - } - } - - def endMetadataChangeDeferral(): Unit = { - if (config.requiresZookeeper) { - throw new IllegalStateException("Partition metadata changes can never be deferred when using ZooKeeper") - } - replicaStateChangeLock synchronized { - deferringMetadataChanges = false - stateChangeLogger.info(s"Metadata changes are no longer being deferred") - } - } - // Visible for testing private[server] val replicaSelectorOpt: Option[ReplicaSelector] = createReplicaSelector() @@ -367,7 +341,7 @@ class ReplicaManager(val config: KafkaConfig, brokerTopicStats.removeMetrics(topic) } - private def completeDelayedFetchOrProduceRequests(topicPartition: TopicPartition): Unit = { + protected def completeDelayedFetchOrProduceRequests(topicPartition: TopicPartition): Unit = { val topicPartitionOperationKey = TopicPartitionOperationKey(topicPartition) delayedProducePurgatory.checkAndComplete(topicPartitionOperationKey) delayedFetchPurgatory.checkAndComplete(topicPartitionOperationKey) @@ -399,9 +373,9 @@ class ReplicaManager(val config: KafkaConfig, } else { this.controllerEpoch = controllerEpoch - val stoppedPartitions = mutable.Map.empty[TopicPartition, StopReplicaPartitionState] + val stoppedPartitions = mutable.Map.empty[TopicPartition, Boolean] partitionStates.forKeyValue { (topicPartition, partitionState) => - val deletePartition = partitionState.deletePartition + val deletePartition = partitionState.deletePartition() getPartition(topicPartition) match { case HostedPartition.Offline => @@ -421,7 +395,7 @@ class ReplicaManager(val config: KafkaConfig, if (requestLeaderEpoch == LeaderAndIsr.EpochDuringDelete || requestLeaderEpoch == LeaderAndIsr.NoEpoch || requestLeaderEpoch > currentLeaderEpoch) { - stoppedPartitions += topicPartition -> partitionState + stoppedPartitions += topicPartition -> deletePartition // Assume that everything will go right. It is overwritten in case of an error. responseMap.put(topicPartition, Errors.NONE) } else if (requestLeaderEpoch < currentLeaderEpoch) { @@ -439,71 +413,84 @@ class ReplicaManager(val config: KafkaConfig, responseMap.put(topicPartition, Errors.FENCED_LEADER_EPOCH) } - case HostedPartition.Deferred(_) => + case _: HostedPartition.Deferred => throw new IllegalStateException("We should never be deferring partition metadata changes and stopping a replica when using ZooKeeper") case HostedPartition.None => // Delete log and corresponding folders in case replica manager doesn't hold them anymore. // This could happen when topic is being deleted while broker is down and recovers. - stoppedPartitions += topicPartition -> partitionState + stoppedPartitions += topicPartition -> deletePartition responseMap.put(topicPartition, Errors.NONE) } } - // First stop fetchers for all partitions. - val partitions = stoppedPartitions.keySet - replicaFetcherManager.removeFetcherForPartitions(partitions) - replicaAlterLogDirsManager.removeFetcherForPartitions(partitions) - - // Second remove deleted partitions from the partition map. Fetchers rely on the - // ReplicaManager to get Partition's information so they must be stopped first. - val deletedPartitions = mutable.Set.empty[TopicPartition] - stoppedPartitions.forKeyValue { (topicPartition, partitionState) => - if (partitionState.deletePartition) { - getPartition(topicPartition) match { - case hostedPartition@HostedPartition.Online(partition) => - if (allPartitions.remove(topicPartition, hostedPartition)) { - maybeRemoveTopicMetrics(topicPartition.topic) - // Logs are not deleted here. They are deleted in a single batch later on. - // This is done to avoid having to checkpoint for every deletions. - partition.delete() - } - - case _ => - } - - deletedPartitions += topicPartition - } - - // If we were the leader, we may have some operations still waiting for completion. - // We force completion to prevent them from timing out. - completeDelayedFetchOrProduceRequests(topicPartition) - } - - // Third delete the logs and checkpoint. - logManager.asyncDelete(deletedPartitions, (topicPartition, exception) => { - exception match { - case e: KafkaStorageException => + stopPartitions(stoppedPartitions).foreach { case (topicPartition, e) => + if (e.isInstanceOf[KafkaStorageException]) { stateChangeLogger.error(s"Ignoring StopReplica request (delete=true) from " + s"controller $controllerId with correlation id $correlationId " + s"epoch $controllerEpoch for partition $topicPartition as the local replica for the " + "partition is in an offline log directory") - responseMap.put(topicPartition, Errors.KAFKA_STORAGE_ERROR) - - case e => - stateChangeLogger.error(s"Ignoring StopReplica request (delete=true) from " + + } else { + stateChangeLogger.error(s"Ignoring StopReplica request (delete=true) from " + s"controller $controllerId with correlation id $correlationId " + s"epoch $controllerEpoch for partition $topicPartition due to an unexpected " + s"${e.getClass.getName} exception: ${e.getMessage}") responseMap.put(topicPartition, Errors.forException(e)) } - }) - + responseMap.put(topicPartition, Errors.forException(e)) + } (responseMap, Errors.NONE) } } } + /** + * Stop the given partitions. + * + * @param partitionsToStop A map from a topic partition to a boolean indicating + * whether the partition should be deleted. + * + * @return A map from partitions to exceptions which occurred. + * If no errors occurred, the map will be empty. + */ + protected def stopPartitions(partitionsToStop: Map[TopicPartition, Boolean]): Map[TopicPartition, Throwable] = { + // First stop fetchers for all partitions. + val partitions = partitionsToStop.keySet + replicaFetcherManager.removeFetcherForPartitions(partitions) + replicaAlterLogDirsManager.removeFetcherForPartitions(partitions) + + // Second remove deleted partitions from the partition map. Fetchers rely on the + // ReplicaManager to get Partition's information so they must be stopped first. + val partitionsToDelete = mutable.Set.empty[TopicPartition] + partitionsToStop.forKeyValue { (topicPartition, shouldDelete) => + if (shouldDelete) { + getPartition(topicPartition) match { + case hostedPartition: NonOffline => + if (allPartitions.remove(topicPartition, hostedPartition)) { + maybeRemoveTopicMetrics(topicPartition.topic) + // Logs are not deleted here. They are deleted in a single batch later on. + // This is done to avoid having to checkpoint for every deletions. + hostedPartition.partition.delete() + } + + case _ => + } + partitionsToDelete += topicPartition + } + // If we were the leader, we may have some operations still waiting for completion. + // We force completion to prevent them from timing out. + completeDelayedFetchOrProduceRequests(topicPartition) + } + + // Third delete the logs and checkpoint. + val errorMap = new mutable.HashMap[TopicPartition, Throwable]() + if (partitionsToDelete.nonEmpty) { + // Delete the logs and checkpoint. + logManager.asyncDelete(partitionsToDelete, (tp, e) => errorMap.put(tp, e)) + } + errorMap + } + def getPartition(topicPartition: TopicPartition): HostedPartition = { Option(allPartitions.get(topicPartition)).getOrElse(HostedPartition.None) } @@ -569,7 +556,7 @@ class ReplicaManager(val config: KafkaConfig, // the local replica has been deleted. Left(Errors.NOT_LEADER_OR_FOLLOWER) - case HostedPartition.Deferred(_) => + case _: HostedPartition.Deferred => // The topic exists, but this broker is deferring metadata changes for it, so we return NOT_LEADER_OR_FOLLOWER // which forces clients to refresh metadata. Left(Errors.NOT_LEADER_OR_FOLLOWER) @@ -766,7 +753,7 @@ class ReplicaManager(val config: KafkaConfig, case HostedPartition.Offline => throw new KafkaStorageException(s"Partition $topicPartition is offline") - case HostedPartition.Deferred(_) => + case _: HostedPartition.Deferred => throw new IllegalStateException(s"Partition $topicPartition is deferred") case HostedPartition.None => // Do nothing @@ -1365,7 +1352,7 @@ class ReplicaManager(val config: KafkaConfig, responseMap.put(topicPartition, Errors.KAFKA_STORAGE_ERROR) None - case HostedPartition.Deferred(_) => + case _: HostedPartition.Deferred => throw new IllegalStateException("We should never be deferring partition metadata changes and becoming a leader or follower when using ZooKeeper") case HostedPartition.Online(partition) => @@ -1426,18 +1413,8 @@ class ReplicaManager(val config: KafkaConfig, else Set.empty[Partition] - /* - * KAFKA-8392 - * For topic partitions of which the broker is no longer a leader, delete metrics related to - * those topics. Note that this means the broker stops being either a replica or a leader of - * partitions of said topics - */ - val leaderTopicSet = leaderPartitionsIterator.map(_.topic).toSet val followerTopicSet = partitionsBecomeFollower.map(_.topic).toSet - followerTopicSet.diff(leaderTopicSet).foreach(brokerTopicStats.removeOldLeaderMetrics) - - // remove metrics for brokers which are not followers of a topic - leaderTopicSet.diff(followerTopicSet).foreach(brokerTopicStats.removeOldFollowerMetrics) + updateLeaderAndFollowerMetrics(followerTopicSet) leaderAndIsrRequest.partitionStates.forEach { partitionState => val topicPartition = new TopicPartition(partitionState.topicName, partitionState.partitionIndex) @@ -1520,7 +1497,21 @@ class ReplicaManager(val config: KafkaConfig, } } - private def maybeAddLogDirFetchers(partitions: Set[Partition], + /** + * KAFKA-8392 + * For topic partitions of which the broker is no longer a leader, delete metrics related to + * those topics. Note that this means the broker stops being either a replica or a leader of + * partitions of said topics + */ + protected def updateLeaderAndFollowerMetrics(newFollowerTopics: Set[String]): Unit = { + val leaderTopicSet = leaderPartitionsIterator.map(_.topic).toSet + newFollowerTopics.diff(leaderTopicSet).foreach(brokerTopicStats.removeOldLeaderMetrics) + + // remove metrics for brokers which are not followers of a topic + leaderTopicSet.diff(newFollowerTopics).foreach(brokerTopicStats.removeOldFollowerMetrics) + } + + protected def maybeAddLogDirFetchers(partitions: Set[Partition], offsetCheckpoints: OffsetCheckpoints): Unit = { val futureReplicasAndInitialOffset = new mutable.HashMap[TopicPartition, InitialFetchState] for (partition <- partitions) { @@ -1753,7 +1744,7 @@ class ReplicaManager(val config: KafkaConfig, * diverging epoch is returned in the response, avoiding the need for a separate * OffsetForLeaderEpoch request. */ - private def initialFetchOffset(log: Log): Long = { + protected def initialFetchOffset(log: Log): Long = { if (ApiVersion.isTruncationOnFetchSupported(config.interBrokerProtocolVersion) && log.latestEpoch.nonEmpty) log.logEndOffset else @@ -1841,7 +1832,6 @@ class ReplicaManager(val config: KafkaConfig, } } - // Used only by test def markPartitionOffline(tp: TopicPartition): Unit = replicaStateChangeLock synchronized { allPartitions.put(tp, HostedPartition.Offline) Partition.removeMetrics(tp) @@ -1964,7 +1954,7 @@ class ReplicaManager(val config: KafkaConfig, .setPartition(offsetForLeaderPartition.partition) .setErrorCode(Errors.NOT_LEADER_OR_FOLLOWER.code) - case HostedPartition.Deferred(_) => + case _: HostedPartition.Deferred => new EpochEndOffset() .setPartition(offsetForLeaderPartition.partition) .setErrorCode(Errors.NOT_LEADER_OR_FOLLOWER.code) diff --git a/core/src/main/scala/kafka/server/metadata/MetadataImage.scala b/core/src/main/scala/kafka/server/metadata/MetadataImage.scala index 199306527989c..f4e5831e8ab1a 100755 --- a/core/src/main/scala/kafka/server/metadata/MetadataImage.scala +++ b/core/src/main/scala/kafka/server/metadata/MetadataImage.scala @@ -86,12 +86,15 @@ case class MetadataImageBuilder(brokerId: Int, } else { _partitionsBuilder.build() } - val nextBrokers = if (_brokersBuilder == null) { + MetadataImage(nextPartitions, _controllerId, brokers()) + } + + def brokers(): MetadataBrokers = { + if (_brokersBuilder == null) { prevImage.brokers } else { _brokersBuilder.build() } - MetadataImage(nextPartitions, _controllerId, nextBrokers) } } diff --git a/core/src/test/scala/unit/kafka/server/RaftReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/RaftReplicaManagerTest.scala new file mode 100644 index 0000000000000..d3d6471712663 --- /dev/null +++ b/core/src/test/scala/unit/kafka/server/RaftReplicaManagerTest.scala @@ -0,0 +1,238 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package kafka.server + +import java.io.File +import java.util +import java.util.concurrent.atomic.AtomicBoolean + +import kafka.cluster.Partition +import kafka.server.QuotaFactory.QuotaManagers +import kafka.server.metadata.{CachedConfigRepository, MetadataBroker, MetadataBrokers, MetadataImage, MetadataImageBuilder, MetadataPartition, RaftMetadataCache} +import kafka.utils.{MockScheduler, MockTime, TestUtils} +import org.apache.kafka.common.{TopicPartition, Uuid} +import org.apache.kafka.common.metadata.PartitionRecord +import org.apache.kafka.common.metrics.Metrics +import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows, assertTrue} +import org.junit.jupiter.api.{AfterEach, BeforeEach, Test} +import org.mockito.{ArgumentCaptor, ArgumentMatchers} +import org.mockito.ArgumentMatchers.any +import org.mockito.Mockito.{mock, never, verify, when} +import org.slf4j.Logger + +import scala.collection.{Set, mutable} + +trait LeadershipChangeHandler { + def onLeadershipChange(updatedLeaders: Iterable[Partition], updatedFollowers: Iterable[Partition]): Unit +} + +class RaftReplicaManagerTest { + private var alterIsrManager: AlterIsrManager = _ + private var config: KafkaConfig = _ + private val configRepository = new CachedConfigRepository() + private val metrics = new Metrics + private var quotaManager: QuotaManagers = _ + private val time = new MockTime + private var mockDelegate: RaftReplicaChangeDelegate = _ + private var imageBuilder: MetadataImageBuilder = _ + private val brokerId0 = 0 + private val metadataBroker0 = new MetadataBroker(brokerId0, null, Map.empty, false) + private val brokerId1 = 1 + private val metadataBroker1 = new MetadataBroker(brokerId1, null, Map.empty, false) + private val topicName = "topicName" + private val topicId = Uuid.randomUuid() + private val partitionId0 = 0 + private val partitionId1 = 1 + private val topicPartition0 = new TopicPartition(topicName, partitionId0) + private val topicPartition1 = new TopicPartition(topicName, partitionId1) + private val topicPartitionRecord0 = new PartitionRecord() + .setPartitionId(partitionId0) + .setTopicId(topicId) + .setReplicas(util.Arrays.asList(brokerId0, brokerId1)) + .setLeader(brokerId0) + .setLeaderEpoch(0) + private val topicPartitionRecord1 = new PartitionRecord() + .setPartitionId(partitionId1) + .setTopicId(topicId) + .setReplicas(util.Arrays.asList(brokerId0, brokerId1)) + .setLeader(brokerId1) + .setLeaderEpoch(0) + private val offset1 = 1L + private val metadataPartition0 = MetadataPartition(topicName, topicPartitionRecord0) + private val metadataPartition1 = MetadataPartition(topicName, topicPartitionRecord1) + private var onLeadershipChangeHandler: LeadershipChangeHandler = _ + private var onLeadershipChange: (Iterable[Partition], Iterable[Partition]) => Unit = _ + private var metadataCache: RaftMetadataCache = _ + + @BeforeEach + def setUp(): Unit = { + alterIsrManager = mock(classOf[AlterIsrManager]) + config = KafkaConfig.fromProps({ + val nodeId = brokerId0 + val props = TestUtils.createBrokerConfig(nodeId, "") + props.put(KafkaConfig.ProcessRolesProp, "broker") + props.put(KafkaConfig.NodeIdProp, nodeId.toString) + props + }) + metadataCache = new RaftMetadataCache(config.brokerId) + quotaManager = QuotaFactory.instantiate(config, metrics, time, "") + mockDelegate = mock(classOf[RaftReplicaChangeDelegate]) + imageBuilder = MetadataImageBuilder(brokerId0, mock(classOf[Logger]), new MetadataImage()) + onLeadershipChangeHandler = mock(classOf[LeadershipChangeHandler]) + onLeadershipChange = onLeadershipChangeHandler.onLeadershipChange _ + } + + @AfterEach + def tearDown(): Unit = { + TestUtils.clearYammerMetrics() + Option(quotaManager).foreach(_.shutdown()) + metrics.close() + } + + def createRaftReplicaManager(): RaftReplicaManager = { + val mockLogMgr = TestUtils.createLogManager(config.logDirs.map(new File(_))) + new RaftReplicaManager(config, metrics, time, new MockScheduler(time), mockLogMgr, + new AtomicBoolean(false), quotaManager, new BrokerTopicStats, + metadataCache, new LogDirFailureChannel(config.logDirs.size), alterIsrManager, + configRepository, None) + } + + @Test + def testRejectsZkConfig(): Unit = { + assertThrows(classOf[IllegalStateException], () => { + val zkConfig = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "")) + val mockLogMgr = TestUtils.createLogManager(zkConfig.logDirs.map(new File(_))) + new RaftReplicaManager(zkConfig, metrics, time, new MockScheduler(time), mockLogMgr, + new AtomicBoolean(false), quotaManager, new BrokerTopicStats, + metadataCache, new LogDirFailureChannel(config.logDirs.size), alterIsrManager, + configRepository) + }) + } + + @Test + def testDefersChangesImmediatelyThenAppliesChanges(): Unit = { + val rrm = createRaftReplicaManager() + rrm.delegate = mockDelegate + val partition0 = Partition(topicPartition0, time, configRepository, rrm) + val partition1 = Partition(topicPartition1, time, configRepository, rrm) + + processTopicPartitionMetadata(rrm) + // verify changes would have been deferred + val partitionsNewMapCaptor: ArgumentCaptor[mutable.Map[Partition, Boolean]] = + ArgumentCaptor.forClass(classOf[mutable.Map[Partition, Boolean]]) + verify(mockDelegate).makeDeferred(partitionsNewMapCaptor.capture(), ArgumentMatchers.eq(offset1)) + val partitionsDeferredMap = partitionsNewMapCaptor.getValue + assertEquals(Map(partition0 -> true, partition1 -> true), partitionsDeferredMap) + verify(mockDelegate, never()).makeFollowers(any(), any(), any(), any(), any()) + + // now mark those topic partitions as being deferred so we can later apply the changes + rrm.markPartitionDeferred(partition0, isNew = true) + rrm.markPartitionDeferred(partition1, isNew = true) + + // apply the changes + // define some return values to avoid NPE + when(mockDelegate.makeLeaders(any(), any(), any(), any())).thenReturn(Set(partition0)) + when(mockDelegate.makeFollowers(any(), any(), any(), any(), any())).thenReturn(Set(partition1)) + rrm.endMetadataChangeDeferral(onLeadershipChange) + // verify that the deferred changes would have been applied + + // leaders... + val leaderPartitionStates = verifyMakeLeaders(mutable.Set(), None) + assertEquals(Map(partition0 -> metadataPartition0), leaderPartitionStates) + + // followers... + val followerPartitionStates = verifyMakeFollowers(mutable.Set(), Set(brokerId0, brokerId1), None) + assertEquals(Map(partition1 -> metadataPartition1), followerPartitionStates) + + // leadership change callbacks + verifyLeadershipChangeCallbacks(List(partition0), List(partition1)) + } + + @Test + def testAppliesChangesWhenNotDeferring(): Unit = { + val rrm = createRaftReplicaManager() + rrm.delegate = mockDelegate + val partition0 = Partition(topicPartition0, time, configRepository, rrm) + val partition1 = Partition(topicPartition1, time, configRepository, rrm) + rrm.endMetadataChangeDeferral(onLeadershipChange) + + // define some return values to avoid NPE + when(mockDelegate.makeLeaders(any(), any(), any(), ArgumentMatchers.eq(Some(offset1)))).thenReturn(Set(partition0)) + when(mockDelegate.makeFollowers(any(), any(), any(), any(), ArgumentMatchers.eq(Some(offset1)))).thenReturn(Set(partition1)) + // process the changes + processTopicPartitionMetadata(rrm) + // verify that the changes would have been applied + + // leaders... + val leaderPartitionStates = verifyMakeLeaders(mutable.Set(), Some(offset1)) + assertEquals(Map(partition0 -> metadataPartition0), leaderPartitionStates) + + // followers... + val followerPartitionStates = verifyMakeFollowers(mutable.Set(), Set(brokerId0, brokerId1), Some(offset1)) + assertEquals(Map(partition1 -> metadataPartition1), followerPartitionStates) + + // leadership change callbacks + verifyLeadershipChangeCallbacks(List(partition0), List(partition1)) + } + + private def verifyMakeLeaders(expectedPrevPartitionsAlreadyExisting: Set[MetadataPartition], + expectedMetadataOffset: Option[Long]): mutable.Map[Partition, MetadataPartition] = { + val leaderPartitionStatesCaptor: ArgumentCaptor[mutable.Map[Partition, MetadataPartition]] = + ArgumentCaptor.forClass(classOf[mutable.Map[Partition, MetadataPartition]]) + verify(mockDelegate).makeLeaders(ArgumentMatchers.eq(expectedPrevPartitionsAlreadyExisting), + leaderPartitionStatesCaptor.capture(), any(), ArgumentMatchers.eq(expectedMetadataOffset)) + leaderPartitionStatesCaptor.getValue + } + + private def verifyMakeFollowers(expectedPrevPartitionsAlreadyExisting: Set[MetadataPartition], + expectedBrokers: Set[Int], + expectedMetadataOffset: Option[Long]): mutable.Map[Partition, MetadataPartition] = { + val followerPartitionStatesCaptor: ArgumentCaptor[mutable.Map[Partition, MetadataPartition]] = + ArgumentCaptor.forClass(classOf[mutable.Map[Partition, MetadataPartition]]) + val brokersCaptor: ArgumentCaptor[MetadataBrokers] = ArgumentCaptor.forClass(classOf[MetadataBrokers]) + verify(mockDelegate).makeFollowers(ArgumentMatchers.eq(expectedPrevPartitionsAlreadyExisting), brokersCaptor.capture(), + followerPartitionStatesCaptor.capture(), any(), ArgumentMatchers.eq(expectedMetadataOffset)) + val brokers = brokersCaptor.getValue + assertEquals(expectedBrokers.size, brokers.size()) + expectedBrokers.foreach(brokerId => assertTrue(brokers.aliveBroker(brokerId).isDefined)) + followerPartitionStatesCaptor.getValue + } + + private def verifyLeadershipChangeCallbacks(expectedUpdatedLeaders: List[Partition], expectedUpdatedFollowers: List[Partition]): Unit = { + val updatedLeadersCaptor: ArgumentCaptor[Iterable[Partition]] = ArgumentCaptor.forClass(classOf[Iterable[Partition]]) + val updatedFollowersCaptor: ArgumentCaptor[Iterable[Partition]] = ArgumentCaptor.forClass(classOf[Iterable[Partition]]) + verify(onLeadershipChangeHandler).onLeadershipChange(updatedLeadersCaptor.capture(), updatedFollowersCaptor.capture()) + assertEquals(expectedUpdatedLeaders, updatedLeadersCaptor.getValue.toList) + assertEquals(expectedUpdatedFollowers, updatedFollowersCaptor.getValue.toList) + } + + private def processTopicPartitionMetadata(raftReplicaManager: RaftReplicaManager): Unit = { + // create brokers + imageBuilder.brokersBuilder().add(metadataBroker0) + imageBuilder.brokersBuilder().add(metadataBroker1) + // create topic + imageBuilder.partitionsBuilder().addUuidMapping(topicName, topicId) + // create deferred partitions + imageBuilder.partitionsBuilder().set(metadataPartition0) + imageBuilder.partitionsBuilder().set(metadataPartition1) + // apply the changes to metadata cache + metadataCache.image(imageBuilder.build()) + // apply the changes to replica manager + raftReplicaManager.handleMetadataRecords(imageBuilder, offset1, onLeadershipChange) + } +}