Skip to content

KAFKA-12332: Error partitions from topics with invalid IDs in LISR requests #10143

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 4 commits into from
Feb 19, 2021
Merged

KAFKA-12332: Error partitions from topics with invalid IDs in LISR requests #10143

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
dc72786
Changed becomeLeaderOrFollower to ignore partitions with invalid IDs …
jolshan Feb 17, 2021
89968d9
Addressed comments. Moved topic ID checks to Partition.
jolshan Feb 18, 2021
06944ab
Cleanups
jolshan Feb 18, 2021
258d721
Updated comment
jolshan Feb 18, 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
94 changes: 50 additions & 44 deletions core/src/main/scala/kafka/server/ReplicaManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1364,34 +1364,59 @@ class ReplicaManager(val config: KafkaConfig,
Some(partition)
}

// Next check partition's leader epoch
// Next check the topic ID and the partition's leader epoch
partitionOpt.foreach { partition =>
val currentLeaderEpoch = partition.getLeaderEpoch
val requestLeaderEpoch = partitionState.leaderEpoch
if (requestLeaderEpoch > currentLeaderEpoch) {
// If the leader epoch is valid 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
if (partitionState.replicas.contains(localBrokerId))
partitionStates.put(partition, partitionState)
else {
stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from controller $controllerId with " +
s"correlation id $correlationId epoch $controllerEpoch for partition $topicPartition as itself is not " +
s"in assigned replica list ${partitionState.replicas.asScala.mkString(",")}")
responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_OR_PARTITION)
val id = topicIds.get(topicPartition.topic())
Comment thread
jolshan marked this conversation as resolved.
Outdated
var invalidId = false

// Ensure we have not received a request from an older protocol
if (id != null && id != Uuid.ZERO_UUID) {
partition.log.foreach { 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.
if (log.topicId == Uuid.ZERO_UUID) {
log.partitionMetadataFile.write(id)
log.topicId = id
// Warn if the topic ID in the request does not match the log.
} else if (log.topicId != id) {
stateChangeLogger.warn(s"Topic Id in memory: ${log.topicId.toString} does not" +
Comment thread
jolshan marked this conversation as resolved.
Outdated
s" match the topic Id provided in the request: " +
s"${id.toString}.")
responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_ID)
Comment thread
jolshan marked this conversation as resolved.
Outdated
invalidId = true
}
}
}

// If we found an invalid ID, we don't need to check the leader epoch
if (!invalidId) {
if (requestLeaderEpoch > currentLeaderEpoch) {

@hachikuji hachikuji Feb 17, 2021 •
edited
Loading

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.

It's a minor thing, but we can avoid this nesting by restructuring the checks a little bit. For example, it would be a good idea to have a helper in Partition which encapsulates the update of the topicId state in Log. Maybe something like this:

class Partition {
  // Update topicid if necessary. 
  // Return false if the update failed because the topicId is inconsistent
  def maybeUpdateTopicId(topicId: Uuid): Boolean
}

// in ReplicaManager
if (!partition.maybeUpdateTopicId(requestTopicId)) {
  error(...)
  responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_ID)
} else if (requestLeaderEpoch > currentLeaderEpoch) {
...

@jolshan jolshan Feb 17, 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.

I was struggling with this since I didn't think it was the best solution. I'll take a look at the version you have here. Would we want to ever update the topic ID though? Maybe just a method to check if they are equal.

@jolshan jolshan Feb 17, 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.

Oh hmmm I see how this could be used on the path for actually setting it. I'll think of a good name

@jolshan jolshan Feb 17, 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.

In this case, if log is None, does it make sense to error here? If it is None, we are unable to check the topic ID

// If the leader epoch is valid 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
if (partitionState.replicas.contains(localBrokerId))
partitionStates.put(partition, partitionState)
else {
stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from controller $controllerId with " +
s"correlation id $correlationId epoch $controllerEpoch for partition $topicPartition as itself is not " +
s"in assigned replica list ${partitionState.replicas.asScala.mkString(",")}")
responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_OR_PARTITION)
}
} else if (requestLeaderEpoch < currentLeaderEpoch) {
stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from " +
s"controller $controllerId with correlation id $correlationId " +
s"epoch $controllerEpoch for partition $topicPartition since its associated " +
s"leader epoch $requestLeaderEpoch is smaller than the current " +
s"leader epoch $currentLeaderEpoch")
responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH)
} else {
stateChangeLogger.info(s"Ignoring LeaderAndIsr request from " +
s"controller $controllerId with correlation id $correlationId " +
s"epoch $controllerEpoch for partition $topicPartition since its associated " +
s"leader epoch $requestLeaderEpoch matches the current leader epoch")
responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH)
}
} else if (requestLeaderEpoch < currentLeaderEpoch) {
stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from " +
s"controller $controllerId with correlation id $correlationId " +
s"epoch $controllerEpoch for partition $topicPartition since its associated " +
s"leader epoch $requestLeaderEpoch is smaller than the current " +
s"leader epoch $currentLeaderEpoch")
responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH)
} else {
stateChangeLogger.info(s"Ignoring LeaderAndIsr request from " +
s"controller $controllerId with correlation id $correlationId " +
s"epoch $controllerEpoch for partition $topicPartition since its associated " +
s"leader epoch $requestLeaderEpoch matches the current leader epoch")
responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH)
}
}
}
Expand Down Expand Up @@ -1424,27 +1449,8 @@ class ReplicaManager(val config: KafkaConfig,
* In this case ReplicaManager.allPartitions will map this topic-partition to an empty Partition object.
* we need to map this topic-partition to OfflinePartition instead.
*/
val local = localLog(topicPartition)
if (local.isEmpty)
if (localLog(topicPartition).isEmpty)
markPartitionOffline(topicPartition)
else {
val id = topicIds.get(topicPartition.topic())
// Ensure we have not received a request from an older protocol
if (id != null && !id.equals(Uuid.ZERO_UUID)) {
val log = local.get
// 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.
if (log.topicId.equals(Uuid.ZERO_UUID)) {
log.partitionMetadataFile.write(id)
log.topicId = id
// Warn if the topic ID in the request does not match the log.
} else if (!log.topicId.equals(id)) {
stateChangeLogger.warn(s"Topic Id in memory: ${log.topicId.toString} does not" +
s" match the topic Id provided in the request: " +
s"${id.toString}.")
}
}
}
}

