KAFKA-5364: ensurePartitionAdded does not handle pending partitions in abortable error state #3231

hachikuji · 2017-06-04T01:30:47Z

No description provided.

…n abortable error state

hachikuji · 2017-06-04T01:36:22Z

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

@@ -278,7 +282,7 @@ public synchronized boolean isAborting() {
    }

    synchronized boolean isInTransaction() {
-        return currentState == State.IN_TRANSACTION || isCompletingTransaction();
+        return currentState == State.IN_TRANSACTION || isCompletingTransaction() || hasAbortableError();


This bug is a bit insidious. We use isInTransaction when dequeuing records from the RecordAccumulator to determine if the batch should be transactional. So if we continued trying to send after an abortable error, we could lose the transactional flag.

Do we need a comment somewhere?

Will add one.

asfbot · 2017-06-04T02:31:29Z

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

asfbot · 2017-06-04T02:58:29Z

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

hachikuji · 2017-06-04T18:35:07Z

cc @apurvam

@ijuma Note that this only fixes part of the problem from KAFKA-5364. I'd recommend merging it for the fix and improved test coverage, but we should leave the JIRA open.

hachikuji · 2017-06-04T20:49:42Z

@apurvam I noticed one other race condition. When we abort a transaction, we clear the partitions in newPartitionsInTransaction and we abort unsent produce requests. The problem is that the latter happens asynchronously, so we could very well see a batch dequeued for a partition which we had already cleared. I've therefore removed the fatal transition in ensurePartitionAdded in the latest patch.

asfbot · 2017-06-04T21:34:50Z

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

asfbot · 2017-06-04T22:37:52Z

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

asfbot · 2017-06-04T23:07:37Z

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

asfbot · 2017-06-05T00:26:23Z

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

asfbot · 2017-06-05T00:58:57Z

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

apurvam

Thanks for the patch. The cleanup in code and logic is good. The tests are great. LGTM barring some small comments.

apurvam · 2017-06-05T19:25:00Z

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

+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void testMaybeAddPartitionTransactionBeforeBeginTransaction() {


Here and elsewhere in these tests:
nit: typo, missing 'To' in testMaybeAddPartitionToTransactionBeforeBeginTransaction

apurvam · 2017-06-05T19:28:37Z

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

+
+        transactionManager.beginTransaction();
+        transactionManager.maybeAddPartitionToTransaction(tp0);
+        sender.run(time.milliseconds());


The sender.run here should be dropped since the intent of this test is to ensure that the send to partition should fail when the partition is pending add.

This is testing the case where the AddPartitionsToTxn request is in-flight.. I will add a comment to clarify.

apurvam · 2017-06-05T19:30:04Z

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

+
+        transactionManager.beginTransaction();
+        transactionManager.maybeAddPartitionToTransaction(tp0);
+        sender.run(time.milliseconds());


Same as above, this sender.run serves no purpose and is slightly confusing. AT least maybe add a comment to the effect that 'the response to the actual AddPartitionToTxn request is never received even though the sender runs'.

apurvam · 2017-06-05T19:30:56Z

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

@@ -289,6 +295,14 @@ synchronized void transitionToFatalError(RuntimeException exception) {
        transitionTo(State.FATAL_ERROR, exception);
    }

+    synchronized boolean isPartitionAdded(TopicPartition partition) {


Perhaps add a '// visible for testing' comment to this and the next method? They really only should be called from tests.

apurvam

Thanks for the patch, LGTM!

asfbot · 2017-06-06T00:12:49Z

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

asfbot · 2017-06-06T00:44:41Z

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

guozhangwang · 2017-06-06T04:23:16Z

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

-        return currentState == State.IN_TRANSACTION || isCompletingTransaction();
+    synchronized boolean hasOngoingTransaction() {
+        // transactions are considered ongoing once started until completion or a fatal error
+        return currentState == State.IN_TRANSACTION || isCompletingTransaction() || hasAbortableError();


Besides unit test this function is only used in RecordAccumulator#drain, and I'm wondering in that case would we really need to consider hasAbortableError as well?

Actually we can simplify this since, for the time being, transactional producers can only send data inside of a transaction. However, I think we should keep hasAbortableError. We consider the transaction ongoing as long as the client still has the capability to complete it.

guozhangwang

LGTM. Just one question for my own knowledge.

…ransactional flag

asfbot · 2017-06-06T08:54:39Z

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

asfbot · 2017-06-06T09:18:33Z

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

ijuma

Thanks for the PR, great to see the exhaustive tests. I didn't do a full review of the impact of the changes on the TransactionManager since @apurvam and @guozhangwang had already done that.

I left some suggestions and comments. If applicable, they can be addressed in a subsequent PR. Merging this to trunk and 0.11.0.

ijuma · 2017-06-06T11:30:15Z

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

        return lastError;
    }

-    public synchronized boolean ensurePartitionAdded(TopicPartition tp) {
+    public synchronized void failIfUnreadyForSend() {


Nit: why not failIfNotReadyForSend? One character longer, but reads a bit better. :)

Funny. I used that name initially and changed it because I thought the current one read nicer. Now I'm reading it again and I'm inclined to change it back.

ijuma · 2017-06-06T11:33:39Z

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

+        }
+    }
+
+    synchronized boolean sendToPartitionAllowed(TopicPartition tp) {


I think this would be clearer if called isSendToPartitionAllowed. At first I thought this method was sending to a partition. :)

ijuma · 2017-06-06T12:48:14Z

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

+    }
+
+    @Test
+    public void testsendToPartitionAllowedWithAddedPartitionAfterAbortableError() {


Seems like there's a capitalisation error here and in subsequent test names. Will fix it before merging.

Ah, thanks. Thought I fixed this, but I guess I missed a few.

ijuma · 2017-06-06T12:52:13Z

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

@@ -993,7 +1314,6 @@ public void testRaiseErrorWhenNoPartitionsPendingOnDrain() throws InterruptedExc
        // 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.hasFatalError());


Should we be asserting something else about the TransactionManager? Possibly assertFalse(transactionManager.hasFatalError()), but there may be others.

Also, the method name says testRaiseErrorWhen... and there's a comment about result in an error on drain, but I am not seeing any errors. Do they need to be updated?

ijuma · 2017-06-06T12:56:26Z

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

+    }
+
+    @Test
+    public void testsendToPartitionAllowedWithUnaddedPartition() {


Nit: I'd say NotAdded instead of Unadded, one more character but reads better. :)

Hmm.. I think I prefer the current name. Maybe ...WithPartitionNotAdded would be better.

Yeah, WithPartitionNotAdded sounds good. This is definitely a nit, but whenever I see something like unadded or unready, it sounds like it was _un_done instead of _not_done. :)

…n abortable error state Author: Jason Gustafson <jason@confluent.io> Reviewers: Apurva Mehta <apurva@confluent.io>, Guozhang Wang <wangguoz@gmail.com>, Ismael Juma <ismael@juma.me.uk> Closes #3231 from hachikuji/KAFKA-5364

ijuma · 2017-06-06T13:01:58Z

I left the JIRA open as per @hachikuji's earlier comment in the PR. It may be that subsequent commits addressed the remaining issues. In that case, please close the JIRA.

KAFKA-5364: ensurePartitionAdded does not handle pending partitions i…

7e65cf7

…n abortable error state

hachikuji commented Jun 4, 2017

View reviewed changes

Jason Gustafson added 3 commits June 4, 2017 12:38

A few improvements and additional test cases

aa64afd

A couple more fixes

0e2f8f3

Explicitly forbid non-transactional sends when transactionalId is set

d31daa4

Fix failIfUnreadyForSend for idempotent producer

6e61abe

Revert change to exception type in failIfUnreadyForSend

fecf311

apurvam reviewed Jun 5, 2017

View reviewed changes

Address Apurva's comments

1d52324

apurvam approved these changes Jun 5, 2017

View reviewed changes

guozhangwang reviewed Jun 6, 2017

View reviewed changes

Use isTransactional instead of hasOngoingTransaction to set batch isT…

2e45722

…ransactional flag

ijuma approved these changes Jun 6, 2017

View reviewed changes

ijuma reviewed Jun 6, 2017

View reviewed changes

asfgit closed this in 3eed194 Jun 6, 2017

hachikuji pushed a commit to hachikuji/kafka that referenced this pull request Jun 7, 2017

Pile on a few minor improvements from apache#3231 review

83c05ad

KAFKA-5364: ensurePartitionAdded does not handle pending partitions in abortable error state #3231

KAFKA-5364: ensurePartitionAdded does not handle pending partitions in abortable error state #3231

Conversation

hachikuji commented Jun 4, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Jun 4, 2017

asfbot commented Jun 4, 2017

hachikuji commented Jun 4, 2017

hachikuji commented Jun 4, 2017

asfbot commented Jun 4, 2017

asfbot commented Jun 4, 2017

asfbot commented Jun 4, 2017

asfbot commented Jun 5, 2017

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

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

asfbot commented Jun 6, 2017

asfbot commented Jun 6, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang left a comment

Choose a reason for hiding this comment

asfbot commented Jun 6, 2017

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

ijuma commented Jun 6, 2017