Fix transient failure in ControllerIntegrationTest.testControllerMoveIncrementsControllerEpoch
The tests previously ignored the fact that the controller does not atomically
create the /controller znode and create/increment the /controller_epoch znode.
Author: Onur Karaman
Reviewers: Ismael Juma
Closes from onurkaraman/KAFKA-5180
@@ -43,35 +43,30 @@ class ControllerIntegrationTest extends ZooKeeperTestHarness {
def testEmptyCluster(): Unit = {
servers = makeServers(1)
TestUtils.waitUntilTrue(() => zkUtils.pathExists(ZkUtils.ControllerPath), "failed to elect a controller")
- TestUtils.waitUntilTrue(() => zkUtils.readData(ZkUtils.ControllerEpochPath)._1.toInt == KafkaController.InitialControllerEpoch,
- "broker failed to set controller epoch")
+ waitUntilControllerEpoch(KafkaController.InitialControllerEpoch, "broker failed to set controller epoch")
}
@Test
def testControllerEpochPersistsWhenAllBrokersDown(): Unit = {
servers = makeServers(1)
TestUtils.waitUntilTrue(() => zkUtils.pathExists(ZkUtils.ControllerPath), "failed to elect a controller")
- TestUtils.waitUntilTrue(() => zkUtils.readData(ZkUtils.ControllerEpochPath)._1.toInt == KafkaController.InitialControllerEpoch,
- "broker failed to set controller epoch")
+ waitUntilControllerEpoch(KafkaController.InitialControllerEpoch, "broker failed to set controller epoch")
servers.head.shutdown()
servers.head.awaitShutdown()
TestUtils.waitUntilTrue(() => !zkUtils.pathExists(ZkUtils.ControllerPath), "failed to kill controller")
- TestUtils.waitUntilTrue(() => zkUtils.readData(ZkUtils.ControllerEpochPath)._1.toInt == KafkaController.InitialControllerEpoch,
- "controller epoch was not persisted after broker failure")
+ waitUntilControllerEpoch(KafkaController.InitialControllerEpoch, "controller epoch was not persisted after broker failure")
}
@Test
def testControllerMoveIncrementsControllerEpoch(): Unit = {
servers = makeServers(1)
TestUtils.waitUntilTrue(() => zkUtils.pathExists(ZkUtils.ControllerPath), "failed to elect a controller")
- TestUtils.waitUntilTrue(() => zkUtils.readData(ZkUtils.ControllerEpochPath)._1.toInt == KafkaController.InitialControllerEpoch,
- "broker failed to set controller epoch")
+ waitUntilControllerEpoch(KafkaController.InitialControllerEpoch, "broker failed to set controller epoch")
servers.head.shutdown()
servers.head.awaitShutdown()
servers.head.startup()
TestUtils.waitUntilTrue(() => zkUtils.pathExists(ZkUtils.ControllerPath), "failed to elect a controller")
- TestUtils.waitUntilTrue(() => zkUtils.readData(ZkUtils.ControllerEpochPath)._1.toInt == KafkaController.InitialControllerEpoch + 1,
- "controller epoch was not persisted after broker failure")
+ waitUntilControllerEpoch(KafkaController.InitialControllerEpoch + 1, "controller epoch was not incremented after controller move")
}
@Test
@@ -284,6 +279,10 @@ class ControllerIntegrationTest extends ZooKeeperTestHarness {
}, "failed to get expected partition state after entire isr went offline")
}
+ private def waitUntilControllerEpoch(epoch: Int, message: String): Unit = {
+ TestUtils.waitUntilTrue(() => zkUtils.readDataMaybeNull(ZkUtils.ControllerEpochPath)._1.map(_.toInt) == Some(epoch), message)
+ }
+
private def waitForPartitionState(tp: TopicAndPartition,
controllerEpoch: Int,
leader: Int,
