You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2021/07/29 22:07:24 UTC
[kafka] branch 3.0 updated: KAFKA-13132;
Ensure topicId is updated on replicas even when the leader epoch is
unchanged (#11126)
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch 3.0
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.0 by this push:
new 168d5c3 KAFKA-13132; Ensure topicId is updated on replicas even when the leader epoch is unchanged (#11126)
168d5c3 is described below
commit 168d5c3a7cf23dcda0aff99e82bfc07a36494d56
Author: Justine Olshan <jo...@confluent.io>
AuthorDate: Thu Jul 29 14:36:44 2021 -0700
KAFKA-13132; Ensure topicId is updated on replicas even when the leader epoch is unchanged (#11126)
In 3.0, there was a change that resulted in no longer assigning topic IDs to the log and the partition.metadata file in certain upgrade scenarios, specifically when upgrading from IBP 2.7 or below to 3.0. In this case, there may not be a bump to the leader epoch when the topicId is assigned by the controller, so the LeaderAndIsr request from the controller would be ignored by the replica. This PR fixes the problem by adding a check for whether we need to handle the LeaderAndIsr reques [...]
Reviewers: Jason Gustafson <ja...@confluent.io>
---
.../main/scala/kafka/server/ReplicaManager.scala | 28 +++++++---
.../controller/ControllerIntegrationTest.scala | 65 +++++++++++++++++++++-
.../unit/kafka/server/ReplicaManagerTest.scala | 57 ++++++++++++++++---
.../test/scala/unit/kafka/utils/TestUtils.scala | 5 +-
4 files changed, 136 insertions(+), 19 deletions(-)
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
index f03571c..84ccd8e 100644
--- a/core/src/main/scala/kafka/server/ReplicaManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -1349,9 +1349,10 @@ class ReplicaManager(val config: KafkaConfig,
val currentLeaderEpoch = partition.getLeaderEpoch
val requestLeaderEpoch = partitionState.leaderEpoch
val requestTopicId = topicIdFromRequest(topicPartition.topic)
+ val logTopicId = partition.topicId
- if (!hasConsistentTopicId(requestTopicId, partition.topicId)) {
- stateChangeLogger.error(s"Topic ID in memory: ${partition.topicId.get} does not" +
+ if (!hasConsistentTopicId(requestTopicId, logTopicId)) {
+ stateChangeLogger.error(s"Topic ID in memory: ${logTopicId.get} does not" +
s" match the topic ID for partition $topicPartition received: " +
s"${requestTopicId.get}.")
responseMap.put(topicPartition, Errors.INCONSISTENT_TOPIC_ID)
@@ -1374,11 +1375,24 @@ class ReplicaManager(val config: KafkaConfig,
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)
+ val error = requestTopicId match {
+ case Some(topicId) if logTopicId.isEmpty =>
+ // The controller may send LeaderAndIsr to upgrade to using topic IDs without bumping the epoch.
+ // If we have a matching epoch, we expect the log to be defined.
+ val log = localLogOrException(partition.topicPartition)
+ log.assignTopicId(topicId)
+ stateChangeLogger.info(s"Updating log for $topicPartition to assign topic ID " +
+ s"$topicId from LeaderAndIsr request from controller $controllerId with correlation " +
+ s"id $correlationId epoch $controllerEpoch")
+ Errors.NONE
+ case _ =>
+ 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")
+ Errors.STALE_CONTROLLER_EPOCH
+ }
+ responseMap.put(topicPartition, error)
}
}
}
diff --git a/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala b/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
index c0e8611..987ec2d 100644
--- a/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
+++ b/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
@@ -1115,6 +1115,66 @@ class ControllerIntegrationTest extends ZooKeeperTestHarness {
assertEquals(topicIdAfterUpgrade.get, topicId)
assertEquals("t", controller2.controllerContext.topicNames(topicId))
+ TestUtils.waitUntilTrue(() => servers(0).logManager.getLog(tp).isDefined, "log was not created")
+
+ val topicIdInLog = servers(0).logManager.getLog(tp).get.topicId
+ assertEquals(Some(topicId), topicIdInLog)
+
+ adminZkClient.deleteTopic(tp.topic)
+ TestUtils.waitUntilTrue(() => !servers.head.kafkaController.controllerContext.allTopics.contains(tp.topic),
+ "topic should have been removed from controller context after deletion")
+ }
+
+ @Test
+ def testTopicIdCreatedOnUpgradeMultiBrokerScenario(): Unit = {
+ // Simulate an upgrade scenario where the controller is still on a pre-topic ID IBP, but the other two brokers are upgraded.
+ servers = makeServers(1, interBrokerProtocolVersion = Some(KAFKA_2_7_IV0))
+ servers = servers ++ makeServers(3, startingIdNumber = 1)
+ val originalControllerId = TestUtils.waitUntilControllerElected(zkClient)
+ assertEquals(0, originalControllerId)
+ val controller = getController().kafkaController
+ assertEquals(KAFKA_2_7_IV0, servers(originalControllerId).config.interBrokerProtocolVersion)
+ val remainingBrokers = servers.filter(_.config.brokerId != originalControllerId)
+ val tp = new TopicPartition("t", 0)
+ // Only the remaining brokers will have the replicas for the partition
+ val assignment = Map(tp.partition -> remainingBrokers.map(_.config.brokerId))
+ TestUtils.createTopic(zkClient, tp.topic, partitionReplicaAssignment = assignment, servers = servers)
+ waitForPartitionState(tp, firstControllerEpoch, remainingBrokers(0).config.brokerId, LeaderAndIsr.initialLeaderEpoch,
+ "failed to get expected partition state upon topic creation")
+ val topicIdAfterCreate = zkClient.getTopicIdsForTopics(Set(tp.topic())).get(tp.topic())
+ assertEquals(None, topicIdAfterCreate)
+ val emptyTopicId = controller.controllerContext.topicIds.get("t")
+ assertEquals(None, emptyTopicId)
+
+ // All partition logs should not have topic IDs
+ remainingBrokers.foreach { server =>
+ TestUtils.waitUntilTrue(() => server.logManager.getLog(tp).isDefined, "log was not created for server" + server.config.brokerId)
+ val topicIdInLog = server.logManager.getLog(tp).get.topicId
+ assertEquals(None, topicIdInLog)
+ }
+
+ // Shut down the controller to transfer the controller to a new IBP broker.
+ servers(originalControllerId).shutdown()
+ servers(originalControllerId).awaitShutdown()
+ // If we were upgrading, this server would be the latest IBP, but it doesn't matter in this test scenario
+ servers(originalControllerId).startup()
+ TestUtils.waitUntilTrue(() => zkClient.getControllerId.isDefined, "failed to elect a controller")
+ val topicIdAfterUpgrade = zkClient.getTopicIdsForTopics(Set(tp.topic())).get(tp.topic())
+ assertNotEquals(emptyTopicId, topicIdAfterUpgrade)
+ val controller2 = getController().kafkaController
+ assertNotEquals(emptyTopicId, controller2.controllerContext.topicIds.get("t"))
+ val topicId = controller2.controllerContext.topicIds.get("t").get
+ assertEquals(topicIdAfterUpgrade.get, topicId)
+ assertEquals("t", controller2.controllerContext.topicNames(topicId))
+
+ // All partition logs should have topic IDs
+ remainingBrokers.foreach { server =>
+ TestUtils.waitUntilTrue(() => server.logManager.getLog(tp).isDefined, "log was not created for server" + server.config.brokerId)
+ val topicIdInLog = server.logManager.getLog(tp).get.topicId
+ assertEquals(Some(topicId), topicIdInLog,
+ s"Server ${server.config.brokerId} had topic ID $topicIdInLog instead of ${Some(topicId)} as expected.")
+ }
+
adminZkClient.deleteTopic(tp.topic)
TestUtils.waitUntilTrue(() => !servers.head.kafkaController.controllerContext.allTopics.contains(tp.topic),
"topic should have been removed from controller context after deletion")
@@ -1287,8 +1347,9 @@ class ControllerIntegrationTest extends ZooKeeperTestHarness {
listenerSecurityProtocolMap : Option[String] = None,
controlPlaneListenerName : Option[String] = None,
interBrokerProtocolVersion: Option[ApiVersion] = None,
- logDirCount: Int = 1) = {
- val configs = TestUtils.createBrokerConfigs(numConfigs, zkConnect, enableControlledShutdown = enableControlledShutdown, logDirCount = logDirCount)
+ logDirCount: Int = 1,
+ startingIdNumber: Int = 0) = {
+ val configs = TestUtils.createBrokerConfigs(numConfigs, zkConnect, enableControlledShutdown = enableControlledShutdown, logDirCount = logDirCount, startingIdNumber = startingIdNumber )
configs.foreach { config =>
config.setProperty(KafkaConfig.AutoLeaderRebalanceEnableProp, autoLeaderRebalanceEnable.toString)
config.setProperty(KafkaConfig.UncleanLeaderElectionEnableProp, uncleanLeaderElectionEnable.toString)
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index fd6ba75..f92b2d2 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -252,7 +252,6 @@ class ReplicaManagerTest {
replicaManager.createPartition(topicPartition)
.createLogIfNotExists(isNew = false, isFutureReplica = false,
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,
Seq(new LeaderAndIsrPartitionState()
@@ -265,7 +264,7 @@ class ReplicaManagerTest {
.setZkVersion(0)
.setReplicas(brokerList)
.setIsNew(true)).asJava,
- topicIds,
+ Collections.singletonMap(topic, Uuid.randomUuid()),
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0), (_, _) => ())
@@ -1228,7 +1227,6 @@ class ReplicaManagerTest {
val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)
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())
val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
Seq(new LeaderAndIsrPartitionState()
@@ -1241,7 +1239,7 @@ class ReplicaManagerTest {
.setZkVersion(0)
.setReplicas(partition0Replicas)
.setIsNew(true)).asJava,
- topicIds,
+ Collections.singletonMap(tp0.topic, Uuid.randomUuid()),
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
replicaManager.becomeLeaderOrFollower(1, becomeLeaderRequest, (_, _) => ())
@@ -1262,7 +1260,7 @@ class ReplicaManagerTest {
.setZkVersion(0)
.setReplicas(partition0Replicas)
.setIsNew(true)).asJava,
- topicIds,
+ Collections.singletonMap(tp0.topic, Uuid.randomUuid()),
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
replicaManager.becomeLeaderOrFollower(0, becomeFollowerRequest, (_, _) => ())
@@ -1279,7 +1277,6 @@ class ReplicaManagerTest {
val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)
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())
val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
Seq(new LeaderAndIsrPartitionState()
@@ -1292,7 +1289,7 @@ class ReplicaManagerTest {
.setZkVersion(0)
.setReplicas(partition0Replicas)
.setIsNew(true)).asJava,
- topicIds,
+ Collections.singletonMap(tp0.topic, Uuid.randomUuid()),
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
replicaManager.becomeLeaderOrFollower(1, becomeLeaderRequest, (_, _) => ())
@@ -1314,7 +1311,7 @@ class ReplicaManagerTest {
.setZkVersion(0)
.setReplicas(partition0Replicas)
.setIsNew(true)).asJava,
- topicIds,
+ Collections.singletonMap(tp0.topic, Uuid.randomUuid()),
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
replicaManager.becomeLeaderOrFollower(0, becomeFollowerRequest, (_, _) => ())
@@ -2357,6 +2354,50 @@ class ReplicaManagerTest {
}
@Test
+ def testPartitionMetadataFileCreatedAfterPreviousRequestWithoutIds(): Unit = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time))
+ try {
+ val brokerList = Seq[Integer](0, 1).asJava
+ val topicPartition = new TopicPartition(topic, 0)
+ val topicId = Uuid.randomUuid()
+ val topicIds = Collections.singletonMap(topic, topicId)
+ val topicNames = Collections.singletonMap(topicId, topic)
+
+ def leaderAndIsrRequest(topicIds: util.Map[String, Uuid], version: Short): LeaderAndIsrRequest =
+ new LeaderAndIsrRequest.Builder(version, 0, 0, brokerEpoch,
+ Seq(new LeaderAndIsrPartitionState()
+ .setTopicName(topic)
+ .setPartitionIndex(0)
+ .setControllerEpoch(0)
+ .setLeader(0)
+ .setLeaderEpoch(0)
+ .setIsr(brokerList)
+ .setZkVersion(0)
+ .setReplicas(brokerList)
+ .setIsNew(true)).asJava,
+ topicIds,
+ Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
+
+ // Send a request without a topic ID so that we have a log without a topic ID associated to the partition.
+ val response = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(Collections.emptyMap(), 4), (_, _) => ())
+ assertEquals(Errors.NONE, response.partitionErrors(Collections.emptyMap()).get(topicPartition))
+ assertTrue(replicaManager.localLog(topicPartition).isDefined)
+ val log = replicaManager.localLog(topicPartition).get
+ assertFalse(log.partitionMetadataFile.exists())
+ assertTrue(log.topicId.isEmpty)
+
+ val response2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(topicIds, ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ())
+ assertEquals(Errors.NONE, response2.partitionErrors(topicNames).get(topicPartition))
+ assertTrue(replicaManager.localLog(topicPartition).isDefined)
+ assertTrue(log.partitionMetadataFile.exists())
+ assertTrue(log.topicId.isDefined)
+ assertEquals(topicId, log.topicId.get)
+
+ assertEquals(topicId, log.partitionMetadataFile.read().topicId)
+ } finally replicaManager.shutdown(checkpointHW = false)
+ }
+
+ @Test
def testInconsistentIdReturnsError(): Unit = {
val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time))
try {
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index 9ff2f3a..bd9c02c 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -204,8 +204,9 @@ object TestUtils extends Logging {
logDirCount: Int = 1,
enableToken: Boolean = false,
numPartitions: Int = 1,
- defaultReplicationFactor: Short = 1): Seq[Properties] = {
- (0 until numConfigs).map { node =>
+ defaultReplicationFactor: Short = 1,
+ startingIdNumber: Int = 0): Seq[Properties] = {
+ (startingIdNumber until numConfigs).map { node =>
createBrokerConfig(node, zkConnect, enableControlledShutdown, enableDeleteTopic, RandomPort,
interBrokerSecurityProtocol, trustStoreFile, saslProperties, enablePlaintext = enablePlaintext, enableSsl = enableSsl,
enableSaslPlaintext = enableSaslPlaintext, enableSaslSsl = enableSaslSsl, rack = rackInfo.get(node), logDirCount = logDirCount, enableToken = enableToken,