Skip to content

KAFKA-13592:Fix flaky test ControllerIntegrationTest.testTopicIdUpgradeAfterReassigningPartitions #11687

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 10 commits into from
Jun 7, 2022
Merged

KAFKA-13592:Fix flaky test ControllerIntegrationTest.testTopicIdUpgradeAfterReassigningPartitions #11687

Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
4824255
Fix flaky test ControllerIntegrationTest.testTopicIdUpgradeAfterReass…
Karonazaba Jan 18, 2022
2822ef0
Fix flaky test ControllerIntegrationTest.testTopicIdUpgradeAfterReass…
Karonazaba Jan 18, 2022
47bef85
Merge branch 'trunk' of github.com:apache/kafka into KAFKA-13592
Karonazaba Jan 18, 2022
a427cf6
guarantee that the controller has already written the topicId to the …
Karonazaba Jan 20, 2022
dfd1d4b
resolve conflicts
Kvicii May 27, 2022
00da0ea
delete test sh
Kvicii May 27, 2022
b0dd93d
fix unit test
Kvicii May 28, 2022
b316207
fix unit test
Kvicii May 31, 2022
d6f240e
fix
Kvicii Jun 1, 2022
f447a3c
Merge branch 'trunk' of github.com:apache/kafka into KAFKA-13592
Kvicii Jun 1, 2022
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
75 changes: 35 additions & 40 deletions core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,7 @@ class ControllerIntegrationTest extends QuorumTestHarness {
TestUtils.waitUntilBrokerMetadataIsPropagated(servers)
val controllerId = TestUtils.waitUntilControllerElected(zkClient)
// Need to make sure the broker we shutdown and startup are not the controller. Otherwise we will send out
// full UpdateMetadataReuqest to all brokers during controller failover.
// full UpdateMetadataRequest to all brokers during controller failover.
val testBroker = servers.filter(e => e.config.brokerId != controllerId).head
val remainingBrokers = servers.filter(_.config.brokerId != testBroker.config.brokerId)
val topic = "topic1"
Expand Down Expand Up @@ -526,7 +526,7 @@ class ControllerIntegrationTest extends QuorumTestHarness {
// create the topic
TestUtils.createTopic(zkClient, topic, partitionReplicaAssignment = expectedReplicaAssignment, servers = servers)

val controllerId = zkClient.getControllerId.get
val controllerId = TestUtils.waitUntilControllerElected(zkClient)
val controller = servers.find(p => p.config.brokerId == controllerId).get.kafkaController
val resultQueue = new LinkedBlockingQueue[Try[collection.Set[TopicPartition]]]()
val controlledShutdownCallback = (controlledShutdownResult: Try[collection.Set[TopicPartition]]) => resultQueue.put(controlledShutdownResult)
Expand Down Expand Up @@ -1253,13 +1253,13 @@ class ControllerIntegrationTest extends QuorumTestHarness {
adminZkClient.createTopic(tp.topic, 1, 1)
waitForPartitionState(tp, firstControllerEpoch, 0, LeaderAndIsr.InitialLeaderEpoch,
"failed to get expected partition state upon topic creation")
val topicIdAfterCreate = zkClient.getTopicIdsForTopics(Set(tp.topic())).get(tp.topic())
val (topicIdAfterCreate, _) = TestUtils.computeUntilTrue(zkClient.getTopicIdsForTopics(Set(tp.topic)).get(tp.topic))(_.nonEmpty)
assertTrue(topicIdAfterCreate.isDefined)
assertEquals(topicIdAfterCreate, servers.head.kafkaController.controllerContext.topicIds.get(tp.topic),
"correct topic ID cannot be found in the controller context")

adminZkClient.addPartitions(tp.topic, assignment, adminZkClient.getBrokerMetadatas(), 2)
val topicIdAfterAddition = zkClient.getTopicIdsForTopics(Set(tp.topic())).get(tp.topic())
val (topicIdAfterAddition, _) = TestUtils.computeUntilTrue(zkClient.getTopicIdsForTopics(Set(tp.topic)).get(tp.topic))(_.nonEmpty)
assertEquals(topicIdAfterCreate, topicIdAfterAddition)
assertEquals(topicIdAfterCreate, servers.head.kafkaController.controllerContext.topicIds.get(tp.topic),
"topic ID changed after partition additions")
Expand All @@ -1279,13 +1279,13 @@ class ControllerIntegrationTest extends QuorumTestHarness {
adminZkClient.createTopic(tp.topic, 1, 1)
waitForPartitionState(tp, firstControllerEpoch, 0, LeaderAndIsr.InitialLeaderEpoch,
"failed to get expected partition state upon topic creation")
val topicIdAfterCreate = zkClient.getTopicIdsForTopics(Set(tp.topic())).get(tp.topic())
val (topicIdAfterCreate, _) = TestUtils.computeUntilTrue(zkClient.getTopicIdsForTopics(Set(tp.topic)).get(tp.topic))(_.nonEmpty)
assertEquals(None, topicIdAfterCreate)
assertEquals(topicIdAfterCreate, servers.head.kafkaController.controllerContext.topicIds.get(tp.topic),
"incorrect topic ID can be found in the controller context")

adminZkClient.addPartitions(tp.topic, assignment, adminZkClient.getBrokerMetadatas(), 2)
val topicIdAfterAddition = zkClient.getTopicIdsForTopics(Set(tp.topic())).get(tp.topic())
val (topicIdAfterAddition, _) = TestUtils.computeUntilTrue(zkClient.getTopicIdsForTopics(Set(tp.topic)).get(tp.topic))(_.nonEmpty)
assertEquals(topicIdAfterCreate, topicIdAfterAddition)
assertEquals(topicIdAfterCreate, servers.head.kafkaController.controllerContext.topicIds.get(tp.topic),
"topic ID changed after partition additions")
Expand Down Expand Up @@ -1406,7 +1406,7 @@ class ControllerIntegrationTest extends QuorumTestHarness {
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())
val (topicIdAfterCreate, _) = TestUtils.computeUntilTrue(zkClient.getTopicIdsForTopics(Set(tp.topic)).get(tp.topic))(_.nonEmpty)
assertEquals(None, topicIdAfterCreate)
val emptyTopicId = controller.controllerContext.topicIds.get("t")
assertEquals(None, emptyTopicId)
Expand Down Expand Up @@ -1467,60 +1467,55 @@ class ControllerIntegrationTest extends QuorumTestHarness {
servers(0).shutdown()
servers(0).awaitShutdown()
servers = makeServers(1)
waitForPartitionState(tp, firstControllerEpoch, 0, LeaderAndIsr.InitialLeaderEpoch,
"failed to get expected partition state upon controller restart")
val topicIdAfterUpgrade = zkClient.getTopicIdsForTopics(Set(tp.topic())).get(tp.topic())
assertEquals(topicIdAfterUpgrade, servers.head.kafkaController.controllerContext.topicIds.get(tp.topic),
"expected same topic ID but it can not be found")
assertEquals(tp.topic(), servers.head.kafkaController.controllerContext.topicNames(topicIdAfterUpgrade.get),
"correct topic name expected but cannot be found in the controller context")

def awaitTopicId(): Uuid = {
// Wait for consistent controller context (Note that `topicIds` is updated before `topicNames`)
val (topicIdOpt, isDefined) = TestUtils.computeUntilTrue {
val topicIdOpt = servers.head.kafkaController.controllerContext.topicIds.get(tp.topic)
topicIdOpt.flatMap { topicId =>
val topicNameOpt = servers.head.kafkaController.controllerContext.topicNames.get(topicId)
if (topicNameOpt.contains(tp.topic)) {
Some(topicId)
} else {
None
}
}
}(_.isDefined)

assertTrue(isDefined, "Timed out waiting for a consistent topicId in controller context")
assertEquals(topicIdOpt, zkClient.getTopicIdsForTopics(Set(tp.topic)).get(tp.topic))
topicIdOpt.get
}

val topicId = awaitTopicId()

// Downgrade back to 2.7
servers(0).shutdown()
servers(0).awaitShutdown()
servers = makeServers(1, interBrokerProtocolVersion = Some(IBP_2_7_IV0))
waitForPartitionState(tp, firstControllerEpoch, 0, LeaderAndIsr.InitialLeaderEpoch,
"failed to get expected partition state upon topic creation")
val topicIdAfterDowngrade = zkClient.getTopicIdsForTopics(Set(tp.topic())).get(tp.topic())
assertTrue(topicIdAfterDowngrade.isDefined)
assertEquals(topicIdAfterUpgrade, topicIdAfterDowngrade,
"expected same topic ID but it can not be found after downgrade")
assertEquals(topicIdAfterDowngrade, servers.head.kafkaController.controllerContext.topicIds.get(tp.topic),
"expected same topic ID in controller context but it is no longer found after downgrade")
assertEquals(tp.topic(), servers.head.kafkaController.controllerContext.topicNames(topicIdAfterUpgrade.get),
"correct topic name expected but cannot be found in the controller context")
assertEquals(topicId, awaitTopicId())

// Reassign partitions
servers(0).kafkaController.eventManager.put(ApiPartitionReassignment(reassignment, _ => ()))
waitForPartitionState(tp, 3, 0, 1,
"failed to get expected partition state upon controller restart")
val topicIdAfterReassignment = zkClient.getTopicIdsForTopics(Set(tp.topic())).get(tp.topic())
assertTrue(topicIdAfterReassignment.isDefined)
assertEquals(topicIdAfterUpgrade, topicIdAfterReassignment,
"expected same topic ID but it can not be found after reassignment")
assertEquals(topicIdAfterUpgrade, servers.head.kafkaController.controllerContext.topicIds.get(tp.topic),
"expected same topic ID in controller context but is no longer found after reassignment")
assertEquals(tp.topic(), servers.head.kafkaController.controllerContext.topicNames(topicIdAfterUpgrade.get),
"correct topic name expected but cannot be found in the controller context")
assertEquals(topicId, awaitTopicId())

// Upgrade back to 2.8
servers(0).shutdown()
servers(0).awaitShutdown()
servers = makeServers(1)
waitForPartitionState(tp, 3, 0, 1,
"failed to get expected partition state upon controller restart")
val topicIdAfterReUpgrade = zkClient.getTopicIdsForTopics(Set(tp.topic())).get(tp.topic())
assertEquals(topicIdAfterUpgrade, topicIdAfterReUpgrade,
"expected same topic ID but it can not be found after re-upgrade")
assertEquals(topicIdAfterReUpgrade, servers.head.kafkaController.controllerContext.topicIds.get(tp.topic),
"topic ID can not be found in controller context after re-upgrading IBP")
assertEquals(tp.topic(), servers.head.kafkaController.controllerContext.topicNames(topicIdAfterReUpgrade.get),
"correct topic name expected but cannot be found in the controller context")
assertEquals(topicId, awaitTopicId())

adminZkClient.deleteTopic(tp.topic)
TestUtils.waitUntilTrue(() => servers.head.kafkaController.controllerContext.topicIds.get(tp.topic).isEmpty,
"topic ID for topic should have been removed from controller context after deletion")
assertTrue(servers.head.kafkaController.controllerContext.topicNames.get(topicIdAfterUpgrade.get).isEmpty)
// Verify removal from controller context (Note that `topicIds` is updated before `topicNames`)
TestUtils.waitUntilTrue(() => !servers.head.kafkaController.controllerContext.topicNames.contains(topicId),
"Timed out waiting for removal of topicId from controller context")
assertEquals(None, servers.head.kafkaController.controllerContext.topicIds.get(tp.topic))
}

private def testControllerMove(fun: () => Unit): Unit = {
Expand Down