[kafka] branch trunk updated: MINOR: Fix transiently failing consumer group admin integration test (#5067)

[jg...@apache.org]

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

jgus 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 e884720  MINOR: Fix transiently failing consumer group admin integration test (#5067)
e884720 is described below

commit e8847205f93851b544899e08e00e17ec2d66f617
Author: Jason Gustafson <ja...@confluent.io>
AuthorDate: Tue May 22 16:24:54 2018 -0700

    MINOR: Fix transiently failing consumer group admin integration test (#5067)
    
    Since the producer is using retries=0, we need to await topic creation before sending any records.
    
    Reviewers: Ismael Juma <is...@juma.me.uk>
---
 .../scala/integration/kafka/api/AdminClientIntegrationTest.scala   | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
index e7dd108..24daa86 100644
--- a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
@@ -963,8 +963,8 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging {
   }
 
   /**
-    * Test the consumer group APIs.
-    */
+   * Test the consumer group APIs.
+   */
   @Test
   def testConsumerGroups(): Unit = {
     val config = createConfig()
@@ -979,6 +979,8 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging {
       val testNumPartitions = 2
       client.createTopics(Collections.singleton(
         new NewTopic(testTopicName, testNumPartitions, 1))).all().get()
+      waitForTopics(client, List(testTopicName), List())
+
       val producer = createNewProducer
       try {
         producer.send(new ProducerRecord(testTopicName, 0, null, null)).get()
@@ -1044,7 +1046,6 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging {
           assertEquals(2, result.all().get().size())
 
           // Test listConsumerGroupOffsets
-          val parts = client.listConsumerGroupOffsets(testGroupId).partitionsToOffsetAndMetadata().get()
           TestUtils.waitUntilTrue(() => {
             val parts = client.listConsumerGroupOffsets(testGroupId).partitionsToOffsetAndMetadata().get()
             val part = new TopicPartition(testTopicName, 0)

-- 
To stop receiving notification emails like this one, please contact
jgus@apache.org.