KAFKA-4318 Migrate ProducerSendTest to the new consumer #2083

baluchicken · 2016-11-01T09:40:31Z

No description provided.

baluchicken · 2016-11-01T09:46:33Z

ijuma

Thanks for the PR, suggested a few changes.

ijuma · 2016-11-15T13:54:50Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

    val overridingProps = new Properties()
    val numServers = 2
    overridingProps.put(KafkaConfig.NumPartitionsProp, 4.toString)
    TestUtils.createBrokerConfigs(numServers, zkConnect, false, interBrokerSecurityProtocol = Some(securityProtocol),
      trustStoreFile = trustStoreFile, saslProperties = saslProperties).map(KafkaConfig.fromProps(_, overridingProps))
  }

-  private var consumer1: SimpleConsumer = null
-  private var consumer2: SimpleConsumer = null
+  private var consumer1: KafkaConsumer[Array[Byte], Array[Byte]] = _


We can name this simply consumer.

ijuma · 2016-11-15T14:24:52Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

+        }
+        if (iters > maxIters)
+          throw new IllegalStateException("Failed to consume the expected records after " + iters + " iterations.")
+        iters += 1


A more concise and better way to do this is:

TestUtils.waitUntilTrue(() => { for (record <- consumer.poll(50).asScala) records.add(record) records.size == numRecords }, "Failed to receive all expected records from the consumer")

Probably worth creating a helper method and reusing it.

Thanks I created a helper method for this.

ijuma · 2016-11-15T14:26:44Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

+        val records = new ArrayList[ConsumerRecord[Array[Byte], Array[Byte]]]()
+        val maxIters = 50
+        var iters = 0
+        consumer1.seek(new TopicPartition(topic, 0),0)


Space missing after ,. Do we need the seek here given that we are seeing to offset 0?

harshach · 2016-11-23T01:42:25Z

@baluchicken can you address the above comments.

asfbot · 2017-01-02T13:22:31Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/418/
Test FAILed (JDK 8 and Scala 2.12).

baluchicken · 2017-01-02T13:53:18Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

-        val records = new ArrayList[ConsumerRecord[Array[Byte], Array[Byte]]]()
-        val maxIters = 50
-        var iters = 0
-        consumer1.seek(new TopicPartition(topic, 0),0)


@ijuma thanks for the reply. I rechecked the code and the seek here was required, because without that, the poll consumes only 100 records (ignores the already consumed ones), so without that seek, the assert on line 472 failed. This assert is from the original code where we used the old consumer API. I think checking 100 records on every iteration does the same thing, so I rebased that part of the code too, with that I can also remove the seek.

@baluchicken, thanks for the explanation. I think the seek is fine then. The error message is a bit more informative if we consume all of the data instead of just what was produced in the last iteration.

To close the loop on this one, using seek makes the test a lot slower, so the approach in the PR where we just consume the records that have been produced since the last iteration is better.

asfbot · 2017-01-02T14:41:34Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/417/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-02T15:04:56Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/419/
Test PASSed (JDK 8 and Scala 2.11).

ijuma

Thanks for the update. Left a few more comments.

ijuma · 2017-01-02T23:52:38Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

-      val messageSet1 = fetchResponse1.messageSet(topic, partition).iterator.toBuffer
-      assertEquals("Should have fetched " + numRecords + " messages", numRecords, messageSet1.size)
+      val records = pollAllExpectedRecord(numRecords)
+      assertEquals("Should have fetched " + numRecords + " messages", numRecords, records.size)


This is not needed since it will always be true. pollAllExpectedRecord ensures that.

ijuma · 2017-01-02T23:53:15Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

@@ -83,6 +79,16 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness {
    producer
  }

+  protected def pollAllExpectedRecord(numRecords: Int) = {


Should be private. Also, it's good to specify the return type in the method signature. Finally, maybe the method name should be pollUntilNumRecords

ijuma · 2017-01-02T23:53:59Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

+    val records = new ArrayList[ConsumerRecord[Array[Byte], Array[Byte]]]()
+    TestUtils.waitUntilTrue(() => {
+      for( record <- consumer.poll(50).asScala)
+        records.add(record)


These two lines can be replaced by records ++= consumer.poll(50).asScala if we use ArrayBuffer as per my other comment.

ijuma · 2017-01-02T23:58:29Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

+      for( record <- consumer.poll(50).asScala)
+        records.add(record)
+      records.size == numRecords
+    }, "Failed to receive all expected records from the consumer")


Maybe we can tweak this message to be: s"Consumed ${records.size} records before timeout, but expected $numRecords records.

ijuma · 2017-01-03T00:02:14Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

-        assertEquals(i.toLong, messageSet1(i).offset)
+        assertEquals(topic, records(i).topic())
+        assertEquals(partition, records(i).partition)
+        assertEquals(i.toLong, records(i).offset)
      }


A bit more idiomatic:

records.zipWithIndex.foreach { (record, i) => assertEquals(topic, record.topic) assertEquals(partition, record.partition) assertEquals(i.toLong, record.offset) ]

