Skip to content

KAFKA-12426: Missing logic to create partition.metadata files in RaftReplicaManager #10282

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 12 commits into from
Apr 1, 2021
Merged

KAFKA-12426: Missing logic to create partition.metadata files in RaftReplicaManager #10282

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
9effa4f
Added partitionMetadata file code to RaftReplicaManager
jolshan Mar 6, 2021
63439be
Check for inconsistent topic ID
jolshan Mar 8, 2021
de493a9
Merge branch 'trunk' of github.com:apache/kafka into KAFKA-12426
jolshan Mar 8, 2021
be36e6c
clean up merge change
jolshan Mar 8, 2021
5e3372a
Cleaned style as recommended by review
jolshan Mar 9, 2021
51b8f7e
Fixed bug that prevented writing to partition.metadata file on first …
jolshan Mar 15, 2021
4e1a7c7
Merge branch 'trunk' of github.com:apache/kafka into KAFKA-12426
jolshan Mar 17, 2021
2893d15
Removed optional argument and placement for partition.metadata file w…
jolshan Mar 18, 2021
ad65a25
Removed optional arguments, minor style changes
jolshan Mar 19, 2021
2e5c8f6
Merge branch 'trunk' of github.com:apache/kafka into KAFKA-12426
jolshan Mar 19, 2021
fac1f7a
Small tweaks to style, made optional keepPartitionMetadata file non-o…
jolshan Mar 31, 2021
c221911
Address comments. Changed error message and cleaned up some code.
jolshan Apr 1, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 21 additions & 44 deletions core/src/main/scala/kafka/cluster/Partition.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand All @@ -341,9 +341,14 @@ 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)
// When running a ZK controller, we may get a log that does not have a topic ID. Assign it here.
if (log.topicId == None && topicId.isDefined) {
Comment thread
jolshan marked this conversation as resolved.
Outdated
log.partitionMetadataFile.write(topicId.get)
log.topicId = Some(topicId.get)
Comment thread
jolshan marked this conversation as resolved.
Outdated
}
log
} finally {
logManager.finishedInitializingLog(topicPartition, maybeLog)
Expand Down Expand Up @@ -425,41 +430,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
Expand Down Expand Up @@ -540,7 +515,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
Expand All @@ -557,7 +533,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 " +
Expand Down Expand Up @@ -624,7 +600,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
Expand All @@ -639,7 +616,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 " +
Expand Down
33 changes: 23 additions & 10 deletions core/src/main/scala/kafka/log/Log.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -262,6 +266,7 @@ class Log(@volatile private var _dir: File,
val producerStateManager: ProducerStateManager,
logDirFailureChannel: LogDirFailureChannel,
private val hadCleanShutdown: Boolean = true,
@volatile var topicId : Option[Uuid] = None,
Comment thread
jolshan marked this conversation as resolved.
Outdated
val keepPartitionMetadataFile: Boolean = true) extends Logging with KafkaMetricsGroup {
Comment thread
jolshan marked this conversation as resolved.
Outdated

import kafka.log.Log._
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -349,11 +352,20 @@ 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 && fileTopicId != topicId.get)

@hachikuji hachikuji Mar 19, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You should be able to do something like !topicId.contains(fileTopicId)

@jolshan jolshan Mar 19, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

seems like if the option is None, this will return true, so I'll do topicId.isDefined && !topicId.contains(fileTopicId)

throw new IllegalStateException(s"Tried to assign topic ID $topicId to log, but log already contained topicId $fileTopicId")
Comment thread
jolshan marked this conversation as resolved.
Outdated
topicId = Some(fileTopicId)
}
} else if (topicId.isDefined && keepPartitionMetadataFile) {

@hachikuji hachikuji Mar 19, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My feeling is that we should be able to get rid of keepPartitionMetadataFile. We should just create the file whenever a topicId is provided. Not sure if there is anything I'm missing. I don't think the file would cause any harm if we downgraded, would it?

@jolshan jolshan Mar 19, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah. We could run into issues with downgrade and reupgrade. If we downgrade below 2.8 on ZK brokers, we may overwrite the topic ID on reupgrade. But the file will have the old version.

@hachikuji hachikuji Mar 20, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure I follow the scenario. Why would we overwrite the file on re-upgrade and what would be the problem with that?

@jolshan jolshan Mar 20, 2021 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We would rewrite in ZK, not the file. So we would always get an inconsistent topic ID.

We can overwrite in ZK when we add partitions/change ISR since this information is stored in the ZNode along with topic ID. When we write to it using an older version, we can't keep the topic ID field and it is lost.

@hachikuji hachikuji Mar 31, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, I think I get it.

partitionMetadataFile.write(topicId.get)
Comment thread
jolshan marked this conversation as resolved.
Outdated
}
}

Expand Down Expand Up @@ -2600,11 +2612,12 @@ object Log {
producerIdExpirationCheckIntervalMs: Int,
logDirFailureChannel: LogDirFailureChannel,
lastShutdownClean: Boolean = true,
topicId: Option[Uuid] = None,
keepPartitionMetadataFile: Boolean = true): 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)
}

/**
Expand Down
7 changes: 4 additions & 3 deletions core/src/main/scala/kafka/log/LogManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ 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}

Expand Down Expand Up @@ -773,12 +773,12 @@ class LogManager(logDirs: Seq[File],
* Otherwise throw KafkaStorageException
*
* @param topicPartition The partition whose log needs to be returned or created
* @param loadConfig A function to retrieve the log config, this is only called if the log is 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 topic used in the case of log creation.
* @throws KafkaStorageException if isNew=false, log is not found in the cache and there is offline log directory on the broker
*/
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 = {
Comment thread
jolshan marked this conversation as resolved.
logCreationOrDeletionLock synchronized {
getLog(topicPartition, isFuture).getOrElse {
// create the log if it has not already been created in another thread
Expand Down Expand Up @@ -827,6 +827,7 @@ class LogManager(logDirs: Seq[File],
time = time,
brokerTopicStats = brokerTopicStats,
logDirFailureChannel = logDirFailureChannel,
topicId = topicId,
keepPartitionMetadataFile = keepPartitionMetadataFile)

if (isFuture)
Expand Down
3 changes: 2 additions & 1 deletion core/src/main/scala/kafka/server/KafkaConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =
Comment thread
jolshan marked this conversation as resolved.
interBrokerProtocolVersion >= KAFKA_2_8_IV0
usesSelfManagedQuorum || interBrokerProtocolVersion >= KAFKA_2_8_IV0

validateValues()

Expand Down
14 changes: 8 additions & 6 deletions core/src/main/scala/kafka/server/RaftReplicaChangeDelegate.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down Expand Up @@ -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] = {

@hachikuji hachikuji Mar 31, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess we could use a strong type here since topicIds are required for KIP-500, but maybe not worth it since we are delegating to Partition in the end.

val partitionsMadeLeaders = mutable.Set[Partition]()
val traceLoggingEnabled = helper.stateChangeLogger.isTraceEnabled
val deferredBatches = metadataOffset.isEmpty
Expand All @@ -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.")
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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 " +
Expand Down
Loading