KAFKA-4761: Fix producer regression handling small or zero batch size #2545

hachikuji · 2017-02-13T19:17:15Z

No description provided.

asfbot · 2017-02-13T19:19:47Z

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

asfbot · 2017-02-13T19:20:30Z

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

apurvam

Would it make sense to add the original reproducing test case? https://issues.apache.org/jira/secure/attachment/12852383/KafkaProducerTest.java

asfbot · 2017-02-13T19:23:58Z

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

guozhangwang

LGTM. Thanks @hachikuji

vahidhashemian · 2017-02-13T19:35:33Z

LGTM too. As @apurvam mentioned It would be great if the unit test in the JIRA can be added too.

apurvam

LGTM. I asked a question to clarify the cause of the original bug and why this is a correct fix. It is more for documentation and my own learning.

apurvam · 2017-02-13T19:36:57Z

clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java

@@ -376,7 +390,7 @@ public boolean isClosed() {
    }

    public boolean isFull() {
-        return isClosed() || this.writeLimit <= estimatedBytesWritten();
+        return isClosed() || (this.numRecords > 0 && this.writeLimit <= estimatedBytesWritten());


I assume that writeLimit is based of batch size?

So, in cases where writeLimit == 0, even an empty batch would be considered full, triggering the NPE on send?

Is that correct?

Yes, correct. The difference from the old code is really in the hasRoomFor method, which now uses isFull().

Should we add a comment here?

hachikuji · 2017-02-13T19:42:34Z

@apurvam @vahidhashemian The integration test I've added effectively does the same thing, right?

apurvam · 2017-02-13T19:43:26Z

@hachikuji, yes it does. Ignore my previous comment.

hachikuji · 2017-02-13T19:44:35Z

@guozhangwang Thanks for reviewing. I'll merge to trunk and 0.10.2 shortly.

vahidhashemian · 2017-02-13T19:45:44Z

@hachikuji Yes, sorry I missed the last file you updated.

ijuma

Thanks for the PR, left a few minor comments, looks good otherwise.

ijuma · 2017-02-13T19:55:47Z

clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java

@@ -376,7 +390,7 @@ public boolean isClosed() {
    }

    public boolean isFull() {
-        return isClosed() || this.writeLimit <= estimatedBytesWritten();
+        return isClosed() || (this.numRecords > 0 && this.writeLimit <= estimatedBytesWritten());


Should we add a comment here?

ijuma · 2017-02-13T19:56:25Z

clients/src/test/java/org/apache/kafka/test/TestUtils.java

@@ -308,6 +308,10 @@ public static void isValidClusterId(String clusterId) {
        assertEquals(toList(it1.iterator()), toList(it2.iterator()));
    }

+    public static <T> List<T> toList(Iterable<? extends T> iterable) {
+        return toList(iterable.iterator());


Nit: one of the checkEquals can maybe use this method now.

ijuma · 2017-02-13T19:57:18Z

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

@@ -175,6 +175,29 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness {
    sendAndVerifyTimestamp(producer, TimestampType.CREATE_TIME)
  }

+  protected def sendAndVerify(producer: KafkaProducer[Array[Byte], Array[Byte]], numRecords: Int = numRecords) {
+    val partition = new Integer(0)


Is there a reason why we're not just saying val partition = 0?

ijuma · 2017-02-13T19:58:59Z

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

+        val record = new ProducerRecord(topic, partition, "key".getBytes, "value".getBytes)
+        (record, producer.send(record))
+      }
+      producer.close(20000L, TimeUnit.MILLISECONDS)


Should we allow this timeout to be passed as a parameter (with a default) as it may need to be different depending on numRecords?

ijuma · 2017-02-13T19:59:25Z

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

+      TestUtils.createTopic(zkUtils, topic, 1, 2, servers)
+
+      val recordAndFutures = for (i <- 1 to numRecords) yield {
+        val record = new ProducerRecord(topic, partition, "key".getBytes, "value".getBytes)


Would it not be a little better to include i in the key and value from a debugging perspective?

asfbot · 2017-02-13T20:07:26Z

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

asfbot · 2017-02-13T20:10:57Z

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

asfbot · 2017-02-13T20:11:17Z

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

asfbot · 2017-02-13T21:07:21Z

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

asfbot · 2017-02-13T21:47:59Z

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

asfbot · 2017-02-13T21:51:16Z

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

ijuma

Thanks for the updates, LGTM.

Author: Jason Gustafson <jason@confluent.io> Reviewers: Apurva Mehta <apurva.1618@gmail.com>, Vahid Hashemian <vahidhashemian@us.ibm.com>, Ismael Juma <ismael@juma.me.uk>, Guozhang Wang <wangguoz@gmail.com> Closes #2545 from hachikuji/KAFKA-4761 (cherry picked from commit 3b36d5c) Signed-off-by: Jason Gustafson <jason@confluent.io>

swen-fuhrmann · 2017-02-14T08:01:06Z

Thanks for the fast fix!
I tested my code with the pull request and it works without exception.

hachikuji · 2017-02-14T17:43:37Z

@smoczarski Awesome, thanks for the update!

Author: Jason Gustafson <jason@confluent.io> Reviewers: Apurva Mehta <apurva.1618@gmail.com>, Vahid Hashemian <vahidhashemian@us.ibm.com>, Ismael Juma <ismael@juma.me.uk>, Guozhang Wang <wangguoz@gmail.com> Closes apache#2545 from hachikuji/KAFKA-4761

hachikuji changed the title ~~KAFKA-4671: Fix producer regression handling small or zero batch size~~ KAFKA-4761: Fix producer regression handling small or zero batch size Feb 13, 2017

KAFKA-4761: Fix producer regression handling small or zero batch size

54528fa

hachikuji force-pushed the KAFKA-4761 branch from d20a5d3 to 54528fa Compare February 13, 2017 19:20

apurvam reviewed Feb 13, 2017

View reviewed changes

Fix checkstyle

7858a3e

guozhangwang approved these changes Feb 13, 2017

View reviewed changes

apurvam approved these changes Feb 13, 2017

View reviewed changes

ijuma reviewed Feb 13, 2017

View reviewed changes

Jason Gustafson added 2 commits February 13, 2017 12:43

Minor cleanups from Ismael's review

a5f40f8

Add comment in MemoryRecordsBuilder

a5a1ed5

ijuma approved these changes Feb 13, 2017

View reviewed changes

asfgit closed this in 3b36d5c Feb 13, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-4761: Fix producer regression handling small or zero batch size #2545

KAFKA-4761: Fix producer regression handling small or zero batch size #2545

hachikuji commented Feb 13, 2017

asfbot commented Feb 13, 2017

asfbot commented Feb 13, 2017

apurvam left a comment

asfbot commented Feb 13, 2017

guozhangwang left a comment

vahidhashemian commented Feb 13, 2017

apurvam left a comment

apurvam Feb 13, 2017

hachikuji Feb 13, 2017

ijuma Feb 13, 2017

hachikuji commented Feb 13, 2017

apurvam commented Feb 13, 2017

hachikuji commented Feb 13, 2017

vahidhashemian commented Feb 13, 2017

ijuma left a comment

ijuma Feb 13, 2017

ijuma Feb 13, 2017

ijuma Feb 13, 2017

ijuma Feb 13, 2017

ijuma Feb 13, 2017

asfbot commented Feb 13, 2017

asfbot commented Feb 13, 2017

asfbot commented Feb 13, 2017

asfbot commented Feb 13, 2017

asfbot commented Feb 13, 2017

asfbot commented Feb 13, 2017

ijuma left a comment

swen-fuhrmann commented Feb 14, 2017

hachikuji commented Feb 14, 2017

KAFKA-4761: Fix producer regression handling small or zero batch size #2545

KAFKA-4761: Fix producer regression handling small or zero batch size #2545

Conversation

hachikuji commented Feb 13, 2017

asfbot commented Feb 13, 2017

asfbot commented Feb 13, 2017

apurvam left a comment

Choose a reason for hiding this comment

asfbot commented Feb 13, 2017

guozhangwang left a comment

Choose a reason for hiding this comment

vahidhashemian commented Feb 13, 2017

apurvam 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

hachikuji commented Feb 13, 2017

apurvam commented Feb 13, 2017

hachikuji commented Feb 13, 2017

vahidhashemian commented Feb 13, 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

asfbot commented Feb 13, 2017

asfbot commented Feb 13, 2017

asfbot commented Feb 13, 2017

asfbot commented Feb 13, 2017

asfbot commented Feb 13, 2017

asfbot commented Feb 13, 2017

ijuma left a comment

Choose a reason for hiding this comment

swen-fuhrmann commented Feb 14, 2017

hachikuji commented Feb 14, 2017