Fix flaky tests in TopicCommandWithAdminClientTest
This change adds waits for metadata updates after killing the broker in order to make the tests more stable.
Author: Viktor Somogyi-Vass
Reviewers: Manikumar Reddy
Closes from viktorsomogyi/flaky-min-isr-test
@@ -28,6 +28,9 @@ import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.admin.{ListTopicsOptions, NewTopic, AdminClient => JAdminClient}
import org.apache.kafka.common.config.{ConfigException, ConfigResource, TopicConfig}
import org.apache.kafka.common.internals.Topic
+import org.apache.kafka.common.network.ListenerName
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.security.auth.SecurityProtocol
import org.junit.Assert.{assertEquals, assertFalse, assertTrue}
import org.junit.{After, Before, Rule, Test}
import org.junit.rules.TestName
@@ -515,7 +518,33 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin
waitForTopicCreated(testTopicName)
try {
+ // check which partition is on broker 0 which we'll kill
+ val testTopicDescription = adminClient.describeTopics(Collections.singletonList(testTopicName))
+ .all().get().asScala(testTopicName)
+ val partitionOnBroker0 = testTopicDescription.partitions().asScala.find(_.leader().id() == 0).get.partition()
+
killBroker(0)
+
+ // wait until the topic metadata for the test topic is propagated to each alive broker
+ TestUtils.waitUntilTrue(() => {
+ servers
+ .filterNot(_.config.brokerId == 0)
+ .foldLeft(true) {
+ (result, server) => {
+ val topicMetadatas = server.dataPlaneRequestProcessor.metadataCache
+ .getTopicMetadata(Set(testTopicName), ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT))
+ val testPartitionMetadata = topicMetadatas.find(_.topic().equals(testTopicName)).get.partitionMetadata().asScala.find(_.partition() == partitionOnBroker0)
+ testPartitionMetadata match {
+ case None => fail(s"Partition metadata is not found in metadata cache")
+ case Some(metadata) => {
+ result && metadata.error() == Errors.LEADER_NOT_AVAILABLE
+ }
+ }
+ }
+ }
+ }, s"Partition metadata for $testTopicName is not propagated")
+
+ // grab the console output and assert
val output = TestUtils.grabConsoleOutput(
topicService.describeTopic(new TopicCommandOptions(Array("--topic", testTopicName, "--unavailable-partitions"))))
val rows = output.split("\n")
@@ -534,6 +563,8 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin
try {
killBroker(0)
+ val aliveServers = servers.filterNot(_.config.brokerId == 0)
+ TestUtils.waitUntilMetadataIsPropagated(aliveServers, testTopicName, 0)
val output = TestUtils.grabConsoleOutput(
topicService.describeTopic(new TopicCommandOptions(Array("--under-replicated-partitions"))))
val rows = output.split("\n")
@@ -554,6 +585,8 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin
try {
killBroker(0)
+ val aliveServers = servers.filterNot(_.config.brokerId == 0)
+ TestUtils.waitUntilMetadataIsPropagated(aliveServers, testTopicName, 0)
val output = TestUtils.grabConsoleOutput(
topicService.describeTopic(new TopicCommandOptions(Array("--under-min-isr-partitions"))))
val rows = output.split("\n")
@@ -596,6 +629,8 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin
try {
killBroker(0)
+ val aliveServers = servers.filterNot(_.config.brokerId == 0)
+ TestUtils.waitUntilMetadataIsPropagated(aliveServers, underMinIsrTopic, 0)
val output = TestUtils.grabConsoleOutput(
topicService.describeTopic(new TopicCommandOptions(Array("--under-min-isr-partitions"))))
val rows = output.split("\n")
