Wait for follower to join the ISR in testUncleanLeaderElectionDisabled Test
Author: Manikumar Reddy
Reviewers: Jason Gustafson Boyang Chen
Closes from omkreddy/unclean-leader
@@ -219,16 +219,16 @@ class UncleanLeaderElectionTest extends ZooKeeperTestHarness {
assertEquals(List("first"), consumeAllMessages(topic, 1))
// shutdown follower server
- servers.filter(server => server.config.brokerId == followerId).map(server => shutdownServer(server))
+ servers.filter(server => server.config.brokerId == followerId).foreach(server => shutdownServer(server))
produceMessage(servers, topic, "second")
assertEquals(List("first", "second"), consumeAllMessages(topic, 2))
//remove any previous unclean election metric
- servers.map(server => server.kafkaController.controllerContext.stats.removeMetric("UncleanLeaderElectionsPerSec"))
+ servers.foreach(server => server.kafkaController.controllerContext.stats.removeMetric("UncleanLeaderElectionsPerSec"))
// shutdown leader and then restart follower
- servers.filter(server => server.config.brokerId == leaderId).map(server => shutdownServer(server))
+ servers.filter(server => server.config.brokerId == leaderId).foreach(server => shutdownServer(server))
val followerServer = servers.find(_.config.brokerId == followerId).get
followerServer.startup()
@@ -247,13 +247,17 @@ class UncleanLeaderElectionTest extends ZooKeeperTestHarness {
assertEquals(List.empty[String], consumeAllMessages(topic, 0))
// restart leader temporarily to send a successfully replicated message
- servers.filter(server => server.config.brokerId == leaderId).map(server => server.startup())
+ servers.filter(server => server.config.brokerId == leaderId).foreach(server => server.startup())
waitUntilLeaderIsElectedOrChanged(zkClient, topic, partitionId, newLeaderOpt = Some(leaderId))
produceMessage(servers, topic, "third")
- waitUntilMetadataIsPropagated(servers, topic, partitionId)
- servers.filter(server => server.config.brokerId == leaderId).map(server => shutdownServer(server))
+ //make sure follower server joins the ISR
+ TestUtils.waitUntilTrue(() => {
+ val partitionInfoOpt = followerServer.metadataCache.getPartitionInfo(topic, partitionId)
+ partitionInfoOpt.isDefined && partitionInfoOpt.get.basePartitionState.isr.contains(followerId)
+ }, "Inconsistent metadata after first server startup")
+ servers.filter(server => server.config.brokerId == leaderId).foreach(server => shutdownServer(server))
// verify clean leader transition to ISR follower
waitUntilLeaderIsElectedOrChanged(zkClient, topic, partitionId, newLeaderOpt = Some(followerId))
