chia7712 commented on code in PR #20068:
URL: https://github.com/apache/kafka/pull/20068#discussion_r2280979477
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -3034,6 +3034,21 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
}
}
+ /**
+ * Waits until the metadata for the given partition has fully propagated and
become consistent across all brokers.
+ *
+ * @param partition The partition whose leader metadata should be verified
across all brokers.
+ */
+ def waitForBrokerMetadataPropagation(partition: TopicPartition): Unit = {
+ while (true) {
+ val headBrokerLeaderId =
brokers.head.metadataCache.getPartitionLeaderEndpoint(partition.topic,
partition.partition(), listenerName).get.id()
Review Comment:
It seems this helper method is used to ensure that "all" brokers see the
same leader, right? If so, we could simplify the code
```scala
def waitForBrokerMetadataPropagation(partition: TopicPartition): Unit = {
while
(brokers.map(_.metadataCache.getPartitionLeaderEndpoint(partition.topic,
partition.partition(), listenerName))
.filter(_.isPresent)
.map(_.get())
.toSet.size != 1)
TimeUnit.MILLISECONDS.sleep(300)
}
```
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
