KAFKA-5364: Don't fail producer if drained partition is not yet in transaction #3202

apurvam · 2017-06-02T00:20:54Z

Due to the async nature of the producer, it is possible to attempt to drain a messages whose partition hasn't been added to the transaction yet. Before this patch, we considered this a fatal error. However, it is only in error if the partition isn't in the queue to be sent to the coordinator.

This patch updates the logic so that we only fail the producer if the partition would never be added to the transaction. If the partition of the batch is yet to be added, we will simply wait for the partition to be added to the transaction before sending the batch to the broker.

apurvam · 2017-06-02T00:21:01Z

cc @hachikuji @guozhangwang

asfbot · 2017-06-02T01:21:58Z

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

asfbot · 2017-06-02T01:37:24Z

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

hachikuji · 2017-06-02T01:45:40Z

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

@@ -428,6 +433,12 @@ synchronized boolean isReady() {
        return isTransactional() && currentState == State.READY;
    }

+    private synchronized boolean wouldTransactionContainPartition(TopicPartition tp) {


nit: this name is a bit awkward. Could we split this into two checks isPartitionAdded and isPartitionPending or something like that?

guozhangwang

Nit comment, otherwise LGTM.

guozhangwang · 2017-06-02T01:45:34Z

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

@@ -428,6 +433,12 @@ synchronized boolean isReady() {
        return isTransactional() && currentState == State.READY;
    }

+    private synchronized boolean wouldTransactionContainPartition(TopicPartition tp) {


nit: rename to currentTransactionContainsPartition?

…ons which would be added to the transaction.

apurvam · 2017-06-02T04:37:43Z

@guozhangwang @hachikuji addressed comments on names. Went with Jason's suggestion to split the function into two -- we actually had one half of it anyway.

hachikuji · 2017-06-02T04:55:00Z

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

-            transitionToFatalError(new IllegalStateException("Attempted to dequeue a record batch to send " +
-                    "for partition " + tp + ", which hasn't been added to the transaction yet"));
-            return false;
+        if (isInTransaction() || hasError()) {


One quick question: what is the expected behavior when we are in a fatal error state? It seems we'd want to ensure that this function always returns false?

That's a good point. In general, I guess there is no point retrying produce requests if you are fenced or have lost your permissions. And in a fatal error, we have to close and restart the producer, resetting all the sequence numbers. So we should always return false if the error is fatal.

Will update.

…ucer is in a fatal error or abortable error state

hachikuji

The fix LGTM. Left a couple test suggestions. Hopefully shouldn't be too different from the tests you already have.

hachikuji · 2017-06-02T05:31:42Z

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

+        // We shouldn't drain batches which haven't been added to the transaction yet.
+        assertTrue(drainedBatches.containsKey(node1.id()));
+        assertTrue(drainedBatches.get(node1.id()).isEmpty());
+        assertTrue(transactionManager.hasError());


Guess we could make this a little stronger with a hasFatalError?

hachikuji · 2017-06-02T05:32:17Z

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

@@ -234,10 +234,17 @@ public RuntimeException lastError() {
    }

    public synchronized boolean ensurePartitionAdded(TopicPartition tp) {


Would it be too much trouble to add a couple basic unit tests just for this function?

Added special tests for this is overkill since it depends so much on the transaction manager state. I have instead added assertions about the expected return values of this function in various existing tests, which achieves the same purpose.

hachikuji · 2017-06-02T05:33:52Z

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

+    }
+
+    @Test
+    public void testRaiseErrorWhenNoPartitionsPendingOnDrain() throws InterruptedException {


Perhaps we should also cover the case where the transaction is in an abortable error state?

I added this case now.

asfbot · 2017-06-02T06:01:39Z

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

asfbot · 2017-06-02T06:12:15Z

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

hachikuji

Thanks for the updates. LGTM

asfbot · 2017-06-02T07:21:48Z

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

asfbot · 2017-06-02T07:51:36Z

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

…ansaction Due to the async nature of the producer, it is possible to attempt to drain a messages whose partition hasn't been added to the transaction yet. Before this patch, we considered this a fatal error. However, it is only in error if the partition isn't in the queue to be sent to the coordinator. This patch updates the logic so that we only fail the producer if the partition would never be added to the transaction. If the partition of the batch is yet to be added, we will simply wait for the partition to be added to the transaction before sending the batch to the broker. Author: Apurva Mehta <apurva@confluent.io> Reviewers: Guozhang Wang <wangguoz@gmail.com>, Jason Gustafson <jason@confluent.io> Closes #3202 from apurvam/KAFKA-5364-ensure-partitions-added-to-txn-before-send (cherry picked from commit 673ab67) Signed-off-by: Jason Gustafson <jason@confluent.io>

hachikuji reviewed Jun 2, 2017

View reviewed changes

guozhangwang reviewed Jun 2, 2017

View reviewed changes

Apurva Mehta added 2 commits June 1, 2017 21:30

Fix logic in ensurePartitionAddedToTransaction to account for partiti…

e4a7f9b

…ons which would be added to the transaction.

Address PR Comments

a538915

apurvam force-pushed the KAFKA-5364-ensure-partitions-added-to-txn-before-send branch from 7dd2f7f to a538915 Compare June 2, 2017 04:36

hachikuji reviewed Jun 2, 2017

View reviewed changes

Update the behavior of ensurePartitionAdded based on whether the prod…

e8a6d93

…ucer is in a fatal error or abortable error state

hachikuji reviewed Jun 2, 2017

View reviewed changes

Add test cases

7b1b120

hachikuji approved these changes Jun 2, 2017

View reviewed changes

asfgit closed this in 673ab67 Jun 2, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-5364: Don't fail producer if drained partition is not yet in transaction #3202

KAFKA-5364: Don't fail producer if drained partition is not yet in transaction #3202

apurvam commented Jun 2, 2017

apurvam commented Jun 2, 2017 •

edited

Loading

asfbot commented Jun 2, 2017

asfbot commented Jun 2, 2017

hachikuji Jun 2, 2017

guozhangwang left a comment

guozhangwang Jun 2, 2017

apurvam commented Jun 2, 2017

hachikuji Jun 2, 2017

apurvam Jun 2, 2017

hachikuji left a comment

hachikuji Jun 2, 2017

hachikuji Jun 2, 2017

apurvam Jun 2, 2017

hachikuji Jun 2, 2017

apurvam Jun 2, 2017

asfbot commented Jun 2, 2017

asfbot commented Jun 2, 2017

hachikuji left a comment

asfbot commented Jun 2, 2017

asfbot commented Jun 2, 2017

		@@ -234,10 +234,17 @@ public RuntimeException lastError() {
		}

		public synchronized boolean ensurePartitionAdded(TopicPartition tp) {

KAFKA-5364: Don't fail producer if drained partition is not yet in transaction #3202

KAFKA-5364: Don't fail producer if drained partition is not yet in transaction #3202

Conversation

apurvam commented Jun 2, 2017

apurvam commented Jun 2, 2017 • edited Loading

asfbot commented Jun 2, 2017

asfbot commented Jun 2, 2017

Choose a reason for hiding this comment

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apurvam commented Jun 2, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji 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 Jun 2, 2017

asfbot commented Jun 2, 2017

hachikuji left a comment

Choose a reason for hiding this comment

asfbot commented Jun 2, 2017

asfbot commented Jun 2, 2017

apurvam commented Jun 2, 2017 •

edited

Loading