Attempt to fix flaky test by waiting on metadata update
Reviewers: dengziming Guozhang Wang
@@ -30,6 +30,7 @@ import org.apache.kafka.common.errors.{ClusterAuthorizationException, InvalidTop
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.requests.MetadataResponse
import org.apache.kafka.common.security.auth.SecurityProtocol
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, TestInfo}
@@ -741,7 +742,18 @@ class TopicCommandIntegrationTest extends KafkaServerTestHarness with Logging wi
try {
killBroker(0)
val aliveServers = brokers.filterNot(_.config.brokerId == 0)
- TestUtils.waitForPartitionMetadata(aliveServers, underMinIsrTopic, 0)
+
+ if (isKRaftTest()) {
+ TestUtils.ensureConsistentKRaftMetadata(aliveServers, controllerServer, "Timeout waiting for topic configs propagating to brokers")
+ } else {
+ TestUtils.waitUntilTrue(
+ () => aliveServers.forall(
+ broker =>
+ broker.metadataCache.getPartitionInfo(underMinIsrTopic, 0).get.isr().size() < 6 &&
+ broker.metadataCache.getPartitionInfo(offlineTopic, 0).get.leader() == MetadataResponse.NO_LEADER_ID),
+ "Timeout waiting for partition metadata propagating to brokers for underMinIsrTopic topic"
+ )
+ }
val output = TestUtils.grabConsoleOutput(
topicService.describeTopic(new TopicCommandOptions(Array("--under-min-isr-partitions"))))
val rows = output.split("\n")
