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 4 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
37 changes: 18 additions & 19 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 @@ -1221,59 +1221,58 @@ 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 id = servers.head.kafkaController.controllerContext.topicIds.get(tp.topic)
val topicIdAfterCreate = zkClient.getTopicIdsForTopics(Set(tp.topic)).get(tp.topic)
assertTrue(topicIdAfterCreate.isEmpty)
assertEquals(topicIdAfterCreate, id,
assertEquals(topicIdAfterCreate, servers.head.kafkaController.controllerContext.topicIds.get(tp.topic),
"expected no topic ID, but one existed")

// Upgrade to IBP 2.8
servers(0).shutdown()
servers(0).awaitShutdown()
servers.head.shutdown()
servers.head.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())
val (topicIdAfterUpgrade, _) = TestUtils.computeUntilTrue(zkClient.getTopicIdsForTopics(Set(tp.topic)).get(tp.topic))(_.nonEmpty)
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),
assertEquals(tp.topic, servers.head.kafkaController.controllerContext.topicNames(topicIdAfterUpgrade.get),
"correct topic name expected but cannot be found in the controller context")

// Downgrade back to 2.7
servers(0).shutdown()
servers(0).awaitShutdown()
servers.head.shutdown()
servers.head.awaitShutdown()
servers = makeServers(1, interBrokerProtocolVersion = Some(KAFKA_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())
val (topicIdAfterDowngrade, _) = TestUtils.computeUntilTrue(zkClient.getTopicIdsForTopics(Set(tp.topic)).get(tp.topic))(_.nonEmpty)
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),
assertEquals(tp.topic, servers.head.kafkaController.controllerContext.topicNames(topicIdAfterUpgrade.get),
"correct topic name expected but cannot be found in the controller context")

// Reassign partitions
servers(0).kafkaController.eventManager.put(ApiPartitionReassignment(reassignment, _ => ()))
servers.head.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())
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),
assertEquals(tp.topic, servers.head.kafkaController.controllerContext.topicNames(topicIdAfterUpgrade.get),
"correct topic name expected but cannot be found in the controller context")

// Upgrade back to 2.8
servers(0).shutdown()
servers(0).awaitShutdown()
servers.head.shutdown()
servers.head.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())
val (topicIdAfterReUpgrade, _) = TestUtils.computeUntilTrue(zkClient.getTopicIdsForTopics(Set(tp.topic)).get(tp.topic))(_.nonEmpty)
Comment thread
Kvicii marked this conversation as resolved.
Outdated
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),
Expand All @@ -1282,9 +1281,9 @@ class ControllerIntegrationTest extends QuorumTestHarness {
"correct topic name expected but cannot be found in the controller context")

adminZkClient.deleteTopic(tp.topic)
TestUtils.waitUntilTrue(() => servers.head.kafkaController.controllerContext.topicIds.get(tp.topic).isEmpty,
TestUtils.waitUntilTrue(() => !servers.head.kafkaController.controllerContext.topicIds.contains(tp.topic),
"topic ID for topic should have been removed from controller context after deletion")
assertTrue(servers.head.kafkaController.controllerContext.topicNames.get(topicIdAfterUpgrade.get).isEmpty)
assertTrue(!servers.head.kafkaController.controllerContext.topicNames.contains(topicIdAfterUpgrade.get))
}

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