diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala index 9d33381815136..0118c0bef38ea 100755 --- a/core/src/main/scala/kafka/cluster/Partition.scala +++ b/core/src/main/scala/kafka/cluster/Partition.scala @@ -308,27 +308,27 @@ class Partition(val topicPartition: TopicPartition, s"different from the requested log dir $logDir") false case None => - createLogIfNotExists(isNew = false, isFutureReplica = true, highWatermarkCheckpoints) + createLogIfNotExists(isNew = false, isFutureReplica = true, highWatermarkCheckpoints, topicId) true } } } } - def createLogIfNotExists(isNew: Boolean, isFutureReplica: Boolean, offsetCheckpoints: OffsetCheckpoints): Unit = { + def createLogIfNotExists(isNew: Boolean, isFutureReplica: Boolean, offsetCheckpoints: OffsetCheckpoints, topicId: Option[Uuid]): Unit = { isFutureReplica match { case true if futureLog.isEmpty => - val log = createLog(isNew, isFutureReplica, offsetCheckpoints) + val log = createLog(isNew, isFutureReplica, offsetCheckpoints, topicId) this.futureLog = Option(log) case false if log.isEmpty => - val log = createLog(isNew, isFutureReplica, offsetCheckpoints) + val log = createLog(isNew, isFutureReplica, offsetCheckpoints, topicId) this.log = Option(log) case _ => trace(s"${if (isFutureReplica) "Future Log" else "Log"} already exists.") } } // Visible for testing - private[cluster] def createLog(isNew: Boolean, isFutureReplica: Boolean, offsetCheckpoints: OffsetCheckpoints): Log = { + private[cluster] def createLog(isNew: Boolean, isFutureReplica: Boolean, offsetCheckpoints: OffsetCheckpoints, topicId: Option[Uuid]): Log = { def updateHighWatermark(log: Log) = { val checkpointHighWatermark = offsetCheckpoints.fetch(log.parentDir, topicPartition).getOrElse { info(s"No checkpointed highwatermark is found for partition $topicPartition") @@ -341,7 +341,7 @@ class Partition(val topicPartition: TopicPartition, logManager.initializingLog(topicPartition) var maybeLog: Option[Log] = None try { - val log = logManager.getOrCreateLog(topicPartition, isNew, isFutureReplica) + val log = logManager.getOrCreateLog(topicPartition, isNew, isFutureReplica, topicId) maybeLog = Some(log) updateHighWatermark(log) log @@ -425,41 +425,11 @@ class Partition(val topicPartition: TopicPartition, } /** - * Checks if the topic ID provided in the request is consistent with the topic ID in the log. - * If a valid topic ID is provided, and the log exists but has no ID set, set the log ID to be the request ID. - * - * @param requestTopicId the topic ID from the request - * @return true if the request topic id is consistent, false otherwise + * @return the topic ID for the log or None if the log or the topic ID does not exist. */ - def checkOrSetTopicId(requestTopicId: Uuid): Boolean = { - // If the request had an invalid topic ID, then we assume that topic IDs are not supported. - // The topic ID was not inconsistent, so return true. - // If the log is empty, then we can not say that topic ID is inconsistent, so return true. - if (requestTopicId == null || requestTopicId == Uuid.ZERO_UUID) - true - else { - log match { - case None => true - case Some(log) => { - // Check if topic ID is in memory, if not, it must be new to the broker and does not have a metadata file. - // This is because if the broker previously wrote it to file, it would be recovered on restart after failure. - // Topic ID is consistent since we are just setting it here. - if (log.topicId == Uuid.ZERO_UUID) { - log.partitionMetadataFile.write(requestTopicId) - log.topicId = requestTopicId - true - } else if (log.topicId != requestTopicId) { - stateChangeLogger.error(s"Topic Id in memory: ${log.topicId} does not" + - s" match the topic Id for partition $topicPartition provided in the request: " + - s"$requestTopicId.") - false - } else { - // topic ID in log exists and matches request topic ID - true - } - } - } - } + def topicId: Option[Uuid] = { + val log = this.log.orElse(logManager.getLog(topicPartition)) + log.flatMap(_.topicId) } // remoteReplicas will be called in the hot path, and must be inexpensive @@ -540,7 +510,8 @@ class Partition(val topicPartition: TopicPartition, * If the leader replica id does not change, return false to indicate the replica manager. */ def makeLeader(partitionState: LeaderAndIsrPartitionState, - highWatermarkCheckpoints: OffsetCheckpoints): Boolean = { + highWatermarkCheckpoints: OffsetCheckpoints, + topicId: Option[Uuid]): Boolean = { val (leaderHWIncremented, isNewLeader) = inWriteLock(leaderIsrUpdateLock) { // record the epoch of the controller that made the leadership decision. This is useful while updating the isr // to maintain the decision maker controller's epoch in the zookeeper path @@ -557,7 +528,7 @@ class Partition(val topicPartition: TopicPartition, removingReplicas = removingReplicas ) try { - createLogIfNotExists(partitionState.isNew, isFutureReplica = false, highWatermarkCheckpoints) + createLogIfNotExists(partitionState.isNew, isFutureReplica = false, highWatermarkCheckpoints, topicId) } catch { case e: ZooKeeperClientException => stateChangeLogger.error(s"A ZooKeeper client exception has occurred and makeLeader will be skipping the " + @@ -624,7 +595,8 @@ class Partition(val topicPartition: TopicPartition, * replica manager that state is already correct and the become-follower steps can be skipped */ def makeFollower(partitionState: LeaderAndIsrPartitionState, - highWatermarkCheckpoints: OffsetCheckpoints): Boolean = { + highWatermarkCheckpoints: OffsetCheckpoints, + topicId: Option[Uuid]): Boolean = { inWriteLock(leaderIsrUpdateLock) { val newLeaderBrokerId = partitionState.leader val oldLeaderEpoch = leaderEpoch @@ -639,7 +611,7 @@ class Partition(val topicPartition: TopicPartition, removingReplicas = partitionState.removingReplicas.asScala.map(_.toInt) ) try { - createLogIfNotExists(partitionState.isNew, isFutureReplica = false, highWatermarkCheckpoints) + createLogIfNotExists(partitionState.isNew, isFutureReplica = false, highWatermarkCheckpoints, topicId) } catch { case e: ZooKeeperClientException => stateChangeLogger.error(s"A ZooKeeper client exception has occurred. makeFollower will be skipping the " + diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala index 0ad82f46b9d4b..1c7ec6e338b57 100644 --- a/core/src/main/scala/kafka/log/Log.scala +++ b/core/src/main/scala/kafka/log/Log.scala @@ -241,12 +241,16 @@ case object SnapshotGenerated extends LogStartOffsetIncrementReason { * @param producerIdExpirationCheckIntervalMs How often to check for producer ids which need to be expired * @param hadCleanShutdown boolean flag to indicate if the Log had a clean/graceful shutdown last time. true means * clean shutdown whereas false means a crash. + * @param topicId optional Uuid to specify the topic ID for the topic if it exists. Should only be specified when + * first creating the log through Partition.makeLeader or Partition.makeFollower. When reloading a log, + * this field will be populated by reading the topic ID value from partition.metadata if it exists. * @param keepPartitionMetadataFile boolean flag to indicate whether the partition.metadata file should be kept in the - * log directory. A partition.metadata file is only created when the controller's - * inter-broker protocol version is at least 2.8. This file will persist the topic ID on - * the broker. If inter-broker protocol is downgraded below 2.8, a topic ID may be lost - * and a new ID generated upon re-upgrade. If the inter-broker protocol version is below - * 2.8, partition.metadata will be deleted to avoid ID conflicts upon re-upgrade. + * log directory. A partition.metadata file is only created when the raft controller is used + * or the ZK controller's inter-broker protocol version is at least 2.8. + * This file will persist the topic ID on the broker. If inter-broker protocol for a ZK controller + * is downgraded below 2.8, a topic ID may be lost and a new ID generated upon re-upgrade. + * If the inter-broker protocol version on a ZK cluster is below 2.8, partition.metadata + * will be deleted to avoid ID conflicts upon re-upgrade. */ @threadsafe class Log(@volatile private var _dir: File, @@ -262,7 +266,8 @@ class Log(@volatile private var _dir: File, val producerStateManager: ProducerStateManager, logDirFailureChannel: LogDirFailureChannel, private val hadCleanShutdown: Boolean = true, - val keepPartitionMetadataFile: Boolean = true) extends Logging with KafkaMetricsGroup { + @volatile var topicId: Option[Uuid], + val keepPartitionMetadataFile: Boolean) extends Logging with KafkaMetricsGroup { import kafka.log.Log._ @@ -315,8 +320,6 @@ class Log(@volatile private var _dir: File, @volatile var partitionMetadataFile : PartitionMetadataFile = null - @volatile var topicId : Uuid = Uuid.ZERO_UUID - locally { // create the log directory if it doesn't exist Files.createDirectories(dir.toPath) @@ -349,11 +352,21 @@ class Log(@volatile private var _dir: File, // Delete partition metadata file if the version does not support topic IDs. // Recover topic ID if present and topic IDs are supported + // If we were provided a topic ID when creating the log, partition metadata files are supported, and one does not yet exist + // write to the partition metadata file. + // Ensure we do not try to assign a provided topicId that is inconsistent with the ID on file. if (partitionMetadataFile.exists()) { if (!keepPartitionMetadataFile) partitionMetadataFile.delete() - else - topicId = partitionMetadataFile.read().topicId + else { + val fileTopicId = partitionMetadataFile.read().topicId + if (topicId.isDefined && !topicId.contains(fileTopicId)) + throw new InconsistentTopicIdException(s"Tried to assign topic ID $topicId to log for topic partition $topicPartition," + + s"but log already contained topic ID $fileTopicId") + topicId = Some(fileTopicId) + } + } else if (keepPartitionMetadataFile) { + topicId.foreach(partitionMetadataFile.write) } } @@ -576,6 +589,12 @@ class Log(@volatile private var _dir: File, partitionMetadataFile = new PartitionMetadataFile(partitionMetadata, logDirFailureChannel) } + /** Only used for ZK clusters when we update and start using topic IDs on existing topics */ + def assignTopicId(topicId: Uuid): Unit = { + partitionMetadataFile.write(topicId) + this.topicId = Some(topicId) + } + private def initializeLeaderEpochCache(): Unit = lock synchronized { val leaderEpochFile = LeaderEpochCheckpointFile.newFile(dir) @@ -2600,11 +2619,12 @@ object Log { producerIdExpirationCheckIntervalMs: Int, logDirFailureChannel: LogDirFailureChannel, lastShutdownClean: Boolean = true, - keepPartitionMetadataFile: Boolean = true): Log = { + topicId: Option[Uuid], + keepPartitionMetadataFile: Boolean): Log = { val topicPartition = Log.parseTopicPartitionName(dir) val producerStateManager = new ProducerStateManager(topicPartition, dir, maxProducerIdExpirationMs) new Log(dir, config, logStartOffset, recoveryPoint, scheduler, brokerTopicStats, time, maxProducerIdExpirationMs, - producerIdExpirationCheckIntervalMs, topicPartition, producerStateManager, logDirFailureChannel, lastShutdownClean, keepPartitionMetadataFile) + producerIdExpirationCheckIntervalMs, topicPartition, producerStateManager, logDirFailureChannel, lastShutdownClean, topicId, keepPartitionMetadataFile) } /** diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala index 432cf4a185b2f..117d2138d9a20 100755 --- a/core/src/main/scala/kafka/log/LogManager.scala +++ b/core/src/main/scala/kafka/log/LogManager.scala @@ -27,9 +27,9 @@ import kafka.server.checkpoints.OffsetCheckpointFile import kafka.server.metadata.ConfigRepository import kafka.server._ import kafka.utils._ -import org.apache.kafka.common.{KafkaException, TopicPartition} +import org.apache.kafka.common.{KafkaException, TopicPartition, Uuid} import org.apache.kafka.common.utils.Time -import org.apache.kafka.common.errors.{KafkaStorageException, LogDirNotFoundException} +import org.apache.kafka.common.errors.{InconsistentTopicIdException, KafkaStorageException, LogDirNotFoundException} import scala.jdk.CollectionConverters._ import scala.collection._ @@ -270,6 +270,7 @@ class LogManager(logDirs: Seq[File], brokerTopicStats = brokerTopicStats, logDirFailureChannel = logDirFailureChannel, lastShutdownClean = hadCleanShutdown, + topicId = None, keepPartitionMetadataFile = keepPartitionMetadataFile) if (logDir.getName.endsWith(Log.DeleteDirSuffix)) { @@ -775,11 +776,13 @@ class LogManager(logDirs: Seq[File], * @param topicPartition The partition whose log needs to be returned or created * @param isNew Whether the replica should have existed on the broker or not * @param isFuture True if the future log of the specified partition should be returned or created + * @param topicId The topic ID of the partition's topic * @throws KafkaStorageException if isNew=false, log is not found in the cache and there is offline log directory on the broker + * @throws InconsistentTopicIdException if the topic ID in the log does not match the topic ID provided */ - def getOrCreateLog(topicPartition: TopicPartition, isNew: Boolean = false, isFuture: Boolean = false): Log = { + def getOrCreateLog(topicPartition: TopicPartition, isNew: Boolean = false, isFuture: Boolean = false, topicId: Option[Uuid]): Log = { logCreationOrDeletionLock synchronized { - getLog(topicPartition, isFuture).getOrElse { + val log = getLog(topicPartition, isFuture).getOrElse { // create the log if it has not already been created in another thread if (!isNew && offlineLogDirs.nonEmpty) throw new KafkaStorageException(s"Can not create log for $topicPartition because log directories ${offlineLogDirs.mkString(",")} are offline") @@ -826,6 +829,7 @@ class LogManager(logDirs: Seq[File], time = time, brokerTopicStats = brokerTopicStats, logDirFailureChannel = logDirFailureChannel, + topicId = topicId, keepPartitionMetadataFile = keepPartitionMetadataFile) if (isFuture) @@ -839,6 +843,20 @@ class LogManager(logDirs: Seq[File], log } + // When running a ZK controller, we may get a log that does not have a topic ID. Assign it here. + if (log.topicId.isEmpty) { + topicId.foreach(log.assignTopicId) + } + + // Ensure topic IDs are consistent + topicId.foreach { topicId => + log.topicId.foreach { logTopicId => + if (topicId != logTopicId) + throw new InconsistentTopicIdException(s"Tried to assign topic ID $topicId to log for topic partition $topicPartition," + + s"but log already contained topic ID $logTopicId") + } + } + log } } diff --git a/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala b/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala index 95de4fe3923b7..b2c16d0eb2190 100644 --- a/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala +++ b/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala @@ -346,6 +346,7 @@ object KafkaMetadataLog { producerIdExpirationCheckIntervalMs = Int.MaxValue, logDirFailureChannel = new LogDirFailureChannel(5), lastShutdownClean = false, + topicId = None, keepPartitionMetadataFile = false ) diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index 39a93aa47acf5..c9d97912a9360 100755 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -1892,8 +1892,9 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO } } + // Topic IDs are used with all self-managed quorum clusters and ZK cluster with IBP greater than or equal to 2.8 def usesTopicId: Boolean = - interBrokerProtocolVersion >= KAFKA_2_8_IV0 + usesSelfManagedQuorum || interBrokerProtocolVersion >= KAFKA_2_8_IV0 validateValues() diff --git a/core/src/main/scala/kafka/server/RaftReplicaChangeDelegate.scala b/core/src/main/scala/kafka/server/RaftReplicaChangeDelegate.scala index 1bf21e074069d..beb4eb52c3d4a 100644 --- a/core/src/main/scala/kafka/server/RaftReplicaChangeDelegate.scala +++ b/core/src/main/scala/kafka/server/RaftReplicaChangeDelegate.scala @@ -22,7 +22,7 @@ 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.{TopicPartition, Uuid} import org.apache.kafka.common.errors.KafkaStorageException import scala.collection.{Map, Set, mutable} @@ -72,7 +72,8 @@ class RaftReplicaChangeDelegate(helper: RaftReplicaChangeDelegateHelper) { def makeLeaders(prevPartitionsAlreadyExisting: Set[MetadataPartition], partitionStates: Map[Partition, MetadataPartition], highWatermarkCheckpoints: OffsetCheckpoints, - metadataOffset: Option[Long]): Set[Partition] = { + metadataOffset: Option[Long], + topicIds: String => Option[Uuid]): Set[Partition] = { val partitionsMadeLeaders = mutable.Set[Partition]() val traceLoggingEnabled = helper.stateChangeLogger.isTraceEnabled val deferredBatches = metadataOffset.isEmpty @@ -94,7 +95,7 @@ class RaftReplicaChangeDelegate(helper: RaftReplicaChangeDelegateHelper) { try { val isrState = state.toLeaderAndIsrPartitionState( !prevPartitionsAlreadyExisting(state)) - if (partition.makeLeader(isrState, highWatermarkCheckpoints)) { + if (partition.makeLeader(isrState, highWatermarkCheckpoints, topicIds(partition.topic))) { partitionsMadeLeaders += partition if (traceLoggingEnabled) { helper.stateChangeLogger.trace(s"$partitionLogMsgPrefix: completed the become-leader state change.") @@ -125,7 +126,8 @@ class RaftReplicaChangeDelegate(helper: RaftReplicaChangeDelegateHelper) { currentBrokers: MetadataBrokers, partitionStates: Map[Partition, MetadataPartition], highWatermarkCheckpoints: OffsetCheckpoints, - metadataOffset: Option[Long]): Set[Partition] = { + metadataOffset: Option[Long], + topicIds: String => Option[Uuid]): Set[Partition] = { val traceLoggingEnabled = helper.stateChangeLogger.isTraceEnabled val deferredBatches = metadataOffset.isEmpty val topLevelLogPrefix = if (deferredBatches) @@ -164,10 +166,10 @@ class RaftReplicaChangeDelegate(helper: RaftReplicaChangeDelegateHelper) { 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) + partition.createLogIfNotExists(isNew, isFutureReplica = false, highWatermarkCheckpoints, topicIds(partition.topic)) } else { val isrState = state.toLeaderAndIsrPartitionState(isNew) - if (partition.makeFollower(isrState, highWatermarkCheckpoints)) { + if (partition.makeFollower(isrState, highWatermarkCheckpoints, topicIds(partition.topic))) { partitionsMadeFollower += partition if (traceLoggingEnabled) { helper.stateChangeLogger.trace(s"$partitionLogMsgPrefix: completed the " + diff --git a/core/src/main/scala/kafka/server/RaftReplicaManager.scala b/core/src/main/scala/kafka/server/RaftReplicaManager.scala index 37dc3e1a23ec3..30960deed0369 100644 --- a/core/src/main/scala/kafka/server/RaftReplicaManager.scala +++ b/core/src/main/scala/kafka/server/RaftReplicaManager.scala @@ -26,8 +26,8 @@ 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.{TopicPartition, Uuid} +import org.apache.kafka.common.errors.{InconsistentTopicIdException, KafkaStorageException} import org.apache.kafka.common.metrics.Metrics import org.apache.kafka.common.utils.Time @@ -155,11 +155,10 @@ class RaftReplicaManager(config: KafkaConfig, partitionsAlreadyExisting += state } } - if (leaderPartitionStates.nonEmpty) - partitionsMadeLeader = delegate.makeLeaders(partitionsAlreadyExisting, leaderPartitionStates, highWatermarkCheckpoints, None) + partitionsMadeLeader = delegate.makeLeaders(partitionsAlreadyExisting, leaderPartitionStates, highWatermarkCheckpoints, None, metadataImage.topicNameToId) if (followerPartitionStates.nonEmpty) - partitionsMadeFollower = delegate.makeFollowers(partitionsAlreadyExisting, brokers, followerPartitionStates, highWatermarkCheckpoints, None) + partitionsMadeFollower = delegate.makeFollowers(partitionsAlreadyExisting, brokers, followerPartitionStates, highWatermarkCheckpoints, None, metadataImage.topicNameToId) // We need to transition anything that hasn't transitioned from Deferred to Offline to the Online state. deferredPartitionsIterator.foreach { deferredPartition => @@ -169,7 +168,7 @@ class RaftReplicaManager(config: KafkaConfig, updateLeaderAndFollowerMetrics(partitionsMadeFollower.map(_.topic).toSet) - maybeAddLogDirFetchers(partitionsMadeFollower, highWatermarkCheckpoints) + maybeAddLogDirFetchers(partitionsMadeFollower, highWatermarkCheckpoints, metadataImage.topicNameToId) replicaFetcherManager.shutdownIdleFetcherThreads() replicaAlterLogDirsManager.shutdownIdleFetcherThreads() @@ -247,6 +246,7 @@ class RaftReplicaManager(config: KafkaConfig, (Some(Partition(topicPartition, time, configRepository, this)), None) } partition.foreach { partition => + checkTopicId(builder.topicNameToId(partition.topic), partition.topicId, partition.topicPartition) val isNew = priorDeferredMetadata match { case Some(alreadyDeferred) => alreadyDeferred.isNew case _ => prevPartitions.topicPartition(topicPartition.topic(), topicPartition.partition()).isEmpty @@ -282,6 +282,7 @@ class RaftReplicaManager(config: KafkaConfig, Some(partition) } partition.foreach { partition => + checkTopicId(builder.topicNameToId(partition.topic), partition.topicId, partition.topicPartition) if (currentState.leaderId == localBrokerId) { partitionsToBeLeader.put(partition, currentState) } else { @@ -299,12 +300,12 @@ class RaftReplicaManager(config: KafkaConfig, val highWatermarkCheckpoints = new LazyOffsetCheckpoints(this.highWatermarkCheckpoints) val partitionsBecomeLeader = if (partitionsToBeLeader.nonEmpty) delegate.makeLeaders(changedPartitionsPreviouslyExisting, partitionsToBeLeader, highWatermarkCheckpoints, - Some(metadataOffset)) + Some(metadataOffset), builder.topicNameToId) else Set.empty[Partition] val partitionsBecomeFollower = if (partitionsToBeFollower.nonEmpty) delegate.makeFollowers(changedPartitionsPreviouslyExisting, nextBrokers, partitionsToBeFollower, highWatermarkCheckpoints, - Some(metadataOffset)) + Some(metadataOffset), builder.topicNameToId) else Set.empty[Partition] updateLeaderAndFollowerMetrics(partitionsBecomeFollower.map(_.topic).toSet) @@ -322,7 +323,7 @@ class RaftReplicaManager(config: KafkaConfig, } } - maybeAddLogDirFetchers(partitionsBecomeFollower, highWatermarkCheckpoints) + maybeAddLogDirFetchers(partitionsBecomeFollower, highWatermarkCheckpoints, builder.topicNameToId) replicaFetcherManager.shutdownIdleFetcherThreads() replicaAlterLogDirsManager.shutdownIdleFetcherThreads() @@ -374,4 +375,30 @@ class RaftReplicaManager(config: KafkaConfig, 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}")) } + + /** + * Checks if the topic ID received from the MetadataPartitionsBuilder is consistent with the topic ID in the log. + * If the log does not exist, logTopicIdOpt will be None. In this case, the ID is not inconsistent. + * + * @param receivedTopicIdOpt the topic ID received from the MetadataRecords if it exists + * @param logTopicIdOpt the topic ID in the log if the log exists + * @param topicPartition the topicPartition for the Partition being checked + * @throws InconsistentTopicIdException if the topic ids are not consistent + * @throws IllegalArgumentException if the MetadataPartitionsBuilder did not have a topic ID associated with the topic + */ + private def checkTopicId(receivedTopicIdOpt: Option[Uuid], logTopicIdOpt: Option[Uuid], topicPartition: TopicPartition): Unit = { + receivedTopicIdOpt match { + case Some(receivedTopicId) => + logTopicIdOpt.foreach { logTopicId => + if (receivedTopicId != logTopicId) { + // not sure if we need both the logger and the error thrown + stateChangeLogger.error(s"Topic ID in memory: $logTopicId does not" + + s" match the topic ID for partition $topicPartition received: " + + s"$receivedTopicId.") + throw new InconsistentTopicIdException(s"Topic partition $topicPartition had an inconsistent topic ID.") + } + } + case None => throw new IllegalStateException(s"Topic partition $topicPartition is missing a topic ID") + } + } } diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 80beadea108aa..a97f29b83d954 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -37,7 +37,7 @@ import kafka.server.metadata.ConfigRepository import kafka.utils._ import kafka.utils.Implicits._ import kafka.zk.KafkaZkClient -import org.apache.kafka.common.{ElectionType, IsolationLevel, Node, TopicPartition} +import org.apache.kafka.common.{ElectionType, IsolationLevel, Node, TopicPartition, Uuid} import org.apache.kafka.common.errors._ import org.apache.kafka.common.internals.Topic import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState @@ -1326,6 +1326,14 @@ class ReplicaManager(val config: KafkaConfig, s"epoch ${leaderAndIsrRequest.controllerEpoch}") } val topicIds = leaderAndIsrRequest.topicIds() + def topicIdFromRequest(topicName: String): Option[Uuid] = { + val topicId = topicIds.get(topicName) + // if invalid topic ID return None + if (topicId == null || topicId == Uuid.ZERO_UUID) + None + else + Some(topicId) + } val response = { if (leaderAndIsrRequest.controllerEpoch < controllerEpoch) { @@ -1367,9 +1375,12 @@ class ReplicaManager(val config: KafkaConfig, partitionOpt.foreach { partition => val currentLeaderEpoch = partition.getLeaderEpoch val requestLeaderEpoch = partitionState.leaderEpoch - val requestTopicId = topicIds.get(topicPartition.topic) + val requestTopicId = topicIdFromRequest(topicPartition.topic) - if (!partition.checkOrSetTopicId(requestTopicId)) { + if (!hasConsistentTopicId(requestTopicId, partition.topicId)) { + stateChangeLogger.error(s"Topic ID in memory: ${partition.topicId.get} does not" + + s" match the topic ID for partition $topicPartition received: " + + s"${requestTopicId.get}.") responseMap.put(topicPartition, Errors.INCONSISTENT_TOPIC_ID) } else if (requestLeaderEpoch > currentLeaderEpoch) { // If the leader epoch is valid record the epoch of the controller that made the leadership decision. @@ -1407,12 +1418,12 @@ class ReplicaManager(val config: KafkaConfig, val highWatermarkCheckpoints = new LazyOffsetCheckpoints(this.highWatermarkCheckpoints) val partitionsBecomeLeader = if (partitionsToBeLeader.nonEmpty) makeLeaders(controllerId, controllerEpoch, partitionsToBeLeader, correlationId, responseMap, - highWatermarkCheckpoints) + highWatermarkCheckpoints, topicIdFromRequest) else Set.empty[Partition] val partitionsBecomeFollower = if (partitionsToBeFollower.nonEmpty) makeFollowers(controllerId, controllerEpoch, partitionsToBeFollower, correlationId, responseMap, - highWatermarkCheckpoints) + highWatermarkCheckpoints, topicIdFromRequest) else Set.empty[Partition] @@ -1435,7 +1446,7 @@ class ReplicaManager(val config: KafkaConfig, // have been completely populated before starting the checkpointing there by avoiding weird race conditions startHighWatermarkCheckPointThread() - maybeAddLogDirFetchers(partitionStates.keySet, highWatermarkCheckpoints) + maybeAddLogDirFetchers(partitionStates.keySet, highWatermarkCheckpoints, topicIdFromRequest) replicaFetcherManager.shutdownIdleFetcherThreads() replicaAlterLogDirsManager.shutdownIdleFetcherThreads() @@ -1473,6 +1484,25 @@ class ReplicaManager(val config: KafkaConfig, } } + /** + * Checks if the topic ID provided in the LeaderAndIsr request is consistent with the topic ID in the log. + * + * If the request had an invalid topic ID (null or zero), then we assume that topic IDs are not supported. + * The topic ID was not inconsistent, so return true. + * If the log does not exist or the topic ID is not yet set, logTopicIdOpt will be None. + * In both cases, the ID is not inconsistent so return true. + * + * @param requestTopicIdOpt the topic ID from the LeaderAndIsr request if it exists + * @param logTopicIdOpt the topic ID in the log if the log and the topic ID exist + * @return true if the request topic id is consistent, false otherwise + */ + private def hasConsistentTopicId(requestTopicIdOpt: Option[Uuid], logTopicIdOpt: Option[Uuid]): Boolean = { + requestTopicIdOpt match { + case None => true + case Some(requestTopicId) => logTopicIdOpt.isEmpty || logTopicIdOpt.contains(requestTopicId) + } + } + /** * KAFKA-8392 * For topic partitions of which the broker is no longer a leader, delete metrics related to @@ -1488,7 +1518,8 @@ class ReplicaManager(val config: KafkaConfig, } protected def maybeAddLogDirFetchers(partitions: Set[Partition], - offsetCheckpoints: OffsetCheckpoints): Unit = { + offsetCheckpoints: OffsetCheckpoints, + topicIds: String => Option[Uuid]): Unit = { val futureReplicasAndInitialOffset = new mutable.HashMap[TopicPartition, InitialFetchState] for (partition <- partitions) { val topicPartition = partition.topicPartition @@ -1500,7 +1531,8 @@ class ReplicaManager(val config: KafkaConfig, partition.createLogIfNotExists( isNew = false, isFutureReplica = true, - offsetCheckpoints) + offsetCheckpoints, + topicIds(partition.topic)) // pause cleaning for partitions that are being moved and start ReplicaAlterDirThread to move // replica from source dir to destination dir @@ -1534,7 +1566,8 @@ class ReplicaManager(val config: KafkaConfig, partitionStates: Map[Partition, LeaderAndIsrPartitionState], correlationId: Int, responseMap: mutable.Map[TopicPartition, Errors], - highWatermarkCheckpoints: OffsetCheckpoints): Set[Partition] = { + highWatermarkCheckpoints: OffsetCheckpoints, + topicIds: String => Option[Uuid]): Set[Partition] = { val traceEnabled = stateChangeLogger.isTraceEnabled partitionStates.keys.foreach { partition => if (traceEnabled) @@ -1555,7 +1588,7 @@ class ReplicaManager(val config: KafkaConfig, // Update the partition information to be the leader partitionStates.forKeyValue { (partition, partitionState) => try { - if (partition.makeLeader(partitionState, highWatermarkCheckpoints)) + if (partition.makeLeader(partitionState, highWatermarkCheckpoints, topicIds(partitionState.topicName))) partitionsToMakeLeaders += partition else stateChangeLogger.info(s"Skipped the become-leader state change after marking its " + @@ -1616,7 +1649,8 @@ class ReplicaManager(val config: KafkaConfig, partitionStates: Map[Partition, LeaderAndIsrPartitionState], correlationId: Int, responseMap: mutable.Map[TopicPartition, Errors], - highWatermarkCheckpoints: OffsetCheckpoints) : Set[Partition] = { + highWatermarkCheckpoints: OffsetCheckpoints, + topicIds: String => Option[Uuid]) : Set[Partition] = { val traceLoggingEnabled = stateChangeLogger.isTraceEnabled partitionStates.forKeyValue { (partition, partitionState) => if (traceLoggingEnabled) @@ -1635,7 +1669,7 @@ class ReplicaManager(val config: KafkaConfig, metadataCache.getAliveBrokers.find(_.id == newLeaderBrokerId) match { // Only change partition state when the leader is available case Some(_) => - if (partition.makeFollower(partitionState, highWatermarkCheckpoints)) + if (partition.makeFollower(partitionState, highWatermarkCheckpoints, topicIds(partitionState.topicName))) partitionsToMakeFollower += partition else stateChangeLogger.info(s"Skipped the become-follower state change after marking its partition as " + @@ -1653,7 +1687,7 @@ class ReplicaManager(val config: KafkaConfig, // 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 = partitionState.isNew, isFutureReplica = false, - highWatermarkCheckpoints) + highWatermarkCheckpoints, topicIds(partitionState.topicName)) } } catch { case e: KafkaStorageException => diff --git a/core/src/main/scala/kafka/server/metadata/MetadataImage.scala b/core/src/main/scala/kafka/server/metadata/MetadataImage.scala index f4e5831e8ab1a..4687c71922067 100755 --- a/core/src/main/scala/kafka/server/metadata/MetadataImage.scala +++ b/core/src/main/scala/kafka/server/metadata/MetadataImage.scala @@ -47,6 +47,14 @@ case class MetadataImageBuilder(brokerId: Int, } } + def topicNameToId(topicName: String): Option[Uuid] = { + if (_partitionsBuilder != null) { + _partitionsBuilder.topicNameToId(topicName) + } else { + prevImage.topicNameToId(topicName) + } + } + def controllerId(controllerId: Option[Int]): Unit = { _controllerId = controllerId } diff --git a/core/src/test/scala/other/kafka/StressTestLog.scala b/core/src/test/scala/other/kafka/StressTestLog.scala index 8162d4d193763..e3c1e65ea4a9f 100755 --- a/core/src/test/scala/other/kafka/StressTestLog.scala +++ b/core/src/test/scala/other/kafka/StressTestLog.scala @@ -51,7 +51,9 @@ object StressTestLog { maxProducerIdExpirationMs = 60 * 60 * 1000, producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs, brokerTopicStats = new BrokerTopicStats, - logDirFailureChannel = new LogDirFailureChannel(10)) + logDirFailureChannel = new LogDirFailureChannel(10), + topicId = None, + keepPartitionMetadataFile = true) val writer = new WriterThread(log) writer.start() val reader = new ReaderThread(log) diff --git a/core/src/test/scala/other/kafka/TestLinearWriteSpeed.scala b/core/src/test/scala/other/kafka/TestLinearWriteSpeed.scala index 24a49c32d50cf..cb2180ae476d8 100755 --- a/core/src/test/scala/other/kafka/TestLinearWriteSpeed.scala +++ b/core/src/test/scala/other/kafka/TestLinearWriteSpeed.scala @@ -211,7 +211,7 @@ object TestLinearWriteSpeed { class LogWritable(val dir: File, config: LogConfig, scheduler: Scheduler, val messages: MemoryRecords) extends Writable { Utils.delete(dir) val log = Log(dir, config, 0L, 0L, scheduler, new BrokerTopicStats, Time.SYSTEM, 60 * 60 * 1000, - LogManager.ProducerIdExpirationCheckIntervalMs, new LogDirFailureChannel(10)) + LogManager.ProducerIdExpirationCheckIntervalMs, new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true) def write(): Int = { log.appendAsLeader(messages, leaderEpoch = 0) messages.sizeInBytes diff --git a/core/src/test/scala/unit/kafka/cluster/AssignmentStateTest.scala b/core/src/test/scala/unit/kafka/cluster/AssignmentStateTest.scala index 5e087c5e407fd..a618825e8a5f1 100644 --- a/core/src/test/scala/unit/kafka/cluster/AssignmentStateTest.scala +++ b/core/src/test/scala/unit/kafka/cluster/AssignmentStateTest.scala @@ -108,7 +108,7 @@ class AssignmentStateTest extends AbstractPartitionTest { if (original.nonEmpty) partition.assignmentState = SimpleAssignmentState(original) // do the test - partition.makeLeader(leaderState, offsetCheckpoints) + partition.makeLeader(leaderState, offsetCheckpoints, None) assertEquals(isReassigning, partition.isReassigning) if (adding.nonEmpty) adding.foreach(r => assertTrue(partition.isAddingReplica(r))) diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala index b105df3110102..a8198c9c02e09 100644 --- a/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala +++ b/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala @@ -27,7 +27,7 @@ import kafka.server._ import kafka.server.checkpoints.OffsetCheckpoints import kafka.utils._ import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState -import org.apache.kafka.common.TopicPartition +import org.apache.kafka.common.{TopicPartition, Uuid} import org.apache.kafka.common.record.{MemoryRecords, SimpleRecord} import org.apache.kafka.common.utils.Utils import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} @@ -142,7 +142,7 @@ class PartitionLockTest extends Logging { .setIsNew(true) val offsetCheckpoints: OffsetCheckpoints = mock(classOf[OffsetCheckpoints]) // Update replica set synchronously first to avoid race conditions - partition.makeLeader(partitionState(secondReplicaSet), offsetCheckpoints) + partition.makeLeader(partitionState(secondReplicaSet), offsetCheckpoints, None) assertTrue(partition.getReplica(replicaToCheck).isDefined, s"Expected replica $replicaToCheck to be defined") val future = executorService.submit((() => { @@ -155,7 +155,7 @@ class PartitionLockTest extends Logging { secondReplicaSet } - partition.makeLeader(partitionState(replicas), offsetCheckpoints) + partition.makeLeader(partitionState(replicas), offsetCheckpoints, None) i += 1 Thread.sleep(1) // just to avoid tight loop @@ -278,8 +278,8 @@ class PartitionLockTest extends Logging { } } - override def createLog(isNew: Boolean, isFutureReplica: Boolean, offsetCheckpoints: OffsetCheckpoints): Log = { - val log = super.createLog(isNew, isFutureReplica, offsetCheckpoints) + override def createLog(isNew: Boolean, isFutureReplica: Boolean, offsetCheckpoints: OffsetCheckpoints, topicId: Option[Uuid]): Log = { + val log = super.createLog(isNew, isFutureReplica, offsetCheckpoints, None) new SlowLog(log, mockTime, appendSemaphore) } } @@ -288,7 +288,7 @@ class PartitionLockTest extends Logging { when(alterIsrManager.submit(ArgumentMatchers.any[AlterIsrItem])) .thenReturn(true) - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val controllerEpoch = 0 val replicas = (0 to numReplicaFetchers).map(i => Integer.valueOf(brokerId + i)).toList.asJava @@ -301,7 +301,7 @@ class PartitionLockTest extends Logging { .setIsr(isr) .setZkVersion(1) .setReplicas(replicas) - .setIsNew(true), offsetCheckpoints), "Expected become leader transition to succeed") + .setIsNew(true), offsetCheckpoints, None), "Expected become leader transition to succeed") partition } @@ -353,7 +353,9 @@ class PartitionLockTest extends Logging { log.producerIdExpirationCheckIntervalMs, log.topicPartition, log.producerStateManager, - new LogDirFailureChannel(1)) { + new LogDirFailureChannel(1), + topicId = None, + keepPartitionMetadataFile = true) { override def appendAsLeader(records: MemoryRecords, leaderEpoch: Int, origin: AppendOrigin, interBrokerProtocolVersion: ApiVersion): LogAppendInfo = { val appendInfo = super.appendAsLeader(records, leaderEpoch, origin, interBrokerProtocolVersion) diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala index 9e66969602645..dd671e543e29c 100644 --- a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala +++ b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala @@ -25,7 +25,7 @@ import kafka.server._ import kafka.server.checkpoints.OffsetCheckpoints import kafka.utils._ import kafka.zk.KafkaZkClient -import org.apache.kafka.common.errors.{ApiException, NotLeaderOrFollowerException, OffsetNotAvailableException, OffsetOutOfRangeException} +import org.apache.kafka.common.errors.{ApiException, InconsistentTopicIdException, NotLeaderOrFollowerException, OffsetNotAvailableException, OffsetOutOfRangeException} import org.apache.kafka.common.message.FetchResponseData import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState import org.apache.kafka.common.protocol.Errors @@ -34,7 +34,7 @@ import org.apache.kafka.common.record._ import org.apache.kafka.common.requests.ListOffsetsRequest import org.apache.kafka.common.requests.OffsetsForLeaderEpochResponse.{UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET} import org.apache.kafka.common.utils.SystemTime -import org.apache.kafka.common.{IsolationLevel, TopicPartition} +import org.apache.kafka.common.{IsolationLevel, TopicPartition, Uuid} import org.junit.jupiter.api.Assertions._ import org.junit.jupiter.api.Test import org.mockito.ArgumentMatchers @@ -51,7 +51,7 @@ class PartitionTest extends AbstractPartitionTest { @Test def testLastFetchedOffsetValidation(): Unit = { - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) def append(leaderEpoch: Int, count: Int): Unit = { val recordArray = (1 to count).map { i => new SimpleRecord(s"$i".getBytes) @@ -130,7 +130,7 @@ class PartitionTest extends AbstractPartitionTest { def testMakeLeaderUpdatesEpochCache(): Unit = { val leaderEpoch = 8 - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) log.appendAsLeader(MemoryRecords.withRecords(0L, CompressionType.NONE, 0, new SimpleRecord("k1".getBytes, "v1".getBytes), new SimpleRecord("k2".getBytes, "v2".getBytes) @@ -156,7 +156,7 @@ class PartitionTest extends AbstractPartitionTest { configRepository.setTopicConfig(topicPartition.topic, LogConfig.MessageFormatVersionProp, kafka.api.KAFKA_0_10_2_IV0.shortVersion) - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) log.appendAsLeader(TestUtils.records(List( new SimpleRecord("k1".getBytes, "v1".getBytes), new SimpleRecord("k2".getBytes, "v2".getBytes)), @@ -185,7 +185,7 @@ class PartitionTest extends AbstractPartitionTest { val latch = new CountDownLatch(1) logManager.maybeUpdatePreferredLogDir(topicPartition, logDir1.getAbsolutePath) - partition.createLogIfNotExists(isNew = true, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = true, isFutureReplica = false, offsetCheckpoints, None) logManager.maybeUpdatePreferredLogDir(topicPartition, logDir2.getAbsolutePath) partition.maybeCreateFutureReplica(logDir2.getAbsolutePath, offsetCheckpoints) @@ -230,13 +230,13 @@ class PartitionTest extends AbstractPartitionTest { logManager, alterIsrManager) { - override def createLog(isNew: Boolean, isFutureReplica: Boolean, offsetCheckpoints: OffsetCheckpoints): Log = { - val log = super.createLog(isNew, isFutureReplica, offsetCheckpoints) + override def createLog(isNew: Boolean, isFutureReplica: Boolean, offsetCheckpoints: OffsetCheckpoints, topicId: Option[Uuid]): Log = { + val log = super.createLog(isNew, isFutureReplica, offsetCheckpoints, None) new SlowLog(log, mockTime, appendSemaphore) } } - partition.createLogIfNotExists(isNew = true, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = true, isFutureReplica = false, offsetCheckpoints, None) val appendThread = new Thread { override def run(): Unit = { @@ -257,7 +257,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(List[Integer](0, 1, 2, brokerId).asJava) .setIsNew(false) - assertTrue(partition.makeFollower(partitionState, offsetCheckpoints)) + assertTrue(partition.makeFollower(partitionState, offsetCheckpoints, None)) appendSemaphore.release() appendThread.join() @@ -271,7 +271,7 @@ class PartitionTest extends AbstractPartitionTest { // active segment def testMaybeReplaceCurrentWithFutureReplicaDifferentBaseOffsets(): Unit = { logManager.maybeUpdatePreferredLogDir(topicPartition, logDir1.getAbsolutePath) - partition.createLogIfNotExists(isNew = true, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = true, isFutureReplica = false, offsetCheckpoints, None) logManager.maybeUpdatePreferredLogDir(topicPartition, logDir2.getAbsolutePath) partition.maybeCreateFutureReplica(logDir2.getAbsolutePath, offsetCheckpoints) @@ -567,7 +567,7 @@ class PartitionTest extends AbstractPartitionTest { .setReplicas(replicas.map(Int.box).asJava) .setIsNew(true) - assertTrue(partition.makeLeader(leaderState, offsetCheckpoints), "Expected first makeLeader() to return 'leader changed'") + assertTrue(partition.makeLeader(leaderState, offsetCheckpoints, None), "Expected first makeLeader() to return 'leader changed'") assertEquals(leaderEpoch, partition.getLeaderEpoch, "Current leader epoch") assertEquals(Set[Integer](leader, follower2), partition.isrState.isr, "ISR") @@ -640,7 +640,7 @@ class PartitionTest extends AbstractPartitionTest { .setReplicas(replicas.map(Int.box).asJava) .setIsNew(false) - assertTrue(partition.makeFollower(followerState, offsetCheckpoints)) + assertTrue(partition.makeFollower(followerState, offsetCheckpoints, None)) // Back to leader, this resets the startLogOffset for this epoch (to 2), we're now in the fault condition val newLeaderState = new LeaderAndIsrPartitionState() @@ -652,7 +652,7 @@ class PartitionTest extends AbstractPartitionTest { .setReplicas(replicas.map(Int.box).asJava) .setIsNew(false) - assertTrue(partition.makeLeader(newLeaderState, offsetCheckpoints)) + assertTrue(partition.makeLeader(newLeaderState, offsetCheckpoints, None)) // Try to get offsets as a client fetchOffsetsForTimestamp(ListOffsetsRequest.LATEST_TIMESTAMP, Some(IsolationLevel.READ_UNCOMMITTED)) match { @@ -713,8 +713,8 @@ class PartitionTest extends AbstractPartitionTest { private def setupPartitionWithMocks(leaderEpoch: Int, isLeader: Boolean, - log: Log = logManager.getOrCreateLog(topicPartition)): Partition = { - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + log: Log = logManager.getOrCreateLog(topicPartition, topicId = None)): Partition = { + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val controllerEpoch = 0 val replicas = List[Integer](brokerId, brokerId + 1).asJava @@ -728,7 +728,7 @@ class PartitionTest extends AbstractPartitionTest { .setIsr(isr) .setZkVersion(1) .setReplicas(replicas) - .setIsNew(true), offsetCheckpoints), "Expected become leader transition to succeed") + .setIsNew(true), offsetCheckpoints, None), "Expected become leader transition to succeed") assertEquals(leaderEpoch, partition.getLeaderEpoch) } else { assertTrue(partition.makeFollower(new LeaderAndIsrPartitionState() @@ -738,7 +738,7 @@ class PartitionTest extends AbstractPartitionTest { .setIsr(isr) .setZkVersion(1) .setReplicas(replicas) - .setIsNew(true), offsetCheckpoints), "Expected become follower transition to succeed") + .setIsNew(true), offsetCheckpoints, None), "Expected become follower transition to succeed") assertEquals(leaderEpoch, partition.getLeaderEpoch) assertEquals(None, partition.leaderLogIfLocal) } @@ -748,7 +748,7 @@ class PartitionTest extends AbstractPartitionTest { @Test def testAppendRecordsAsFollowerBelowLogStartOffset(): Unit = { - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val log = partition.localLogOrException val initialLogStartOffset = 5L @@ -802,7 +802,7 @@ class PartitionTest extends AbstractPartitionTest { val replicas = List[Integer](brokerId, brokerId + 1).asJava val isr = replicas - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) assertTrue(partition.makeLeader(new LeaderAndIsrPartitionState() .setControllerEpoch(controllerEpoch) @@ -811,7 +811,7 @@ class PartitionTest extends AbstractPartitionTest { .setIsr(isr) .setZkVersion(1) .setReplicas(replicas) - .setIsNew(true), offsetCheckpoints), "Expected become leader transition to succeed") + .setIsNew(true), offsetCheckpoints, None), "Expected become leader transition to succeed") assertEquals(leaderEpoch, partition.getLeaderEpoch) val records = createTransactionalRecords(List( @@ -881,7 +881,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(List[Integer](0, 1, 2, brokerId).asJava) .setIsNew(false) - partition.makeFollower(partitionState, offsetCheckpoints) + partition.makeFollower(partitionState, offsetCheckpoints, None) // Request with same leader and epoch increases by only 1, do become-follower steps partitionState = new LeaderAndIsrPartitionState() @@ -892,7 +892,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(List[Integer](0, 1, 2, brokerId).asJava) .setIsNew(false) - assertTrue(partition.makeFollower(partitionState, offsetCheckpoints)) + assertTrue(partition.makeFollower(partitionState, offsetCheckpoints, None)) // Request with same leader and same epoch, skip become-follower steps partitionState = new LeaderAndIsrPartitionState() @@ -902,7 +902,7 @@ class PartitionTest extends AbstractPartitionTest { .setIsr(List[Integer](0, 1, 2, brokerId).asJava) .setZkVersion(1) .setReplicas(List[Integer](0, 1, 2, brokerId).asJava) - assertFalse(partition.makeFollower(partitionState, offsetCheckpoints)) + assertFalse(partition.makeFollower(partitionState, offsetCheckpoints, None)) } @Test @@ -930,7 +930,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas) .setIsNew(true) - assertTrue(partition.makeLeader(leaderState, offsetCheckpoints), "Expected first makeLeader() to return 'leader changed'") + assertTrue(partition.makeLeader(leaderState, offsetCheckpoints, None), "Expected first makeLeader() to return 'leader changed'") assertEquals(leaderEpoch, partition.getLeaderEpoch, "Current leader epoch") assertEquals(Set[Integer](leader, follower2), partition.isrState.isr, "ISR") @@ -964,7 +964,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas) .setIsNew(false) - partition.makeFollower(followerState, offsetCheckpoints) + partition.makeFollower(followerState, offsetCheckpoints, None) val newLeaderState = new LeaderAndIsrPartitionState() .setControllerEpoch(controllerEpoch) @@ -974,7 +974,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas) .setIsNew(false) - assertTrue(partition.makeLeader(newLeaderState, offsetCheckpoints), + assertTrue(partition.makeLeader(newLeaderState, offsetCheckpoints, None), "Expected makeLeader() to return 'leader changed' after makeFollower()") val currentLeaderEpochStartOffset = partition.localLogOrException.logEndOffset @@ -1043,13 +1043,13 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas) .setIsNew(true) - partition.makeLeader(leaderState, offsetCheckpoints) + partition.makeLeader(leaderState, offsetCheckpoints, None) assertTrue(partition.isAtMinIsr) } @Test def testUpdateFollowerFetchState(): Unit = { - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) seedLogData(log, numRecords = 6, leaderEpoch = 4) val controllerEpoch = 0 @@ -1058,7 +1058,7 @@ class PartitionTest extends AbstractPartitionTest { val replicas = List[Integer](brokerId, remoteBrokerId).asJava val isr = replicas - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val initializeTimeMs = time.milliseconds() assertTrue(partition.makeLeader( @@ -1070,7 +1070,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas) .setIsNew(true), - offsetCheckpoints), "Expected become leader transition to succeed") + offsetCheckpoints, None), "Expected become leader transition to succeed") val remoteReplica = partition.getReplica(remoteBrokerId).get assertEquals(initializeTimeMs, remoteReplica.lastCaughtUpTimeMs) @@ -1105,7 +1105,7 @@ class PartitionTest extends AbstractPartitionTest { @Test def testIsrExpansion(): Unit = { - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) seedLogData(log, numRecords = 10, leaderEpoch = 4) val controllerEpoch = 0 @@ -1114,7 +1114,7 @@ class PartitionTest extends AbstractPartitionTest { val replicas = List(brokerId, remoteBrokerId) val isr = List[Integer](brokerId).asJava - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) assertTrue(partition.makeLeader( new LeaderAndIsrPartitionState() .setControllerEpoch(controllerEpoch) @@ -1124,7 +1124,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas.map(Int.box).asJava) .setIsNew(true), - offsetCheckpoints), "Expected become leader transition to succeed") + offsetCheckpoints, None), "Expected become leader transition to succeed") assertEquals(Set(brokerId), partition.isrState.isr) val remoteReplica = partition.getReplica(remoteBrokerId).get @@ -1166,7 +1166,7 @@ class PartitionTest extends AbstractPartitionTest { @Test def testIsrNotExpandedIfUpdateFails(): Unit = { - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) seedLogData(log, numRecords = 10, leaderEpoch = 4) val controllerEpoch = 0 @@ -1175,7 +1175,7 @@ class PartitionTest extends AbstractPartitionTest { val replicas = List[Integer](brokerId, remoteBrokerId).asJava val isr = List[Integer](brokerId).asJava - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) assertTrue(partition.makeLeader( new LeaderAndIsrPartitionState() .setControllerEpoch(controllerEpoch) @@ -1185,7 +1185,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas) .setIsNew(true), - offsetCheckpoints), "Expected become leader transition to succeed") + offsetCheckpoints, None), "Expected become leader transition to succeed") assertEquals(Set(brokerId), partition.isrState.isr) val remoteReplica = partition.getReplica(remoteBrokerId).get @@ -1220,7 +1220,7 @@ class PartitionTest extends AbstractPartitionTest { @Test def testMaybeShrinkIsr(): Unit = { - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) seedLogData(log, numRecords = 10, leaderEpoch = 4) val controllerEpoch = 0 @@ -1230,7 +1230,7 @@ class PartitionTest extends AbstractPartitionTest { val isr = List[Integer](brokerId, remoteBrokerId).asJava val initializeTimeMs = time.milliseconds() - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) assertTrue(partition.makeLeader( new LeaderAndIsrPartitionState() .setControllerEpoch(controllerEpoch) @@ -1240,7 +1240,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas.map(Int.box).asJava) .setIsNew(true), - offsetCheckpoints), "Expected become leader transition to succeed") + offsetCheckpoints, None), "Expected become leader transition to succeed") assertEquals(Set(brokerId, remoteBrokerId), partition.isrState.isr) assertEquals(0L, partition.localLogOrException.highWatermark) @@ -1267,7 +1267,7 @@ class PartitionTest extends AbstractPartitionTest { @Test def testShouldNotShrinkIsrIfPreviousFetchIsCaughtUp(): Unit = { - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) seedLogData(log, numRecords = 10, leaderEpoch = 4) val controllerEpoch = 0 @@ -1277,7 +1277,7 @@ class PartitionTest extends AbstractPartitionTest { val isr = List[Integer](brokerId, remoteBrokerId).asJava val initializeTimeMs = time.milliseconds() - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) assertTrue(partition.makeLeader( new LeaderAndIsrPartitionState() .setControllerEpoch(controllerEpoch) @@ -1287,7 +1287,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas.map(Int.box).asJava) .setIsNew(true), - offsetCheckpoints), "Expected become leader transition to succeed") + offsetCheckpoints, None), "Expected become leader transition to succeed") assertEquals(Set(brokerId, remoteBrokerId), partition.isrState.isr) assertEquals(0L, partition.localLogOrException.highWatermark) @@ -1332,7 +1332,7 @@ class PartitionTest extends AbstractPartitionTest { @Test def testShouldNotShrinkIsrIfFollowerCaughtUpToLogEnd(): Unit = { - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) seedLogData(log, numRecords = 10, leaderEpoch = 4) val controllerEpoch = 0 @@ -1342,7 +1342,7 @@ class PartitionTest extends AbstractPartitionTest { val isr = List[Integer](brokerId, remoteBrokerId).asJava val initializeTimeMs = time.milliseconds() - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) assertTrue(partition.makeLeader( new LeaderAndIsrPartitionState() .setControllerEpoch(controllerEpoch) @@ -1352,7 +1352,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas.map(Int.box).asJava) .setIsNew(true), - offsetCheckpoints), "Expected become leader transition to succeed") + offsetCheckpoints, None), "Expected become leader transition to succeed") assertEquals(Set(brokerId, remoteBrokerId), partition.isrState.isr) assertEquals(0L, partition.localLogOrException.highWatermark) @@ -1383,7 +1383,7 @@ class PartitionTest extends AbstractPartitionTest { @Test def testIsrNotShrunkIfUpdateFails(): Unit = { - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) seedLogData(log, numRecords = 10, leaderEpoch = 4) val controllerEpoch = 0 @@ -1393,7 +1393,7 @@ class PartitionTest extends AbstractPartitionTest { val isr = List[Integer](brokerId, remoteBrokerId).asJava val initializeTimeMs = time.milliseconds() - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) assertTrue(partition.makeLeader( new LeaderAndIsrPartitionState() .setControllerEpoch(controllerEpoch) @@ -1403,7 +1403,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas) .setIsNew(true), - offsetCheckpoints), "Expected become leader transition to succeed") + offsetCheckpoints, None), "Expected become leader transition to succeed") assertEquals(Set(brokerId, remoteBrokerId), partition.inSyncReplicaIds) assertEquals(0L, partition.localLogOrException.highWatermark) @@ -1462,7 +1462,7 @@ class PartitionTest extends AbstractPartitionTest { } def handleAlterIsrFailure(error: Errors, callback: (Int, Int, Partition) => Unit): Unit = { - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) seedLogData(log, numRecords = 10, leaderEpoch = 4) val controllerEpoch = 0 @@ -1471,7 +1471,7 @@ class PartitionTest extends AbstractPartitionTest { val replicas = List[Integer](brokerId, remoteBrokerId).asJava val isr = List[Integer](brokerId).asJava - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) assertTrue(partition.makeLeader( new LeaderAndIsrPartitionState() .setControllerEpoch(controllerEpoch) @@ -1481,7 +1481,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas) .setIsNew(true), - offsetCheckpoints), "Expected become leader transition to succeed") + offsetCheckpoints, None), "Expected become leader transition to succeed") assertEquals(Set(brokerId), partition.isrState.isr) val remoteReplica = partition.getReplica(remoteBrokerId).get @@ -1509,7 +1509,7 @@ class PartitionTest extends AbstractPartitionTest { @Test def testSingleInFlightAlterIsr(): Unit = { - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) seedLogData(log, numRecords = 10, leaderEpoch = 4) val controllerEpoch = 0 @@ -1522,7 +1522,7 @@ class PartitionTest extends AbstractPartitionTest { doNothing().when(delayedOperations).checkAndCompleteAll() - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) assertTrue(partition.makeLeader( new LeaderAndIsrPartitionState() .setControllerEpoch(controllerEpoch) @@ -1532,7 +1532,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas) .setIsNew(true), - offsetCheckpoints), "Expected become leader transition to succeed") + offsetCheckpoints, None), "Expected become leader transition to succeed") assertEquals(Set(brokerId, follower1, follower2), partition.isrState.isr) assertEquals(0L, partition.localLogOrException.highWatermark) @@ -1574,7 +1574,7 @@ class PartitionTest extends AbstractPartitionTest { logManager, zkIsrManager) - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) seedLogData(log, numRecords = 10, leaderEpoch = 4) val controllerEpoch = 0 @@ -1587,7 +1587,7 @@ class PartitionTest extends AbstractPartitionTest { doNothing().when(delayedOperations).checkAndCompleteAll() - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) assertTrue(partition.makeLeader( new LeaderAndIsrPartitionState() .setControllerEpoch(controllerEpoch) @@ -1597,7 +1597,7 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas) .setIsNew(true), - offsetCheckpoints), "Expected become leader transition to succeed") + offsetCheckpoints, None), "Expected become leader transition to succeed") assertEquals(Set(brokerId, follower1, follower2), partition.isrState.isr) assertEquals(0L, partition.localLogOrException.highWatermark) @@ -1617,7 +1617,7 @@ class PartitionTest extends AbstractPartitionTest { @Test def testUseCheckpointToInitializeHighWatermark(): Unit = { - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) seedLogData(log, numRecords = 6, leaderEpoch = 5) when(offsetCheckpoints.fetch(logDir1.getAbsolutePath, topicPartition)) @@ -1633,10 +1633,108 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas) .setIsNew(false) - partition.makeLeader(leaderState, offsetCheckpoints) + partition.makeLeader(leaderState, offsetCheckpoints, None) assertEquals(4, partition.localLogOrException.highWatermark) } + @Test + def testTopicIdAndPartitionMetadataFileForLeader(): Unit = { + val controllerEpoch = 3 + val leaderEpoch = 5 + val topicId = Uuid.randomUuid() + val replicas = List[Integer](brokerId, brokerId + 1).asJava + val leaderState = new LeaderAndIsrPartitionState() + .setControllerEpoch(controllerEpoch) + .setLeader(brokerId) + .setLeaderEpoch(leaderEpoch) + .setIsr(replicas) + .setZkVersion(1) + .setReplicas(replicas) + .setIsNew(false) + partition.makeLeader(leaderState, offsetCheckpoints, Some(topicId)) + + checkTopicId(topicId, partition) + + // Create new Partition object for same topicPartition + val partition2 = new Partition(topicPartition, + replicaLagTimeMaxMs = Defaults.ReplicaLagTimeMaxMs, + interBrokerProtocolVersion = ApiVersion.latestVersion, + localBrokerId = brokerId, + time, + isrChangeListener, + delayedOperations, + metadataCache, + logManager, + alterIsrManager) + + // partition2 should not yet be associated with the log, but should be able to get ID + assertTrue(partition2.topicId.isDefined) + assertEquals(topicId, partition2.topicId.get) + assertFalse(partition2.log.isDefined) + + // Calling makeLeader with a new topic ID should not overwrite the old topic ID. We should get an InconsistentTopicIdException. + // This scenario should not occur, since the topic ID check will fail. + assertThrows(classOf[InconsistentTopicIdException], () => partition2.makeLeader(leaderState, offsetCheckpoints, Some(Uuid.randomUuid()))) + + // Calling makeLeader with no topic ID should not overwrite the old topic ID. We should get the original log. + partition2.makeLeader(leaderState, offsetCheckpoints, None) + checkTopicId(topicId, partition2) + } + + @Test + def testTopicIdAndPartitionMetadataFileForFollower(): Unit = { + val controllerEpoch = 3 + val leaderEpoch = 5 + val topicId = Uuid.randomUuid() + val replicas = List[Integer](brokerId, brokerId + 1).asJava + val leaderState = new LeaderAndIsrPartitionState() + .setControllerEpoch(controllerEpoch) + .setLeader(brokerId) + .setLeaderEpoch(leaderEpoch) + .setIsr(replicas) + .setZkVersion(1) + .setReplicas(replicas) + .setIsNew(false) + partition.makeFollower(leaderState, offsetCheckpoints, Some(topicId)) + + checkTopicId(topicId, partition) + + // Create new Partition object for same topicPartition + val partition2 = new Partition(topicPartition, + replicaLagTimeMaxMs = Defaults.ReplicaLagTimeMaxMs, + interBrokerProtocolVersion = ApiVersion.latestVersion, + localBrokerId = brokerId, + time, + isrChangeListener, + delayedOperations, + metadataCache, + logManager, + alterIsrManager) + + // partition2 should not yet be associated with the log, but should be able to get ID + assertTrue(partition2.topicId.isDefined) + assertEquals(topicId, partition2.topicId.get) + assertFalse(partition2.log.isDefined) + + // Calling makeFollower with a new topic ID should not overwrite the old topic ID. We should get an InconsistentTopicIdException. + // This scenario should not occur, since the topic ID check will fail. + assertThrows(classOf[InconsistentTopicIdException], () => partition2.makeFollower(leaderState, offsetCheckpoints, Some(Uuid.randomUuid()))) + + // Calling makeFollower with no topic ID should not overwrite the old topic ID. We should get the original log. + partition2.makeFollower(leaderState, offsetCheckpoints, None) + checkTopicId(topicId, partition2) + } + + def checkTopicId(expectedTopicId: Uuid, partition: Partition): Unit = { + assertTrue(partition.topicId.isDefined) + assertEquals(expectedTopicId, partition.topicId.get) + assertTrue(partition.log.isDefined) + val log = partition.log.get + assertEquals(expectedTopicId, log.topicId.get) + assertTrue(log.partitionMetadataFile.exists()) + assertEquals(expectedTopicId, log.partitionMetadataFile.read().topicId) + } + @Test def testAddAndRemoveMetrics(): Unit = { val metricsToCheck = List( @@ -1674,11 +1772,11 @@ class PartitionTest extends AbstractPartitionTest { .setZkVersion(1) .setReplicas(replicas) .setIsNew(false) - partition.makeLeader(leaderState, offsetCheckpoints) + partition.makeLeader(leaderState, offsetCheckpoints, None) assertTrue(partition.isUnderReplicated) leaderState = leaderState.setIsr(replicas) - partition.makeLeader(leaderState, offsetCheckpoints) + partition.makeLeader(leaderState, offsetCheckpoints, None) assertFalse(partition.isUnderReplicated) } @@ -1738,7 +1836,7 @@ class PartitionTest extends AbstractPartitionTest { spyLogManager, alterIsrManager) - partition.createLog(isNew = true, isFutureReplica = false, offsetCheckpoints) + partition.createLog(isNew = true, isFutureReplica = false, offsetCheckpoints, topicId = None) // Validate that initializingLog and finishedInitializingLog was called verify(spyLogManager).initializingLog(ArgumentMatchers.eq(topicPartition)) @@ -1776,7 +1874,7 @@ class PartitionTest extends AbstractPartitionTest { spyLogManager, alterIsrManager) - partition.createLog(isNew = true, isFutureReplica = false, offsetCheckpoints) + partition.createLog(isNew = true, isFutureReplica = false, offsetCheckpoints, topicId = None) // Validate that initializingLog and finishedInitializingLog was called verify(spyLogManager).initializingLog(ArgumentMatchers.eq(topicPartition)) @@ -1817,7 +1915,7 @@ class PartitionTest extends AbstractPartitionTest { spyLogManager, alterIsrManager) - partition.createLog(isNew = true, isFutureReplica = false, offsetCheckpoints) + partition.createLog(isNew = true, isFutureReplica = false, offsetCheckpoints, topicId = None) // Validate that initializingLog and finishedInitializingLog was called verify(spyLogManager).initializingLog(ArgumentMatchers.eq(topicPartition)) @@ -1848,7 +1946,9 @@ class PartitionTest extends AbstractPartitionTest { log.producerIdExpirationCheckIntervalMs, log.topicPartition, log.producerStateManager, - new LogDirFailureChannel(1)) { + new LogDirFailureChannel(1), + topicId = None, + keepPartitionMetadataFile = true) { override def appendAsFollower(records: MemoryRecords): LogAppendInfo = { appendSemaphore.acquire() diff --git a/core/src/test/scala/unit/kafka/cluster/ReplicaTest.scala b/core/src/test/scala/unit/kafka/cluster/ReplicaTest.scala index 598c9e2339fae..9ec90719419c8 100644 --- a/core/src/test/scala/unit/kafka/cluster/ReplicaTest.scala +++ b/core/src/test/scala/unit/kafka/cluster/ReplicaTest.scala @@ -50,7 +50,9 @@ class ReplicaTest { time = time, maxProducerIdExpirationMs = 60 * 60 * 1000, producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs, - logDirFailureChannel = new LogDirFailureChannel(10)) + logDirFailureChannel = new LogDirFailureChannel(10), + topicId = None, + keepPartitionMetadataFile = true) } @AfterEach diff --git a/core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala b/core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala index 18a5f813a2958..99c85b6e1dd22 100644 --- a/core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala +++ b/core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala @@ -110,7 +110,9 @@ abstract class AbstractLogCleanerIntegrationTest { brokerTopicStats = new BrokerTopicStats, maxProducerIdExpirationMs = 60 * 60 * 1000, producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs, - logDirFailureChannel = new LogDirFailureChannel(10)) + logDirFailureChannel = new LogDirFailureChannel(10), + topicId = None, + keepPartitionMetadataFile = true) logMap.put(partition, log) this.logs += log } diff --git a/core/src/test/scala/unit/kafka/log/BrokerCompressionTest.scala b/core/src/test/scala/unit/kafka/log/BrokerCompressionTest.scala index 7e5197312fbee..af6265ccec5f4 100755 --- a/core/src/test/scala/unit/kafka/log/BrokerCompressionTest.scala +++ b/core/src/test/scala/unit/kafka/log/BrokerCompressionTest.scala @@ -55,7 +55,7 @@ class BrokerCompressionTest { val log = Log(logDir, LogConfig(logProps), logStartOffset = 0L, recoveryPoint = 0L, scheduler = time.scheduler, time = time, brokerTopicStats = new BrokerTopicStats, maxProducerIdExpirationMs = 60 * 60 * 1000, producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs, - logDirFailureChannel = new LogDirFailureChannel(10)) + logDirFailureChannel = new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true) /* append two messages */ log.appendAsLeader(MemoryRecords.withRecords(CompressionType.forId(messageCompressionCode.codec), 0, diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala index 267e8d8876a75..16aac94dc5df6 100644 --- a/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala @@ -98,7 +98,8 @@ class LogCleanerManagerTest extends Logging { // the exception should be catched and the partition that caused it marked as uncleanable class LogMock(dir: File, config: LogConfig) extends Log(dir, config, 0L, 0L, time.scheduler, new BrokerTopicStats, time, 60 * 60 * 1000, LogManager.ProducerIdExpirationCheckIntervalMs, - topicPartition, new ProducerStateManager(tp, tpDir, 60 * 60 * 1000), new LogDirFailureChannel(10)) { + topicPartition, new ProducerStateManager(tp, tpDir, 60 * 60 * 1000), + new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true) { // Throw an error in getFirstBatchTimestampForSegments since it is called in grabFilthiestLog() override def getFirstBatchTimestampForSegments(segments: Iterable[LogSegment]): Iterable[Long] = @@ -755,7 +756,9 @@ class LogCleanerManagerTest extends Logging { brokerTopicStats = new BrokerTopicStats, maxProducerIdExpirationMs = 60 * 60 * 1000, producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs, - logDirFailureChannel = new LogDirFailureChannel(10)) + logDirFailureChannel = new LogDirFailureChannel(10), + topicId = None, + keepPartitionMetadataFile = true) } private def createLowRetentionLogConfig(segmentSize: Int, cleanupPolicy: String): LogConfig = { @@ -799,7 +802,7 @@ class LogCleanerManagerTest extends Logging { Log(dir = dir, config = config, logStartOffset = 0L, recoveryPoint = 0L, scheduler = time.scheduler, time = time, brokerTopicStats = new BrokerTopicStats, maxProducerIdExpirationMs = 60 * 60 * 1000, producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs, - logDirFailureChannel = new LogDirFailureChannel(10)) + logDirFailureChannel = new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true) private def records(key: Int, value: Int, timestamp: Long) = MemoryRecords.withRecords(CompressionType.NONE, new SimpleRecord(timestamp, key.toString.getBytes, value.toString.getBytes)) diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala index dc51ce6d4e506..45727a0f639c5 100755 --- a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala @@ -112,7 +112,9 @@ class LogCleanerTest { producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs, topicPartition = topicPartition, producerStateManager = producerStateManager, - logDirFailureChannel = new LogDirFailureChannel(10)) { + logDirFailureChannel = new LogDirFailureChannel(10), + topicId = None, + keepPartitionMetadataFile = true) { override def replaceSegments(newSegments: Seq[LogSegment], oldSegments: Seq[LogSegment], isRecoveredSwapFile: Boolean = false): Unit = { deleteStartLatch.countDown() if (!deleteCompleteLatch.await(5000, TimeUnit.MILLISECONDS)) { @@ -1674,7 +1676,7 @@ class LogCleanerTest { Log(dir = dir, config = config, logStartOffset = 0L, recoveryPoint = recoveryPoint, scheduler = time.scheduler, time = time, brokerTopicStats = new BrokerTopicStats, maxProducerIdExpirationMs = 60 * 60 * 1000, producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs, - logDirFailureChannel = new LogDirFailureChannel(10)) + logDirFailureChannel = new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true) private def makeCleaner(capacity: Int, checkDone: TopicPartition => Unit = _ => (), maxMessageSize: Int = 64*1024) = new Cleaner(id = 0, diff --git a/core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala b/core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala index cde75ea12b7dc..3efe2bef03566 100644 --- a/core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala @@ -150,7 +150,9 @@ class LogConcurrencyTest { time = Time.SYSTEM, maxProducerIdExpirationMs = 60 * 60 * 1000, producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs, - logDirFailureChannel = new LogDirFailureChannel(10)) + logDirFailureChannel = new LogDirFailureChannel(10), + topicId = None, + keepPartitionMetadataFile = true) } private def validateConsumedData(log: Log, consumedBatches: Iterable[FetchedBatch]): Unit = { diff --git a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala index c3698dcc6ca66..b679fbb712b74 100755 --- a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala @@ -79,7 +79,7 @@ class LogManagerTest { */ @Test def testCreateLog(): Unit = { - val log = logManager.getOrCreateLog(new TopicPartition(name, 0)) + val log = logManager.getOrCreateLog(new TopicPartition(name, 0), topicId = None) assertEquals(1, logManager.liveLogDirs.size) val logFile = new File(logDir, name + "-0") @@ -104,8 +104,8 @@ class LogManagerTest { assertEquals(2, logManagerForTest.get.liveLogDirs.size) logManagerForTest.get.startup(Set.empty) - val log1 = logManagerForTest.get.getOrCreateLog(new TopicPartition(name, 0)) - val log2 = logManagerForTest.get.getOrCreateLog(new TopicPartition(name, 1)) + val log1 = logManagerForTest.get.getOrCreateLog(new TopicPartition(name, 0), topicId = None) + val log2 = logManagerForTest.get.getOrCreateLog(new TopicPartition(name, 1), topicId = None) val logFile1 = new File(logDir1, name + "-0") assertTrue(logFile1.exists) @@ -147,7 +147,7 @@ class LogManagerTest { logManager = createLogManager(dirs) logManager.startup(Set.empty) - val log = logManager.getOrCreateLog(new TopicPartition(name, 0), isNew = true) + val log = logManager.getOrCreateLog(new TopicPartition(name, 0), isNew = true, topicId = None) val logFile = new File(logDir, name + "-0") assertTrue(logFile.exists) log.appendAsLeader(TestUtils.singletonRecords("test".getBytes()), leaderEpoch = 0) @@ -179,7 +179,7 @@ class LogManagerTest { // Request creating a new log. // LogManager should try using all configured log directories until one succeeds. - logManager.getOrCreateLog(new TopicPartition(name, 0), isNew = true) + logManager.getOrCreateLog(new TopicPartition(name, 0), isNew = true, topicId = None) // Verify that half the directories were considered broken, assertEquals(dirs.length / 2, brokenDirs.size) @@ -208,7 +208,7 @@ class LogManagerTest { */ @Test def testCleanupExpiredSegments(): Unit = { - val log = logManager.getOrCreateLog(new TopicPartition(name, 0)) + val log = logManager.getOrCreateLog(new TopicPartition(name, 0), topicId = None) var offset = 0L for(_ <- 0 until 200) { val set = TestUtils.singletonRecords("test".getBytes()) @@ -254,7 +254,7 @@ class LogManagerTest { logManager.startup(Set.empty) // create a log - val log = logManager.getOrCreateLog(new TopicPartition(name, 0)) + val log = logManager.getOrCreateLog(new TopicPartition(name, 0), topicId = None) var offset = 0L // add a bunch of messages that should be larger than the retentionSize @@ -306,7 +306,7 @@ class LogManagerTest { configRepository.setTopicConfig(name, LogConfig.CleanupPolicyProp, policy) logManager = createLogManager(configRepository = configRepository) - val log = logManager.getOrCreateLog(new TopicPartition(name, 0)) + val log = logManager.getOrCreateLog(new TopicPartition(name, 0), topicId = None) var offset = 0L for (_ <- 0 until 200) { val set = TestUtils.singletonRecords("test".getBytes(), key="test".getBytes()) @@ -334,7 +334,7 @@ class LogManagerTest { logManager = createLogManager(configRepository = configRepository) logManager.startup(Set.empty) - val log = logManager.getOrCreateLog(new TopicPartition(name, 0)) + val log = logManager.getOrCreateLog(new TopicPartition(name, 0), topicId = None) val lastFlush = log.lastFlushTime for (_ <- 0 until 200) { val set = TestUtils.singletonRecords("test".getBytes()) @@ -358,7 +358,7 @@ class LogManagerTest { // verify that logs are always assigned to the least loaded partition for(partition <- 0 until 20) { - logManager.getOrCreateLog(new TopicPartition("test", partition)) + logManager.getOrCreateLog(new TopicPartition("test", partition), topicId = None) assertEquals(partition + 1, logManager.allLogs.size, "We should have created the right number of logs") val counts = logManager.allLogs.groupBy(_.dir.getParent).values.map(_.size) assertTrue(counts.max <= counts.min + 1, "Load should balance evenly") @@ -404,7 +404,7 @@ class LogManagerTest { } private def verifyCheckpointRecovery(topicPartitions: Seq[TopicPartition], logManager: LogManager, logDir: File): Unit = { - val logs = topicPartitions.map(logManager.getOrCreateLog(_)) + val logs = topicPartitions.map(logManager.getOrCreateLog(_, topicId = None)) logs.foreach { log => for (_ <- 0 until 50) log.appendAsLeader(TestUtils.singletonRecords("test".getBytes()), leaderEpoch = 0) @@ -431,7 +431,7 @@ class LogManagerTest { @Test def testFileReferencesAfterAsyncDelete(): Unit = { - val log = logManager.getOrCreateLog(new TopicPartition(name, 0)) + val log = logManager.getOrCreateLog(new TopicPartition(name, 0), topicId = None) val activeSegment = log.activeSegment val logName = activeSegment.log.file.getName val indexName = activeSegment.offsetIndex.file.getName @@ -468,7 +468,7 @@ class LogManagerTest { @Test def testCreateAndDeleteOverlyLongTopic(): Unit = { val invalidTopicName = String.join("", Collections.nCopies(253, "x")) - logManager.getOrCreateLog(new TopicPartition(invalidTopicName, 0)) + logManager.getOrCreateLog(new TopicPartition(invalidTopicName, 0), topicId = None) logManager.asyncDelete(new TopicPartition(invalidTopicName, 0)) } @@ -481,7 +481,7 @@ class LogManagerTest { new TopicPartition("test-b", 0), new TopicPartition("test-b", 1)) - val allLogs = tps.map(logManager.getOrCreateLog(_)) + val allLogs = tps.map(logManager.getOrCreateLog(_, topicId = None)) allLogs.foreach { log => for (_ <- 0 until 50) log.appendAsLeader(TestUtils.singletonRecords("test".getBytes), leaderEpoch = 0) @@ -616,7 +616,7 @@ class LogManagerTest { } // Create the Log and assert that the metrics are present - logManager.getOrCreateLog(tp) + logManager.getOrCreateLog(tp, topicId = None) verifyMetrics() // Trigger the deletion and assert that the metrics have been removed @@ -624,7 +624,7 @@ class LogManagerTest { assertTrue(logMetrics.isEmpty) // Recreate the Log and assert that the metrics are present - logManager.getOrCreateLog(tp) + logManager.getOrCreateLog(tp, topicId = None) verifyMetrics() // Advance time past the file deletion delay and assert that the removed log has been deleted but the metrics @@ -657,9 +657,9 @@ class LogManagerTest { // Create the current and future logs and verify that metrics are present for both current and future logs logManager.maybeUpdatePreferredLogDir(tp, dir1.getAbsolutePath) - logManager.getOrCreateLog(tp) + logManager.getOrCreateLog(tp, topicId = None) logManager.maybeUpdatePreferredLogDir(tp, dir2.getAbsolutePath) - logManager.getOrCreateLog(tp, isFuture = true) + logManager.getOrCreateLog(tp, isFuture = true, topicId = None) verifyMetrics(2) // Replace the current log with the future one and verify that only one set of metrics are present diff --git a/core/src/test/scala/unit/kafka/log/LogTest.scala b/core/src/test/scala/unit/kafka/log/LogTest.scala index 1e5257df49853..42974d165e277 100755 --- a/core/src/test/scala/unit/kafka/log/LogTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogTest.scala @@ -110,7 +110,7 @@ class LogTest { val producerStateManager = new ProducerStateManager(topicPartition, logDir, maxPidExpirationMs) val log = new Log(logDir, config, logStartOffset, logRecoveryPoint, time.scheduler, brokerTopicStats, time, maxPidExpirationMs, - LogManager.ProducerIdExpirationCheckIntervalMs, topicPartition, producerStateManager, logDirFailureChannel, hadCleanShutdown) { + LogManager.ProducerIdExpirationCheckIntervalMs, topicPartition, producerStateManager, logDirFailureChannel, hadCleanShutdown, None, true) { override def recoverLog(): Long = { if (simulateError) throw new RuntimeException @@ -127,7 +127,7 @@ class LogTest { val cleanShutdownFile = new File(logDir, Log.CleanShutdownFile) val logManager: LogManager = interceptedLogManager(logConfig, logDirs) - log = logManager.getOrCreateLog(topicPartition, isNew = true) + log = logManager.getOrCreateLog(topicPartition, isNew = true, topicId = None) // Load logs after a clean shutdown Files.createFile(cleanShutdownFile.toPath) @@ -1025,7 +1025,7 @@ class LogTest { // Intercept all segment read calls new Log(logDir, logConfig, logStartOffset = 0, recoveryPoint = recoveryPoint, mockTime.scheduler, brokerTopicStats, mockTime, maxProducerIdExpirationMs, LogManager.ProducerIdExpirationCheckIntervalMs, - topicPartition, producerStateManager, new LogDirFailureChannel(10), hadCleanShutdown = false) { + topicPartition, producerStateManager, new LogDirFailureChannel(10), hadCleanShutdown = false, topicId = None, keepPartitionMetadataFile = true) { override def addSegment(segment: LogSegment): LogSegment = { val wrapper = new LogSegment(segment.log, segment.lazyOffsetIndex, segment.lazyTimeIndex, segment.txnIndex, segment.baseOffset, @@ -1128,7 +1128,9 @@ class LogTest { topicPartition = Log.parseTopicPartitionName(logDir), producerStateManager = stateManager, logDirFailureChannel = new LogDirFailureChannel(1), - hadCleanShutdown = false) + hadCleanShutdown = false, + topicId = None, + keepPartitionMetadataFile = true) EasyMock.verify(stateManager) @@ -1206,7 +1208,9 @@ class LogTest { producerIdExpirationCheckIntervalMs = 30000, topicPartition = Log.parseTopicPartitionName(logDir), producerStateManager = stateManager, - logDirFailureChannel = null) + logDirFailureChannel = null, + topicId = None, + keepPartitionMetadataFile = true) EasyMock.verify(stateManager) } @@ -1244,7 +1248,9 @@ class LogTest { producerIdExpirationCheckIntervalMs = 30000, topicPartition = Log.parseTopicPartitionName(logDir), producerStateManager = stateManager, - logDirFailureChannel = null) + logDirFailureChannel = null, + topicId = None, + keepPartitionMetadataFile = true) EasyMock.verify(stateManager) } @@ -1284,7 +1290,9 @@ class LogTest { producerIdExpirationCheckIntervalMs = 30000, topicPartition = Log.parseTopicPartitionName(logDir), producerStateManager = stateManager, - logDirFailureChannel = null) + logDirFailureChannel = null, + topicId = None, + keepPartitionMetadataFile = true) EasyMock.verify(stateManager) } @@ -2559,10 +2567,29 @@ class LogTest { // test recovery case log = createLog(logDir, logConfig) - assertTrue(log.topicId == topicId) + assertTrue(log.topicId.isDefined) + assertTrue(log.topicId.get == topicId) log.close() } + @Test + def testLogFailsWhenInconsistentTopicIdSet(): Unit = { + val logConfig = LogTest.createLogConfig() + var log = createLog(logDir, logConfig) + + val topicId = Uuid.randomUuid() + log.partitionMetadataFile.write(topicId) + log.close() + + // test creating a log with a new ID + try { + log = createLog(logDir, logConfig, topicId = Some(Uuid.randomUuid())) + log.close() + } catch { + case e: Throwable => assertTrue(e.isInstanceOf[InconsistentTopicIdException]) + } + } + /** * Test building the time index on the follower by setting assignOffsets to false. */ @@ -3123,7 +3150,7 @@ class LogTest { // Write a topic ID to the partition metadata file to ensure it is transferred correctly. val id = Uuid.randomUuid() - log.topicId = id + log.topicId = Some(id) log.partitionMetadataFile.write(id) log.appendAsLeader(TestUtils.records(List(new SimpleRecord("foo".getBytes()))), leaderEpoch = 5) @@ -3138,7 +3165,8 @@ class LogTest { assertFalse(PartitionMetadataFile.newFile(this.logDir).exists()) // Check the topic ID remains in memory and was copied correctly. - assertEquals(id, log.topicId) + assertTrue(log.topicId.isDefined) + assertEquals(id, log.topicId.get) assertEquals(id, log.partitionMetadataFile.read().topicId) } @@ -4853,9 +4881,10 @@ class LogTest { time: Time = mockTime, maxProducerIdExpirationMs: Int = 60 * 60 * 1000, producerIdExpirationCheckIntervalMs: Int = LogManager.ProducerIdExpirationCheckIntervalMs, - lastShutdownClean: Boolean = true): Log = { + lastShutdownClean: Boolean = true, + topicId: Option[Uuid] = None): Log = { LogTest.createLog(dir, config, brokerTopicStats, scheduler, time, logStartOffset, recoveryPoint, - maxProducerIdExpirationMs, producerIdExpirationCheckIntervalMs, lastShutdownClean) + maxProducerIdExpirationMs, producerIdExpirationCheckIntervalMs, lastShutdownClean, topicId = topicId) } private def createLogWithOffsetOverflow(logConfig: LogConfig): (Log, LogSegment) = { @@ -4921,7 +4950,8 @@ object LogTest { recoveryPoint: Long = 0L, maxProducerIdExpirationMs: Int = 60 * 60 * 1000, producerIdExpirationCheckIntervalMs: Int = LogManager.ProducerIdExpirationCheckIntervalMs, - lastShutdownClean: Boolean = true): Log = { + lastShutdownClean: Boolean = true, + topicId: Option[Uuid] = None): Log = { Log(dir = dir, config = config, logStartOffset = logStartOffset, @@ -4932,7 +4962,9 @@ object LogTest { maxProducerIdExpirationMs = maxProducerIdExpirationMs, producerIdExpirationCheckIntervalMs = producerIdExpirationCheckIntervalMs, logDirFailureChannel = new LogDirFailureChannel(10), - lastShutdownClean = lastShutdownClean) + lastShutdownClean = lastShutdownClean, + topicId = topicId, + keepPartitionMetadataFile = true) } /** diff --git a/core/src/test/scala/unit/kafka/server/BrokerEpochIntegrationTest.scala b/core/src/test/scala/unit/kafka/server/BrokerEpochIntegrationTest.scala index cbb6471fe2628..46be884349197 100755 --- a/core/src/test/scala/unit/kafka/server/BrokerEpochIntegrationTest.scala +++ b/core/src/test/scala/unit/kafka/server/BrokerEpochIntegrationTest.scala @@ -25,7 +25,7 @@ import kafka.controller.{ControllerChannelManager, ControllerContext, StateChang import kafka.utils.TestUtils import kafka.utils.TestUtils.createTopic import kafka.zk.ZooKeeperTestHarness -import org.apache.kafka.common.{TopicPartition, Uuid} +import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState import org.apache.kafka.common.message.StopReplicaRequestData.{StopReplicaPartitionState, StopReplicaTopicState} import org.apache.kafka.common.message.UpdateMetadataRequestData.{UpdateMetadataBroker, UpdateMetadataEndpoint, UpdateMetadataPartitionState} @@ -112,10 +112,10 @@ class BrokerEpochIntegrationTest extends ZooKeeperTestHarness { private def testControlRequestWithBrokerEpoch(epochInRequestDiffFromCurrentEpoch: Long): Unit = { val tp = new TopicPartition("new-topic", 0) - val topicIds = Collections.singletonMap("new-topic", Uuid.randomUuid) // create topic with 1 partition, 2 replicas, one on each broker createTopic(zkClient, tp.topic(), partitionReplicaAssignment = Map(0 -> Seq(brokerId1, brokerId2)), servers = servers) + val topicIds = getController.kafkaController.controllerContext.topicIds.toMap.asJava val controllerId = 2 val controllerEpoch = zkClient.getControllerEpoch.get._1 diff --git a/core/src/test/scala/unit/kafka/server/HighwatermarkPersistenceTest.scala b/core/src/test/scala/unit/kafka/server/HighwatermarkPersistenceTest.scala index c39ae7cf03e4f..d7a16d22cc4c7 100755 --- a/core/src/test/scala/unit/kafka/server/HighwatermarkPersistenceTest.scala +++ b/core/src/test/scala/unit/kafka/server/HighwatermarkPersistenceTest.scala @@ -75,7 +75,7 @@ class HighwatermarkPersistenceTest { val tp0 = new TopicPartition(topic, 0) val partition0 = replicaManager.createPartition(tp0) // create leader and follower replicas - val log0 = logManagers.head.getOrCreateLog(new TopicPartition(topic, 0)) + val log0 = logManagers.head.getOrCreateLog(new TopicPartition(topic, 0), topicId = None) partition0.setLog(log0, isFutureLog = false) partition0.updateAssignmentAndIsr( @@ -124,7 +124,7 @@ class HighwatermarkPersistenceTest { val t1p0 = new TopicPartition(topic1, 0) val topic1Partition0 = replicaManager.createPartition(t1p0) // create leader log - val topic1Log0 = logManagers.head.getOrCreateLog(t1p0) + val topic1Log0 = logManagers.head.getOrCreateLog(t1p0, topicId = None) // create a local replica for topic1 topic1Partition0.setLog(topic1Log0, isFutureLog = false) replicaManager.checkpointHighWatermarks() @@ -141,7 +141,7 @@ class HighwatermarkPersistenceTest { val t2p0 = new TopicPartition(topic2, 0) val topic2Partition0 = replicaManager.createPartition(t2p0) // create leader log - val topic2Log0 = logManagers.head.getOrCreateLog(t2p0) + val topic2Log0 = logManagers.head.getOrCreateLog(t2p0, topicId = None) // create a local replica for topic2 topic2Partition0.setLog(topic2Log0, isFutureLog = false) replicaManager.checkpointHighWatermarks() diff --git a/core/src/test/scala/unit/kafka/server/LogOffsetTest.scala b/core/src/test/scala/unit/kafka/server/LogOffsetTest.scala index 746374583434d..bf9ad3e435504 100755 --- a/core/src/test/scala/unit/kafka/server/LogOffsetTest.scala +++ b/core/src/test/scala/unit/kafka/server/LogOffsetTest.scala @@ -159,7 +159,7 @@ class LogOffsetTest extends BaseRequestTest { createTopic(topic, 3, 1) val logManager = server.getLogManager - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) for (_ <- 0 until 20) log.appendAsLeader(TestUtils.singletonRecords(value = Integer.toString(42).getBytes()), leaderEpoch = 0) @@ -188,7 +188,7 @@ class LogOffsetTest extends BaseRequestTest { createTopic(topic, 3, 1) val logManager = server.getLogManager - val log = logManager.getOrCreateLog(topicPartition) + val log = logManager.getOrCreateLog(topicPartition, topicId = None) for (_ <- 0 until 20) log.appendAsLeader(TestUtils.singletonRecords(value = Integer.toString(42).getBytes()), leaderEpoch = 0) log.flush() diff --git a/core/src/test/scala/unit/kafka/server/RaftReplicaChangeDelegateTest.scala b/core/src/test/scala/unit/kafka/server/RaftReplicaChangeDelegateTest.scala index 609f1cf3b3625..ad35452a3b0e0 100644 --- a/core/src/test/scala/unit/kafka/server/RaftReplicaChangeDelegateTest.scala +++ b/core/src/test/scala/unit/kafka/server/RaftReplicaChangeDelegateTest.scala @@ -24,7 +24,7 @@ import kafka.server.checkpoints.OffsetCheckpoints import kafka.server.metadata.{MetadataBroker, MetadataBrokers, MetadataPartition} import org.apache.kafka.common.message.LeaderAndIsrRequestData import org.apache.kafka.common.network.ListenerName -import org.apache.kafka.common.{Node, TopicPartition} +import org.apache.kafka.common.{Node, TopicPartition, Uuid} import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.params.ParameterizedTest import org.junit.jupiter.params.provider.ValueSource @@ -43,10 +43,13 @@ class RaftReplicaChangeDelegateTest { val leaderId = 0 val topicPartition = new TopicPartition("foo", 5) val replicas = Seq(0, 1, 2).map(Int.box).asJava + val topicId = Uuid.randomUuid() + val topicIds = (topicName: String) => if (topicName == "foo") Some(topicId) else None val helper = mockedHelper() val partition = mock(classOf[Partition]) when(partition.topicPartition).thenReturn(topicPartition) + when(partition.topic).thenReturn(topicPartition.topic) val highWatermarkCheckpoints = mock(classOf[OffsetCheckpoints]) when(highWatermarkCheckpoints.fetch( @@ -81,16 +84,17 @@ class RaftReplicaChangeDelegateTest { val delegate = new RaftReplicaChangeDelegate(helper) val updatedPartitions = if (isLeader) { - when(partition.makeLeader(expectedLeaderAndIsr, highWatermarkCheckpoints)) + when(partition.makeLeader(expectedLeaderAndIsr, highWatermarkCheckpoints, Some(topicId))) .thenReturn(true) delegate.makeLeaders( prevPartitionsAlreadyExisting = Set.empty, partitionStates = Map(partition -> metadataPartition), highWatermarkCheckpoints, - metadataOffset = Some(500) + metadataOffset = Some(500), + topicIds ) } else { - when(partition.makeFollower(expectedLeaderAndIsr, highWatermarkCheckpoints)) + when(partition.makeFollower(expectedLeaderAndIsr, highWatermarkCheckpoints, Some(topicId))) .thenReturn(true) when(partition.leaderReplicaIdOpt).thenReturn(Some(leaderId)) delegate.makeFollowers( @@ -98,7 +102,8 @@ class RaftReplicaChangeDelegateTest { currentBrokers = aliveBrokers(replicas), partitionStates = Map(partition -> metadataPartition), highWatermarkCheckpoints, - metadataOffset = Some(500) + metadataOffset = Some(500), + topicIds ) } diff --git a/core/src/test/scala/unit/kafka/server/RaftReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/RaftReplicaManagerTest.scala index d3d6471712663..a37214b605fc7 100644 --- a/core/src/test/scala/unit/kafka/server/RaftReplicaManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/RaftReplicaManagerTest.scala @@ -23,8 +23,10 @@ import java.util.concurrent.atomic.AtomicBoolean import kafka.cluster.Partition import kafka.server.QuotaFactory.QuotaManagers +import kafka.server.checkpoints.LazyOffsetCheckpoints import kafka.server.metadata.{CachedConfigRepository, MetadataBroker, MetadataBrokers, MetadataImage, MetadataImageBuilder, MetadataPartition, RaftMetadataCache} import kafka.utils.{MockScheduler, MockTime, TestUtils} +import org.apache.kafka.common.errors.InconsistentTopicIdException import org.apache.kafka.common.{TopicPartition, Uuid} import org.apache.kafka.common.metadata.PartitionRecord import org.apache.kafka.common.metrics.Metrics @@ -138,7 +140,7 @@ class RaftReplicaManagerTest { 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()) + verify(mockDelegate, never()).makeFollowers(any(), 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) @@ -146,8 +148,15 @@ class RaftReplicaManagerTest { // 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)) + when(mockDelegate.makeLeaders(any(), any(), any(), any(), any())).thenReturn(Set(partition0)) + when(mockDelegate.makeFollowers(any(), any(), any(), any(), any(), any())).thenReturn(Set(partition1)) + + // Simulate creation of logs in makeLeaders/makeFollowers + partition0.createLogIfNotExists(isNew = false, isFutureReplica = false, + new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints), Some(topicId)) + partition1.createLogIfNotExists(isNew = false, isFutureReplica = false, + new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints), Some(topicId)) + rrm.endMetadataChangeDeferral(onLeadershipChange) // verify that the deferred changes would have been applied @@ -161,6 +170,9 @@ class RaftReplicaManagerTest { // leadership change callbacks verifyLeadershipChangeCallbacks(List(partition0), List(partition1)) + + // partition.metadata file + verifyPartitionMetadataFile(rrm, List(topicPartition0, topicPartition1)) } @Test @@ -169,11 +181,19 @@ class RaftReplicaManagerTest { 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)) + when(mockDelegate.makeLeaders(any(), any(), any(), ArgumentMatchers.eq(Some(offset1)), any())).thenReturn(Set(partition0)) + when(mockDelegate.makeFollowers(any(), any(), any(), any(), ArgumentMatchers.eq(Some(offset1)), any())).thenReturn(Set(partition1)) + + // Simulate creation of logs in makeLeaders/makeFollowers + partition0.createLogIfNotExists(isNew = false, isFutureReplica = false, + new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints), Some(topicId)) + partition1.createLogIfNotExists(isNew = false, isFutureReplica = false, + new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints), Some(topicId)) + // process the changes processTopicPartitionMetadata(rrm) // verify that the changes would have been applied @@ -188,6 +208,53 @@ class RaftReplicaManagerTest { // leadership change callbacks verifyLeadershipChangeCallbacks(List(partition0), List(partition1)) + + // partition.metadata file + verifyPartitionMetadataFile(rrm, List(topicPartition0, topicPartition1)) + } + + @Test + def testInconsistentTopicIdDefersChanges(): Unit = { + val rrm = createRaftReplicaManager() + rrm.delegate = mockDelegate + val partition0 = rrm.createPartition(topicPartition0) + val partition1 = rrm.createPartition(topicPartition1) + + partition0.createLogIfNotExists(isNew = false, isFutureReplica = false, + new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints), Some(Uuid.randomUuid())) + partition1.createLogIfNotExists(isNew = false, isFutureReplica = false, + new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints), Some(Uuid.randomUuid())) + + try { + processTopicPartitionMetadata(rrm) + } catch { + case e: Throwable => assertTrue(e.isInstanceOf[InconsistentTopicIdException]) + } + } + + @Test + def testInconsistentTopicIdWhenNotDeferring(): Unit = { + val rrm = createRaftReplicaManager() + rrm.delegate = mockDelegate + val partition0 = rrm.createPartition(topicPartition0) + val partition1 = rrm.createPartition(topicPartition1) + + partition0.createLogIfNotExists(isNew = false, isFutureReplica = false, + new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints), Some(Uuid.randomUuid())) + partition1.createLogIfNotExists(isNew = false, isFutureReplica = false, + new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints), Some(Uuid.randomUuid())) + + rrm.endMetadataChangeDeferral(onLeadershipChange) + + // define some return values to avoid NPE + when(mockDelegate.makeLeaders(any(), any(), any(), ArgumentMatchers.eq(Some(offset1)), any())).thenReturn(Set(partition0)) + when(mockDelegate.makeFollowers(any(), any(), any(), any(), ArgumentMatchers.eq(Some(offset1)), any())).thenReturn(Set(partition1)) + + try { + processTopicPartitionMetadata(rrm) + } catch { + case e: Throwable => assertTrue(e.isInstanceOf[InconsistentTopicIdException]) + } } private def verifyMakeLeaders(expectedPrevPartitionsAlreadyExisting: Set[MetadataPartition], @@ -195,7 +262,7 @@ class RaftReplicaManagerTest { 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.capture(), any(), ArgumentMatchers.eq(expectedMetadataOffset), any()) leaderPartitionStatesCaptor.getValue } @@ -206,7 +273,7 @@ class RaftReplicaManagerTest { 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)) + followerPartitionStatesCaptor.capture(), any(), ArgumentMatchers.eq(expectedMetadataOffset), any()) val brokers = brokersCaptor.getValue assertEquals(expectedBrokers.size, brokers.size()) expectedBrokers.foreach(brokerId => assertTrue(brokers.aliveBroker(brokerId).isDefined)) @@ -221,6 +288,18 @@ class RaftReplicaManagerTest { assertEquals(expectedUpdatedFollowers, updatedFollowersCaptor.getValue.toList) } + private def verifyPartitionMetadataFile(rrm: RaftReplicaManager, topicPartitions: List[TopicPartition]) = { + topicPartitions.foreach ( topicPartition => { + val log = rrm.getLog(topicPartition).get + assertTrue(log.partitionMetadataFile.exists()) + val partitionMetadata = log.partitionMetadataFile.read() + + // Current version of PartitionMetadataFile is 0. + assertEquals(0, partitionMetadata.version) + assertEquals(topicId, partitionMetadata.topicId) + }) + } + private def processTopicPartitionMetadata(raftReplicaManager: RaftReplicaManager): Unit = { // create brokers imageBuilder.brokersBuilder().add(metadataBroker0) diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala index 76ce6c7a5f145..ced71864bc892 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala @@ -97,7 +97,7 @@ class ReplicaManagerTest { try { val partition = rm.createPartition(new TopicPartition(topic, 1)) partition.createLogIfNotExists(isNew = false, isFutureReplica = false, - new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints)) + new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints), None) rm.checkpointHighWatermarks() } finally { // shutdown the replica manager upon test completion @@ -117,7 +117,7 @@ class ReplicaManagerTest { try { val partition = rm.createPartition(new TopicPartition(topic, 1)) partition.createLogIfNotExists(isNew = false, isFutureReplica = false, - new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints)) + new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints), None) rm.checkpointHighWatermarks() } finally { // shutdown the replica manager upon test completion @@ -171,7 +171,7 @@ class ReplicaManagerTest { val partition = rm.createPartition(new TopicPartition(topic, 0)) partition.createLogIfNotExists(isNew = false, isFutureReplica = false, - new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints)) + new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints), None) // Make this replica the leader. val leaderAndIsrRequest1 = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, Seq(new LeaderAndIsrPartitionState() @@ -231,7 +231,7 @@ class ReplicaManagerTest { val topicPartition = new TopicPartition(topic, 0) replicaManager.createPartition(topicPartition) .createLogIfNotExists(isNew = false, isFutureReplica = false, - new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)) + new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints), None) val topicIds = Collections.singletonMap(topic, Uuid.randomUuid()) def leaderAndIsrRequest(epoch: Int): LeaderAndIsrRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, @@ -291,7 +291,7 @@ class ReplicaManagerTest { val partition = replicaManager.createPartition(new TopicPartition(topic, 0)) partition.createLogIfNotExists(isNew = false, isFutureReplica = false, - new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)) + new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints), None) // Make this replica the leader. val leaderAndIsrRequest1 = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, @@ -352,7 +352,7 @@ class ReplicaManagerTest { val partition = replicaManager.createPartition(new TopicPartition(topic, 0)) partition.createLogIfNotExists(isNew = false, isFutureReplica = false, - new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)) + new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints), None) // Make this replica the leader. val leaderAndIsrRequest1 = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, @@ -459,7 +459,7 @@ class ReplicaManagerTest { val brokerList = Seq[Integer](0, 1).asJava val partition = replicaManager.createPartition(new TopicPartition(topic, 0)) partition.createLogIfNotExists(isNew = false, isFutureReplica = false, - new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)) + new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints), None) // Make this replica the leader. val leaderAndIsrRequest1 = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, @@ -536,7 +536,7 @@ class ReplicaManagerTest { val partition = rm.createPartition(new TopicPartition(topic, 0)) partition.createLogIfNotExists(isNew = false, isFutureReplica = false, - new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints)) + new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints), None) // Make this replica the leader. val leaderAndIsrRequest1 = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, @@ -696,8 +696,8 @@ class ReplicaManagerTest { val tp0 = new TopicPartition(topic, 0) val tp1 = new TopicPartition(topic, 1) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) - replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) - replicaManager.createPartition(tp1).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) + replicaManager.createPartition(tp1).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val partition0Replicas = Seq[Integer](0, 1).asJava val partition1Replicas = Seq[Integer](0, 2).asJava val topicIds = Map(tp0.topic -> Uuid.randomUuid(), tp1.topic -> Uuid.randomUuid()).asJava @@ -809,6 +809,7 @@ class ReplicaManagerTest { private def verifyBecomeFollowerWhenLeaderIsUnchangedButMissedLeaderUpdate(extraProps: Properties, expectTruncation: Boolean): Unit = { val topicPartition = 0 + val topicId = Uuid.randomUuid() val followerBrokerId = 0 val leaderBrokerId = 1 val controllerId = 0 @@ -821,16 +822,17 @@ class ReplicaManagerTest { // Prepare the mocked components for the test val (replicaManager, mockLogMgr) = prepareReplicaManagerAndLogManager(new MockTimer(time), topicPartition, leaderEpoch + leaderEpochIncrement, followerBrokerId, leaderBrokerId, countDownLatch, - expectTruncation = expectTruncation, localLogOffset = Some(10), extraProps = extraProps) + expectTruncation = expectTruncation, localLogOffset = Some(10), extraProps = extraProps, topicId = Some(topicId)) // Initialize partition state to follower, with leader = 1, leaderEpoch = 1 val tp = new TopicPartition(topic, topicPartition) val partition = replicaManager.createPartition(tp) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) partition.makeFollower( leaderAndIsrPartitionState(tp, leaderEpoch, leaderBrokerId, aliveBrokerIds), - offsetCheckpoints) + offsetCheckpoints, + None) // Make local partition a follower - because epoch increased by more than 1, truncation should // trigger even though leader does not change @@ -838,7 +840,7 @@ class ReplicaManagerTest { val leaderAndIsrRequest0 = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, controllerId, controllerEpoch, brokerEpoch, Seq(leaderAndIsrPartitionState(tp, leaderEpoch, leaderBrokerId, aliveBrokerIds)).asJava, - Collections.singletonMap(topic, Uuid.randomUuid()), + Collections.singletonMap(topic, topicId), Set(new Node(followerBrokerId, "host1", 0), new Node(leaderBrokerId, "host2", 1)).asJava).build() replicaManager.becomeLeaderOrFollower(correlationId, leaderAndIsrRequest0, @@ -868,11 +870,11 @@ class ReplicaManagerTest { val partition = replicaManager.createPartition(tp) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) partition.makeLeader( leaderAndIsrPartitionState(tp, leaderEpoch, leaderBrokerId, aliveBrokerIds), - offsetCheckpoints - ) + offsetCheckpoints, + None) val metadata: ClientMetadata = new DefaultClientMetadata("rack-a", "client-id", InetAddress.getByName("localhost"), KafkaPrincipal.ANONYMOUS, "default") @@ -886,6 +888,7 @@ class ReplicaManagerTest { @Test def testPreferredReplicaAsFollower(): Unit = { val topicPartition = 0 + val topicId = Uuid.randomUuid() val followerBrokerId = 0 val leaderBrokerId = 1 val leaderEpoch = 1 @@ -895,13 +898,13 @@ class ReplicaManagerTest { // Prepare the mocked components for the test val (replicaManager, _) = prepareReplicaManagerAndLogManager(new MockTimer(time), topicPartition, leaderEpoch + leaderEpochIncrement, followerBrokerId, - leaderBrokerId, countDownLatch, expectTruncation = true) + leaderBrokerId, countDownLatch, expectTruncation = true, topicId = Some(topicId)) val brokerList = Seq[Integer](0, 1).asJava val tp0 = new TopicPartition(topic, 0) - replicaManager.createPartition(new TopicPartition(topic, 0)) + initializeLogAndTopicId(replicaManager, tp0, topicId) // Make this replica the follower val leaderAndIsrRequest2 = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, @@ -915,7 +918,7 @@ class ReplicaManagerTest { .setZkVersion(0) .setReplicas(brokerList) .setIsNew(false)).asJava, - Collections.singletonMap(topic, Uuid.randomUuid()), + Collections.singletonMap(topic, topicId), Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build() replicaManager.becomeLeaderOrFollower(1, leaderAndIsrRequest2, (_, _) => ()) @@ -936,6 +939,7 @@ class ReplicaManagerTest { @Test def testPreferredReplicaAsLeader(): Unit = { val topicPartition = 0 + val topicId = Uuid.randomUuid() val followerBrokerId = 0 val leaderBrokerId = 1 val leaderEpoch = 1 @@ -945,13 +949,13 @@ class ReplicaManagerTest { // Prepare the mocked components for the test val (replicaManager, _) = prepareReplicaManagerAndLogManager(new MockTimer(time), topicPartition, leaderEpoch + leaderEpochIncrement, followerBrokerId, - leaderBrokerId, countDownLatch, expectTruncation = true) + leaderBrokerId, countDownLatch, expectTruncation = true, topicId = Some(topicId)) val brokerList = Seq[Integer](0, 1).asJava val tp0 = new TopicPartition(topic, 0) - replicaManager.createPartition(new TopicPartition(topic, 0)) + initializeLogAndTopicId(replicaManager, tp0, topicId) // Make this replica the follower val leaderAndIsrRequest2 = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, @@ -965,7 +969,7 @@ class ReplicaManagerTest { .setZkVersion(0) .setReplicas(brokerList) .setIsNew(false)).asJava, - Collections.singletonMap(topic, Uuid.randomUuid()), + Collections.singletonMap(topic, topicId), Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build() replicaManager.becomeLeaderOrFollower(1, leaderAndIsrRequest2, (_, _) => ()) @@ -986,6 +990,7 @@ class ReplicaManagerTest { @Test def testFollowerFetchWithDefaultSelectorNoForcedHwPropagation(): Unit = { val topicPartition = 0 + val topicId = Uuid.randomUuid() val followerBrokerId = 0 val leaderBrokerId = 1 val leaderEpoch = 1 @@ -996,13 +1001,13 @@ class ReplicaManagerTest { // Prepare the mocked components for the test val (replicaManager, _) = prepareReplicaManagerAndLogManager(timer, topicPartition, leaderEpoch + leaderEpochIncrement, followerBrokerId, - leaderBrokerId, countDownLatch, expectTruncation = true) + leaderBrokerId, countDownLatch, expectTruncation = true, topicId = Some(topicId)) val brokerList = Seq[Integer](0, 1).asJava val tp0 = new TopicPartition(topic, 0) - replicaManager.createPartition(new TopicPartition(topic, 0)) + initializeLogAndTopicId(replicaManager, tp0, topicId) // Make this replica the follower val leaderAndIsrRequest2 = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, @@ -1016,7 +1021,7 @@ class ReplicaManagerTest { .setZkVersion(0) .setReplicas(brokerList) .setIsNew(false)).asJava, - Collections.singletonMap(topic, Uuid.randomUuid()), + Collections.singletonMap(topic, topicId), Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build() replicaManager.becomeLeaderOrFollower(1, leaderAndIsrRequest2, (_, _) => ()) @@ -1064,6 +1069,16 @@ class ReplicaManagerTest { leaderBrokerId, countDownLatch, expectTruncation = true, extraProps = props)) } + // Due to some limitations to EasyMock, we need to create the log so that the Partition.topicId does not call + // LogManager.getLog with a default argument + // TODO: convert tests to using Mockito to avoid this issue. + private def initializeLogAndTopicId(replicaManager: ReplicaManager, topicPartition: TopicPartition, topicId: Uuid): Unit = { + val partition = replicaManager.createPartition(new TopicPartition(topic, 0)) + val log = replicaManager.logManager.getOrCreateLog(topicPartition, false, false, Some(topicId)) + log.topicId = Some(topicId) + partition.log = Some(log) + } + @Test def testDefaultReplicaSelector(): Unit = { val topicPartition = 0 @@ -1085,7 +1100,7 @@ class ReplicaManagerTest { val tp0 = new TopicPartition(topic, 0) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) - replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val partition0Replicas = Seq[Integer](0, 1).asJava val becomeFollowerRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, Seq(new LeaderAndIsrPartitionState() @@ -1125,7 +1140,7 @@ class ReplicaManagerTest { val tp0 = new TopicPartition(topic, 0) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) - replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val partition0Replicas = Seq[Integer](0, 1).asJava val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, @@ -1171,7 +1186,7 @@ class ReplicaManagerTest { val tp0 = new TopicPartition(topic, 0) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) - replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val partition0Replicas = Seq[Integer](0, 1).asJava val topicIds = Collections.singletonMap(tp0.topic, Uuid.randomUuid()) @@ -1222,7 +1237,7 @@ class ReplicaManagerTest { val tp0 = new TopicPartition(topic, 0) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) - replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val partition0Replicas = Seq[Integer](0, 1).asJava val topicIds = Collections.singletonMap(tp0.topic, Uuid.randomUuid()) @@ -1273,7 +1288,7 @@ class ReplicaManagerTest { val tp0 = new TopicPartition(topic, 0) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) - replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val partition0Replicas = Seq[Integer](0, 1).asJava val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, @@ -1316,7 +1331,7 @@ class ReplicaManagerTest { val tp0 = new TopicPartition(topic, 0) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) - replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val partition0Replicas = Seq[Integer](0, 1).asJava val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, @@ -1360,7 +1375,7 @@ class ReplicaManagerTest { val tp0 = new TopicPartition(topic, 0) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) - replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val partition0Replicas = Seq[Integer](0, 1).asJava val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, @@ -1456,7 +1471,8 @@ class ReplicaManagerTest { localLogOffset: Option[Long] = None, offsetFromLeader: Long = 5, leaderEpochFromLeader: Int = 3, - extraProps: Properties = new Properties()) : (ReplicaManager, LogManager) = { + extraProps: Properties = new Properties(), + topicId: Option[Uuid] = None) : (ReplicaManager, LogManager) = { val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect) props.put("log.dir", TestUtils.tempRelativeDir("data").getAbsolutePath) props.asScala ++= extraProps.asScala @@ -1478,7 +1494,9 @@ class ReplicaManagerTest { topicPartition = new TopicPartition(topic, topicPartition), producerStateManager = new ProducerStateManager(new TopicPartition(topic, topicPartition), new File(new File(config.logDirs.head), s"$topic-$topicPartition"), 30000), - logDirFailureChannel = mockLogDirFailureChannel) { + logDirFailureChannel = mockLogDirFailureChannel, + topicId = topicId, + keepPartitionMetadataFile = true) { override def endOffsetForEpoch(leaderEpoch: Int): Option[OffsetAndEpoch] = { assertEquals(leaderEpoch, leaderEpochFromLeader) @@ -1500,7 +1518,7 @@ class ReplicaManagerTest { val mockLogMgr: LogManager = EasyMock.createMock(classOf[LogManager]) EasyMock.expect(mockLogMgr.liveLogDirs).andReturn(config.logDirs.map(new File(_).getAbsoluteFile)).anyTimes EasyMock.expect(mockLogMgr.getOrCreateLog(EasyMock.eq(topicPartitionObj), - isNew = EasyMock.eq(false), isFuture = EasyMock.eq(false))).andReturn(mockLog).anyTimes + isNew = EasyMock.eq(false), isFuture = EasyMock.eq(false), EasyMock.anyObject())).andReturn(mockLog).anyTimes if (expectTruncation) { EasyMock.expect(mockLogMgr.truncateTo(Map(topicPartitionObj -> offsetFromLeader), isFuture = false)).once @@ -1965,7 +1983,7 @@ class ReplicaManagerTest { val tp0 = new TopicPartition(topic, 0) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) - replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 10, brokerEpoch, Seq(leaderAndIsrPartitionState(tp0, 1, 0, Seq(0, 1), true)).asJava, @@ -1992,7 +2010,7 @@ class ReplicaManagerTest { val tp0 = new TopicPartition(topic, 0) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) - replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + replicaManager.createPartition(tp0).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, Seq(leaderAndIsrPartitionState(tp0, 1, 0, Seq(0, 1), true)).asJava, @@ -2034,7 +2052,7 @@ class ReplicaManagerTest { val replicaManager = setupReplicaManagerWithMockedPurgatories(mockTimer, aliveBrokerIds = Seq(0, 1)) val tp0 = new TopicPartition(topic, 0) - val log = replicaManager.logManager.getOrCreateLog(tp0, true) + val log = replicaManager.logManager.getOrCreateLog(tp0, true, topicId = None) if (throwIOException) { // Delete the underlying directory to trigger an KafkaStorageException @@ -2129,7 +2147,7 @@ class ReplicaManagerTest { val tp0 = new TopicPartition(topic, 0) val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints) val partition = replicaManager.createPartition(tp0) - partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints) + partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None) val logDirFailureChannel = new LogDirFailureChannel(replicaManager.config.logDirs.size) val logDir = partition.log.get.parentDirFile @@ -2215,14 +2233,57 @@ class ReplicaManagerTest { } @Test - def testPartitionMetadataFile() = { + def testPartitionMetadataFile(): Unit = { val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time)) try { val brokerList = Seq[Integer](0, 1).asJava val topicPartition = new TopicPartition(topic, 0) - replicaManager.createPartition(topicPartition) - .createLogIfNotExists(isNew = false, isFutureReplica = false, - new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)) + val topicIds = Collections.singletonMap(topic, Uuid.randomUuid()) + val topicNames = topicIds.asScala.map(_.swap).asJava + + def leaderAndIsrRequest(epoch: Int, topicIds: java.util.Map[String, Uuid]): LeaderAndIsrRequest = + new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, + Seq(new LeaderAndIsrPartitionState() + .setTopicName(topic) + .setPartitionIndex(0) + .setControllerEpoch(0) + .setLeader(0) + .setLeaderEpoch(epoch) + .setIsr(brokerList) + .setZkVersion(0) + .setReplicas(brokerList) + .setIsNew(true)).asJava, + topicIds, + Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build() + + val response = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, topicIds), (_, _) => ()) + assertEquals(Errors.NONE, response.partitionErrors(topicNames).get(topicPartition)) + assertFalse(replicaManager.localLog(topicPartition).isEmpty) + val id = topicIds.get(topicPartition.topic()) + val log = replicaManager.localLog(topicPartition).get + assertTrue(log.partitionMetadataFile.exists()) + val partitionMetadata = log.partitionMetadataFile.read() + + // Current version of PartitionMetadataFile is 0. + assertEquals(0, partitionMetadata.version) + assertEquals(id, partitionMetadata.topicId) + } finally replicaManager.shutdown(checkpointHW = false) + } + + @Test + def testPartitionMetadataFileCreatedWithExistingLog(): Unit = { + val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time)) + try { + val brokerList = Seq[Integer](0, 1).asJava + val topicPartition = new TopicPartition(topic, 0) + + replicaManager.logManager.getOrCreateLog(topicPartition, isNew = true, topicId = None) + + assertTrue(replicaManager.getLog(topicPartition).isDefined) + var log = replicaManager.getLog(topicPartition).get + assertEquals(None, log.topicId) + assertFalse(log.partitionMetadataFile.exists()) + val topicIds = Collections.singletonMap(topic, Uuid.randomUuid()) val topicNames = topicIds.asScala.map(_.swap).asJava @@ -2244,7 +2305,7 @@ class ReplicaManagerTest { assertEquals(Errors.NONE, response.partitionErrors(topicNames).get(topicPartition)) assertFalse(replicaManager.localLog(topicPartition).isEmpty) val id = topicIds.get(topicPartition.topic()) - val log = replicaManager.localLog(topicPartition).get + log = replicaManager.localLog(topicPartition).get assertTrue(log.partitionMetadataFile.exists()) val partitionMetadata = log.partitionMetadataFile.read() @@ -2255,14 +2316,11 @@ class ReplicaManagerTest { } @Test - def testInvalidIdReturnsError() = { + def testInconsistentIdReturnsError(): Unit = { val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time)) try { val brokerList = Seq[Integer](0, 1).asJava val topicPartition = new TopicPartition(topic, 0) - replicaManager.createPartition(topicPartition) - .createLogIfNotExists(isNew = false, isFutureReplica = false, - new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)) val topicIds = Collections.singletonMap(topic, Uuid.randomUuid()) val topicNames = topicIds.asScala.map(_.swap).asJava @@ -2290,7 +2348,7 @@ class ReplicaManagerTest { val response2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(1, topicIds), (_, _) => ()) assertEquals(Errors.NONE, response2.partitionErrors(topicNames).get(topicPartition)) - // Send request with invalid ID. + // Send request with inconsistent ID. val response3 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(1, invalidTopicIds), (_, _) => ()) assertEquals(Errors.INCONSISTENT_TOPIC_ID, response3.partitionErrors(invalidTopicNames).get(topicPartition)) @@ -2300,15 +2358,13 @@ class ReplicaManagerTest { } @Test - def testPartitionMetadataFileNotCreated() = { + def testPartitionMetadataFileNotCreated(): Unit = { val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time)) try { val brokerList = Seq[Integer](0, 1).asJava val topicPartition = new TopicPartition(topic, 0) val topicPartitionFoo = new TopicPartition("foo", 0) - replicaManager.createPartition(topicPartition) - .createLogIfNotExists(isNew = false, isFutureReplica = false, - new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)) + val topicPartitionFake = new TopicPartition("fakeTopic", 0) val topicIds = Map(topic -> Uuid.ZERO_UUID, "foo" -> Uuid.randomUuid()).asJava val topicNames = topicIds.asScala.map(_.swap).asJava @@ -2329,28 +2385,28 @@ class ReplicaManagerTest { // There is no file if the topic does not have an associated topic ID. val response = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "fakeTopic", ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ()) - assertFalse(replicaManager.localLog(topicPartition).isEmpty) - val log = replicaManager.localLog(topicPartition).get + assertTrue(replicaManager.localLog(topicPartitionFake).isDefined) + val log = replicaManager.localLog(topicPartitionFake).get assertFalse(log.partitionMetadataFile.exists()) assertEquals(Errors.NONE, response.partitionErrors(topicNames).get(topicPartition)) // There is no file if the topic has the default UUID. val response2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, topic, ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ()) - assertFalse(replicaManager.localLog(topicPartition).isEmpty) + assertTrue(replicaManager.localLog(topicPartition).isDefined) val log2 = replicaManager.localLog(topicPartition).get assertFalse(log2.partitionMetadataFile.exists()) assertEquals(Errors.NONE, response2.partitionErrors(topicNames).get(topicPartition)) // There is no file if the request an older version val response3 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "foo", 0), (_, _) => ()) - assertFalse(replicaManager.localLog(topicPartitionFoo).isEmpty) + assertTrue(replicaManager.localLog(topicPartitionFoo).isDefined) val log3 = replicaManager.localLog(topicPartitionFoo).get assertFalse(log3.partitionMetadataFile.exists()) assertEquals(Errors.NONE, response3.partitionErrors(topicNames).get(topicPartitionFoo)) // There is no file if the request is an older version val response4 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(1, "foo", 4), (_, _) => ()) - assertFalse(replicaManager.localLog(topicPartitionFoo).isEmpty) + assertTrue(replicaManager.localLog(topicPartitionFoo).isDefined) val log4 = replicaManager.localLog(topicPartitionFoo).get assertFalse(log4.partitionMetadataFile.exists()) assertEquals(Errors.NONE, response4.partitionErrors(topicNames).get(topicPartitionFoo)) diff --git a/core/src/test/scala/unit/kafka/tools/DumpLogSegmentsTest.scala b/core/src/test/scala/unit/kafka/tools/DumpLogSegmentsTest.scala index 1a5e51e5fe3da..256262a99b4e0 100644 --- a/core/src/test/scala/unit/kafka/tools/DumpLogSegmentsTest.scala +++ b/core/src/test/scala/unit/kafka/tools/DumpLogSegmentsTest.scala @@ -61,7 +61,7 @@ class DumpLogSegmentsTest { log = Log(logDir, LogConfig(props), logStartOffset = 0L, recoveryPoint = 0L, scheduler = time.scheduler, time = time, brokerTopicStats = new BrokerTopicStats, maxProducerIdExpirationMs = 60 * 60 * 1000, producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs, - logDirFailureChannel = new LogDirFailureChannel(10)) + logDirFailureChannel = new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true) } def addSimpleRecords(): Unit = { diff --git a/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala b/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala index eb7e81297c2ba..913d3e2449a4b 100644 --- a/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala +++ b/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala @@ -124,7 +124,7 @@ class SchedulerTest { val producerStateManager = new ProducerStateManager(topicPartition, logDir, maxProducerIdExpirationMs) val log = new Log(logDir, logConfig, logStartOffset = 0, recoveryPoint = recoveryPoint, scheduler, brokerTopicStats, mockTime, maxProducerIdExpirationMs, LogManager.ProducerIdExpirationCheckIntervalMs, - topicPartition, producerStateManager, new LogDirFailureChannel(10)) + topicPartition, producerStateManager, new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true) assertTrue(scheduler.taskRunning(log.producerExpireCheck)) log.close() assertFalse(scheduler.taskRunning(log.producerExpireCheck)) diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java index 453fa40096b2f..05e5f84649a74 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java @@ -44,6 +44,7 @@ import kafka.utils.KafkaScheduler; import kafka.utils.Pool; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; import org.apache.kafka.common.message.FetchResponseData; import org.apache.kafka.common.message.LeaderAndIsrRequestData; import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.OffsetForLeaderPartition; @@ -104,6 +105,7 @@ public class ReplicaFetcherThreadBenchmark { private File logDir = new File(System.getProperty("java.io.tmpdir"), UUID.randomUUID().toString()); private KafkaScheduler scheduler = new KafkaScheduler(1, "scheduler", true); private Pool pool = new Pool(Option.empty()); + private Option topicId = OptionConverters.toScala(Optional.of(Uuid.randomUuid())); @Setup(Level.Trial) public void setup() throws IOException { @@ -159,7 +161,7 @@ public void setup() throws IOException { 0, Time.SYSTEM, isrChangeListener, new DelayedOperationsMock(tp), Mockito.mock(MetadataCache.class), logManager, isrChannelManager); - partition.makeFollower(partitionState, offsetCheckpoints); + partition.makeFollower(partitionState, offsetCheckpoints, topicId); pool.put(tp, partition); initialFetchStates.put(tp, new InitialFetchState(new BrokerEndPoint(3, "host", 3000), 0, 0)); BaseRecords fetched = new BaseRecords() { diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/PartitionMakeFollowerBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/PartitionMakeFollowerBenchmark.java index ece6f86a0fed8..81cb1fa9af105 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/PartitionMakeFollowerBenchmark.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/PartitionMakeFollowerBenchmark.java @@ -33,6 +33,7 @@ import kafka.server.metadata.CachedConfigRepository; import kafka.utils.KafkaScheduler; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; import org.apache.kafka.common.message.LeaderAndIsrRequestData; import org.apache.kafka.common.record.CompressionType; import org.apache.kafka.common.record.MemoryRecords; @@ -54,6 +55,7 @@ import org.openjdk.jmh.annotations.Warmup; import scala.Option; import scala.collection.JavaConverters; +import scala.compat.java8.OptionConverters; import java.io.File; import java.io.IOException; @@ -62,6 +64,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Optional; import java.util.Properties; import java.util.UUID; import java.util.concurrent.ExecutorService; @@ -84,6 +87,7 @@ public class PartitionMakeFollowerBenchmark { private OffsetCheckpoints offsetCheckpoints = Mockito.mock(OffsetCheckpoints.class); private DelayedOperations delayedOperations = Mockito.mock(DelayedOperations.class); private ExecutorService executorService = Executors.newSingleThreadExecutor(); + private Option topicId; @Setup(Level.Trial) public void setup() throws IOException { @@ -114,6 +118,7 @@ public void setup() throws IOException { true); TopicPartition tp = new TopicPartition("topic", 0); + topicId = OptionConverters.toScala(Optional.of(Uuid.randomUuid())); Mockito.when(offsetCheckpoints.fetch(logDir.getAbsolutePath(), tp)).thenReturn(Option.apply(0L)); IsrChangeListener isrChangeListener = Mockito.mock(IsrChangeListener.class); @@ -122,7 +127,7 @@ public void setup() throws IOException { ApiVersion$.MODULE$.latestVersion(), 0, Time.SYSTEM, isrChangeListener, delayedOperations, Mockito.mock(MetadataCache.class), logManager, alterIsrManager); - partition.createLogIfNotExists(true, false, offsetCheckpoints); + partition.createLogIfNotExists(true, false, offsetCheckpoints, topicId); executorService.submit((Runnable) () -> { SimpleRecord[] simpleRecords = new SimpleRecord[] { new SimpleRecord(1L, "foo".getBytes(StandardCharsets.UTF_8), "1".getBytes(StandardCharsets.UTF_8)), @@ -155,7 +160,7 @@ public boolean testMakeFollower() { .setZkVersion(1) .setReplicas(replicas) .setIsNew(true); - return partition.makeFollower(partitionState, offsetCheckpoints); + return partition.makeFollower(partitionState, offsetCheckpoints, topicId); } private static LogConfig createLogConfig() { diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/UpdateFollowerFetchStateBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/UpdateFollowerFetchStateBenchmark.java index a82c6a0e7744f..e62191b45f409 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/UpdateFollowerFetchStateBenchmark.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/UpdateFollowerFetchStateBenchmark.java @@ -34,6 +34,7 @@ import kafka.server.metadata.CachedConfigRepository; import kafka.utils.KafkaScheduler; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState; import org.apache.kafka.common.utils.Time; import org.mockito.Mockito; @@ -51,11 +52,13 @@ import org.openjdk.jmh.annotations.Warmup; import scala.Option; import scala.collection.JavaConverters; +import scala.compat.java8.OptionConverters; import java.io.File; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Optional; import java.util.Properties; import java.util.UUID; import java.util.concurrent.TimeUnit; @@ -68,6 +71,7 @@ @OutputTimeUnit(TimeUnit.NANOSECONDS) public class UpdateFollowerFetchStateBenchmark { private TopicPartition topicPartition = new TopicPartition(UUID.randomUUID().toString(), 0); + private Option topicId = OptionConverters.toScala(Optional.of(Uuid.randomUuid())); private File logDir = new File(System.getProperty("java.io.tmpdir"), topicPartition.toString()); private KafkaScheduler scheduler = new KafkaScheduler(1, "scheduler", true); private BrokerTopicStats brokerTopicStats = new BrokerTopicStats(); @@ -120,7 +124,7 @@ public void setUp() { ApiVersion$.MODULE$.latestVersion(), 0, Time.SYSTEM, isrChangeListener, delayedOperations, Mockito.mock(MetadataCache.class), logManager, alterIsrManager); - partition.makeLeader(partitionState, offsetCheckpoints); + partition.makeLeader(partitionState, offsetCheckpoints, topicId); } // avoid mocked DelayedOperations to avoid mocked class affecting benchmark results diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java index 45d2cb1d3406f..6c16efb816d5f 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java @@ -33,6 +33,7 @@ import kafka.utils.MockTime; import kafka.utils.Scheduler; import kafka.utils.TestUtils; +import org.apache.kafka.common.Uuid; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.utils.Utils; @@ -145,7 +146,7 @@ public void setup() { OffsetCheckpoints checkpoints = (logDir, topicPartition) -> Option.apply(0L); for (TopicPartition topicPartition : topicPartitions) { final Partition partition = this.replicaManager.createPartition(topicPartition); - partition.createLogIfNotExists(true, false, checkpoints); + partition.createLogIfNotExists(true, false, checkpoints, Option.apply(Uuid.randomUuid())); } replicaManager.checkpointHighWatermarks();