showuon commented on a change in pull request #9777:
URL: https://github.com/apache/kafka/pull/9777#discussion_r567650065
##########
File path: core/src/test/scala/unit/kafka/utils/TestUtils.scala
##########
@@ -885,6 +891,24 @@ object TestUtils extends Logging {
), "Timed out waiting for broker metadata to propagate to all servers",
timeout)
}
+ /**
+ * Wait until the expected number of partitions is in the metadata cache in
each broker.
+ *
+ * @param servers The list of servers that the metadata should reach to
+ * @param topic The topic name
+ * @param expectedNumPartitions The expected number of partitions
+ */
+ def waitUntilMetadataIsPropagatedWithExpectedSize(servers: Seq[KafkaServer],
topic: String, expectedNumPartitions: Int): Unit = {
+ waitUntilTrue(
+ () => servers.forall { server =>
+ server.dataPlaneRequestProcessor.metadataCache.numPartitions(topic)
match {
Review comment:
Good suggestion. Updated.
##########
File path: core/src/test/scala/unit/kafka/utils/TestUtils.scala
##########
@@ -885,6 +891,24 @@ object TestUtils extends Logging {
), "Timed out waiting for broker metadata to propagate to all servers",
timeout)
}
+ /**
+ * Wait until the expected number of partitions is in the metadata cache in
each broker.
+ *
+ * @param servers The list of servers that the metadata should reach to
+ * @param topic The topic name
+ * @param expectedNumPartitions The expected number of partitions
+ */
+ def waitUntilMetadataIsPropagatedWithExpectedSize(servers: Seq[KafkaServer],
topic: String, expectedNumPartitions: Int): Unit = {
Review comment:
Good suggestion. Updated.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