ijuma · 2017-01-03T00:04:14Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

 import java.util.concurrent.TimeUnit

+import collection.JavaConverters._
+import collection.JavaConversions._


JavaConversions is deprecated in Scala 2.12 and we should not use it.

ijuma · 2017-01-03T00:04:53Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

@@ -83,6 +79,16 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness {
    producer
  }

+  protected def pollAllExpectedRecord(numRecords: Int) = {
+    val records = new ArrayList[ConsumerRecord[Array[Byte], Array[Byte]]]()


Is there a reason why we are using ArrayList instead of ArrayBuffer?

ijuma · 2017-01-03T00:05:56Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

@@ -276,20 +282,16 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness {
        assertEquals(partition, recordMetadata.partition)
      }

-      val leader1 = leaders(partition)
+      consumer.subscribe(List(topic))


Is there a reason we are using subscribe instead of assign? The latter is closer to the simple consumer.

ijuma · 2017-01-03T00:07:29Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

-        consumer2.fetch(new FetchRequestBuilder().addFetch(topic, 0, 0, Int.MaxValue).build())
+      val records = new ArrayList[ConsumerRecord[Array[Byte], Array[Byte]]]()
+      for (record <- consumer.poll(50)) {
+        records.add(record)


Why do we do this instead of just using the result of consumer.poll(50)?

ijuma · 2017-01-03T00:09:57Z

core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala

-          expectedNumRecords, fetchResponse.messageSet(topic, 0).size)
+        val records = pollAllExpectedRecord(numRecords)
+        assertEquals("Fetch response to partition 0 should have %d messages.".format(numRecords),
+          numRecords, records.size())


This assert is not needed since it's already ensured by the previous line.

baluchicken · 2017-01-03T14:09:36Z

@ijuma thanks for the review, I updated the PR.

ijuma

Thanks, LGTM. Will merge to trunk with a couple of trivial tweaks.

baluchicken · 2017-01-03T14:28:44Z

Thanks, and also thanks for your patience, I learned a lot from your comments.

asfbot · 2017-01-03T15:38:13Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/435/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-01-03T15:41:37Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/436/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-03T15:47:07Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/437/
Test FAILed (JDK 8 and Scala 2.11).

ijuma · 2017-01-03T15:51:30Z

This PR only changes a test and that test passed. The builds that failed are unrelated to this change.

Author: Balint Molnar <balintmolnar91@gmail.com> Reviewers: Ismael Juma <ismael@juma.me.uk> Closes apache#2083 from baluchicken/KAFKA-4318

ijuma reviewed Nov 15, 2016

View reviewed changes

baluchicken force-pushed the KAFKA-4318 branch from 6376d7e to bff9eec Compare January 2, 2017 13:18

baluchicken commented Jan 2, 2017

View reviewed changes

ijuma reviewed Jan 3, 2017

View reviewed changes

baluchicken added 3 commits January 3, 2017 15:03

KAFKA-4318 Migrate ProducerSendTest to the new consumer

751b10e

Address the requested changes

44beee1

Address the requested changes apache#2

dfe5802

baluchicken force-pushed the KAFKA-4318 branch from bff9eec to dfe5802 Compare January 3, 2017 14:05

ijuma approved these changes Jan 3, 2017

View reviewed changes

asfgit closed this in 7411ed8 Jan 3, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-4318 Migrate ProducerSendTest to the new consumer #2083

KAFKA-4318 Migrate ProducerSendTest to the new consumer #2083

baluchicken commented Nov 1, 2016

baluchicken commented Nov 1, 2016

ijuma left a comment

ijuma Nov 15, 2016

ijuma Nov 15, 2016

baluchicken Jan 2, 2017

ijuma Nov 15, 2016

harshach commented Nov 23, 2016

asfbot commented Jan 2, 2017

baluchicken Jan 2, 2017

ijuma Jan 3, 2017

ijuma Jan 3, 2017

asfbot commented Jan 2, 2017

asfbot commented Jan 2, 2017

ijuma left a comment

ijuma Jan 2, 2017

ijuma Jan 2, 2017

ijuma Jan 2, 2017

ijuma Jan 2, 2017

ijuma Jan 3, 2017

ijuma Jan 3, 2017

ijuma Jan 3, 2017

ijuma Jan 3, 2017

ijuma Jan 3, 2017

ijuma Jan 3, 2017

baluchicken commented Jan 3, 2017

ijuma left a comment

baluchicken commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

ijuma commented Jan 3, 2017

KAFKA-4318 Migrate ProducerSendTest to the new consumer #2083

KAFKA-4318 Migrate ProducerSendTest to the new consumer #2083

Conversation

baluchicken commented Nov 1, 2016

baluchicken commented Nov 1, 2016

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

harshach commented Nov 23, 2016

asfbot commented Jan 2, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Jan 2, 2017

asfbot commented Jan 2, 2017

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

baluchicken commented Jan 3, 2017

ijuma left a comment

Choose a reason for hiding this comment

baluchicken commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

ijuma commented Jan 3, 2017