root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-13746: Attempt to fix flaky test by waiting on metadata update (#12104) 

Reviewers: dengziming <dengziming1993@gmail.com>, Guozhang Wang <wangguoz@gmail.com>
This commit is contained in:
vamossagar12 2022-05-14 05:39:47 +05:30 committed by GitHub
parent 467bce04ae
commit f96e381387
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 13 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
core/src/test/scala/integration/kafka/admin/TopicCommandIntegrationTest.scala
View File

@ -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")