// we initialize highwatermark thread after the first leaderisrrequest. This ensures that all the partitions
Expand Down
59 changes: 54 additions & 5 deletions core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2229,6 +2229,7 @@ class ReplicaManagerTest {
.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): LeaderAndIsrRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
Seq(new LeaderAndIsrPartitionState()
Expand All @@ -2244,7 +2245,8 @@ class ReplicaManagerTest {
topicIds,
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()

replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0), (_, _) => ())
val response = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0), (_, _) => ())
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
Expand All @@ -2257,6 +2259,48 @@ class ReplicaManagerTest {
} finally replicaManager.shutdown(checkpointHW = false)
}

@Test
def testInvalidIdReturnsError() = {
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

val invalidTopicIds = Collections.singletonMap(topic, Uuid.randomUuid())
val invalidTopicNames = invalidTopicIds.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))

// Send request with invalid ID.
val response2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, invalidTopicIds), (_, _) => ())
assertEquals(Errors.UNKNOWN_TOPIC_ID, response2.partitionErrors(invalidTopicNames).get(topicPartition))

val response3 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(1, invalidTopicIds), (_, _) => ())
assertEquals(Errors.UNKNOWN_TOPIC_ID, response3.partitionErrors(invalidTopicNames).get(topicPartition))
} finally replicaManager.shutdown(checkpointHW = false)
}

@Test
def testPartitionMetadataFileNotCreated() = {
val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time))
Expand All @@ -2268,6 +2312,7 @@ class ReplicaManagerTest {
.createLogIfNotExists(isNew = false, isFutureReplica = false,
new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints))
val topicIds = Map(topic -> Uuid.ZERO_UUID, "foo" -> Uuid.randomUuid()).asJava
val topicNames = topicIds.asScala.map(_.swap).asJava

def leaderAndIsrRequest(epoch: Int, name: String, version: Short): LeaderAndIsrRequest = LeaderAndIsrRequest.parse(
new LeaderAndIsrRequest.Builder(version, 0, 0, brokerEpoch,
Expand All @@ -2285,28 +2330,32 @@ class ReplicaManagerTest {
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build().serialize(), version)

// There is no file if the topic does not have an associated topic ID.
replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "fakeTopic", ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ())
val response = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "fakeTopic", ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ())
assertFalse(replicaManager.localLog(topicPartition).isEmpty)
val log = replicaManager.localLog(topicPartition).get
assertFalse(log.partitionMetadataFile.exists())
assertEquals(Errors.NONE, response.partitionErrors(topicNames).get(topicPartition))

// There is no file if the topic has the default UUID.
replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, topic, ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ())
val response2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, topic, ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ())
assertFalse(replicaManager.localLog(topicPartition).isEmpty)
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
replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "foo", 0), (_, _) => ())
val response3 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "foo", 0), (_, _) => ())
assertFalse(replicaManager.localLog(topicPartitionFoo).isEmpty)
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
replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "foo", 4), (_, _) => ())
val response4 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(1, "foo", 4), (_, _) => ())
assertFalse(replicaManager.localLog(topicPartitionFoo).isEmpty)
val log4 = replicaManager.localLog(topicPartitionFoo).get
assertFalse(log4.partitionMetadataFile.exists())
assertEquals(Errors.NONE, response4.partitionErrors(topicNames).get(topicPartitionFoo))
} finally replicaManager.shutdown(checkpointHW = false)
}

Expand Down