KAFKA-5429: Ignore produce response if batch was previously aborted #3300

hachikuji · 2017-06-12T01:12:16Z

No description provided.

asfbot · 2017-06-12T02:37:03Z

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

asfbot · 2017-06-12T02:44:22Z

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

ijuma · 2017-06-12T10:54:08Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java

    private boolean retry;

+    private enum FinalState { ABORTED, FAILED, SUCCEEDED };
+    private AtomicReference<FinalState> finalState = new AtomicReference<>(null);


final missing.

ijuma · 2017-06-12T10:55:20Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java

+        if (!this.finalState.compareAndSet(null, finalState)) {
+            if (this.finalState.get() == FinalState.ABORTED) {
+                log.debug("ProduceResponse returned for {} after batch had already been aborted.", topicPartition);
+                return;


So, it's expected that we may call done (successful or failed) after aborting and we should just ignore the response, in that case?

Yeah, not much else we can do since we've already signaled the future and invoked callbacks. I think it would be justifiable to elevate the log level to info if that helps.

I'd just elevate the log level if there's user visible impact. Supposedly, this happens with transactions and the produced data won't be committed?

Yes, that's right. Another case might be on producer shutdown, but that might be much harder to hit.

The log level is a tricky one. In the producer, messages for errors are all at debug level (for instance when we transition to error state in the transaction manager). So having this higher than debug may not add much value.

ijuma · 2017-06-12T10:57:52Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java

@@ -366,7 +384,7 @@ public void close() {
        }
    }

-    public void abort() {
+    public void abortRecordAppends() {


It may be worth adding a comment explaining how abort and abortRecordAppends differ in one of the methods and add a reference in the other one. It seems like the reason we have two separate methods is that we do one of them with the lock held and the other with no lock held (as it invokes callbacks).

apurvam

LGTM!

apurvam · 2017-06-12T17:59:47Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java

+        if (!this.finalState.compareAndSet(null, finalState)) {
+            if (this.finalState.get() == FinalState.ABORTED) {
+                log.debug("ProduceResponse returned for {} after batch had already been aborted.", topicPartition);
+                return;


The log level is a tricky one. In the producer, messages for errors are all at debug level (for instance when we transition to error state in the transaction manager). So having this higher than debug may not add much value.

asfbot · 2017-06-12T19:53:17Z

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

asfbot · 2017-06-12T20:12:29Z

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

ijuma

Thanks for the updates, a few comments/questions.

ijuma · 2017-06-12T22:06:28Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java

    private boolean retry;

+    private enum FinalState { ABORTED, FAILED, SUCCEEDED };
+    private final AtomicReference<FinalState> finalState = new AtomicReference<>(null);


Nit: should this be with other final fields?

ijuma · 2017-06-12T22:09:16Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java

+     * @param exception The exception to use to complete the future and awaiting callbacks.
+     */
+    public void abort(RuntimeException exception) {
+        abortRecordAppends();


Is this safe? It seems like we should document abort and done as thread-safe and hence we should not invoke abortRecordAppends, which is not thread-safe. Or am I missing something?

Ack. The code seemed weird without it, but it's not actually needed.

ijuma · 2017-06-12T22:10:33Z

clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerBatchTest.java

+
+        KafkaException exception = new KafkaException();
+        batch.abort(exception);
+


Should we check future.isDone here too?

ijuma · 2017-06-12T22:13:28Z

clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerBatchTest.java

@@ -55,6 +59,67 @@ public void testChecksumNullForMagicV2() {
    }

    @Test
+    public void testBatchAbort() throws Exception {


Probably need one like this but where we call done with an exception.

ijuma

Thanks for the updates, LGTM.

asfbot · 2017-06-12T23:25:38Z

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

asfbot · 2017-06-12T23:25:49Z

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

hachikuji · 2017-06-12T23:28:27Z

Merging to trunk and 0.11.0.

Author: Jason Gustafson <jason@confluent.io> Reviewers: Apurva Mehta <apurva@confluent.io>, Ismael Juma <ismael@juma.me.uk> Closes #3300 from hachikuji/KAFKA-5429 (cherry picked from commit 6c92fc5) Signed-off-by: Jason Gustafson <jason@confluent.io>

KAFKA-5429: Ignore produce response if batch was previously aborted

e7caaac

hachikuji force-pushed the KAFKA-5429 branch from 0c270f1 to e7caaac Compare June 12, 2017 01:13

ijuma reviewed Jun 12, 2017

View reviewed changes

apurvam approved these changes Jun 12, 2017

View reviewed changes

Add a few comments on abort usage

b624454

ijuma reviewed Jun 12, 2017

View reviewed changes

A few improvements from review

e9c1679

ijuma approved these changes Jun 12, 2017

View reviewed changes

asfgit closed this in 6c92fc5 Jun 12, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-5429: Ignore produce response if batch was previously aborted #3300

KAFKA-5429: Ignore produce response if batch was previously aborted #3300

hachikuji commented Jun 12, 2017

asfbot commented Jun 12, 2017

asfbot commented Jun 12, 2017

ijuma Jun 12, 2017

ijuma Jun 12, 2017

hachikuji Jun 12, 2017

ijuma Jun 12, 2017

hachikuji Jun 12, 2017

apurvam Jun 12, 2017

ijuma Jun 12, 2017

apurvam left a comment

apurvam Jun 12, 2017

asfbot commented Jun 12, 2017

asfbot commented Jun 12, 2017

ijuma left a comment

ijuma Jun 12, 2017

ijuma Jun 12, 2017

hachikuji Jun 12, 2017

ijuma Jun 12, 2017

ijuma Jun 12, 2017

ijuma left a comment

asfbot commented Jun 12, 2017

asfbot commented Jun 12, 2017

hachikuji commented Jun 12, 2017


		KafkaException exception = new KafkaException();
		batch.abort(exception);

KAFKA-5429: Ignore produce response if batch was previously aborted #3300

KAFKA-5429: Ignore produce response if batch was previously aborted #3300

Conversation

hachikuji commented Jun 12, 2017

asfbot commented Jun 12, 2017

asfbot commented Jun 12, 2017

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

apurvam left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Jun 12, 2017

asfbot commented Jun 12, 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

ijuma left a comment

Choose a reason for hiding this comment

asfbot commented Jun 12, 2017

asfbot commented Jun 12, 2017

hachikuji commented Jun 12, 2017