jolshan commented on a change in pull request #11126:
URL: https://github.com/apache/kafka/pull/11126#discussion_r679387634
##########
File path:
core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
##########
@@ -1115,6 +1115,63 @@ 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
+ val remainingBrokers = servers.filter(_.config.brokerId !=
originalControllerId)
+ val tp = new TopicPartition("t", 0)
+ val assignment = Map(tp.partition -> servers.map(_.config.brokerId))
Review comment:
We can do that. As an aside, we don't store the topic ID on the original
controller since it has a lower IBP when it restarts in this test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
