[kafka] branch trunk updated: MINOR: make sure all partition info is propagated (#13037)

[sh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

showuon pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new c56fec21993 MINOR: make sure all partition info is propagated (#13037)
c56fec21993 is described below

commit c56fec2199398641a8a0a21392bf8135ac637988
Author: Luke Chen <sh...@gmail.com>
AuthorDate: Thu Dec 29 09:58:09 2022 +0800

    MINOR: make sure all partition info is propagated (#13037)
    
    Recently, we have some flaky tests failed with: (ex: build #1457, #1459, #1460
    
    java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.
            at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
            at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999)
            at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165)
            at kafka.utils.TestUtils$.describeTopic(TestUtils.scala:475)
            at kafka.utils.TestUtils$.topicHasSameNumPartitionsAndReplicationFactor(TestUtils.scala:484)
    
    The line failed is when we tried to create a topic but got TopicExistsException, so we want to make sure this topic has the expected partition num and replication factor. It failed while trying to describe this topic, and got UnknownTopicOrPartitionException. That will happen when topic is created but the metadata haven't propagated to all brokers. So, fix it by explicitly waiting for partition info propagated to all brokers before verifying topicHasSameNumPartitionsAndReplicationFactor.
    
    Reviewers: Ismael Juma <is...@juma.me.uk>, dengziming <de...@gmail.com>, Justine Olshan <jolshan@confluent.io >
---
 .../test/scala/unit/kafka/utils/TestUtils.scala    | 22 ++++++++++++++--------
 1 file changed, 14 insertions(+), 8 deletions(-)

diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index 5d1abea0e66..8718ff68575 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -385,7 +385,7 @@ object TestUtils extends Logging {
       config.setProperty(KafkaConfig.LogMessageFormatVersionProp, version.version)
   }
 
- def createAdminClient[B <: KafkaBroker](
+  def createAdminClient[B <: KafkaBroker](
     brokers: Seq[B],
     listenerName: ListenerName,
     adminConfig: Properties = new Properties
@@ -424,7 +424,7 @@ object TestUtils extends Logging {
     }
 
     result.topicId(topic).get()
-}
+  }
 
   def createTopicWithAdmin[B <: KafkaBroker](
     admin: Admin,
@@ -441,6 +441,14 @@ object TestUtils extends Logging {
       replicaAssignment.size
     }
 
+    def isTopicExistsAndHasSameNumPartitionsAndReplicationFactor(cause: Throwable): Boolean = {
+      cause != null &&
+        cause.isInstanceOf[TopicExistsException] &&
+        // wait until all partitions metadata are propagated before verifying partitions number and replication factor
+        !waitForAllPartitionsMetadata(brokers, topic, effectiveNumPartitions).isEmpty &&
+        topicHasSameNumPartitionsAndReplicationFactor(admin, topic, effectiveNumPartitions, replicationFactor)
+    }
+
     try {
       createTopicWithAdminRaw(
         admin,
@@ -451,12 +459,10 @@ object TestUtils extends Logging {
         topicConfig
       )
     } catch {
-      case e: ExecutionException => if (!(e.getCause != null &&
-          e.getCause.isInstanceOf[TopicExistsException] &&
-          topicHasSameNumPartitionsAndReplicationFactor(admin, topic,
-            effectiveNumPartitions, replicationFactor))) {
-        throw e
-      }
+      case e: ExecutionException =>
+        if (!isTopicExistsAndHasSameNumPartitionsAndReplicationFactor(e.getCause)) {
+          throw e
+        }
     }
 
     // wait until we've propagated all partitions metadata to all brokers