KAFKA-7961: Ignore assignment for un-subscribed partitions by jsancio · Pull Request #6304 · apache/kafka

jsancio · 2019-02-22T00:46:10Z

Whenever the consumer coordinator sends a response that doesn't match the client consumer subscription, ignore the assignment and rejoin the group.

Testing strategy: create a mocked client that first sends an assignment response that doesn't match the client subscription followed by an assignment response that does match the client subscription.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

Added "testInvalidCoordinatorAssignment" test wich result in the following failure: ``` org.apache.kafka.clients.consumer.internals.ConsumerCoordinatorTest > testInvalidCoordinatorAssignment FAILED java.lang.IllegalArgumentException: Assigned partition test2-0 for non-subscribed topic; subscription is [test1] at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignFromSubscribed(SubscriptionState.java:195) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:249) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:410) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:344) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:340) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinatorTest.testInvalidCoordinatorAssignment(ConsumerCoordinatorTest.java:462) 1 test completed, 1 failed ```

jsancio · 2019-02-22T00:49:48Z

cc @guozhangwang @ijuma @hachikuji

hachikuji

Thanks for picking this up. Looks good overall. Left a few comments.

Throw an IllegalStateException if the leader sent and assignment that doesn't match for the subscription requested. Modify testInvalidCoordinatorAssignment and add testOutdatedCoordinatorAssignmetn tests for cover this condition.

jsancio

@hachikuji Here are the failures when we run these tests against the old code.

jsancio · 2019-02-22T23:30:59Z

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


For this test, the old code fails as follow:

org.apache.kafka.clients.consumer.internals.ConsumerCoordinatorTest > testInvalidCoordinatorAssignment FAILED java.lang.Exception: Unexpected exception, expected<java.lang.IllegalStateException> but was<java.lang.IllegalArgumentException> Caused by: java.lang.IllegalArgumentException: Assigned partition test2-0 for non-subscribed topic; subscription is [test1] at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignFromSubscribed(SubscriptionState.java:195) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:249) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:410) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:344) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:340) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinatorTest.testInvalidCoordinatorAssignment(ConsumerCoordinatorTest.java:506)

jsancio · 2019-02-22T23:31:38Z

    }

+    @Test
+    public void testOutdatedCoordinatorAssignment() {


For this test the old code fails as follow:

org.apache.kafka.clients.consumer.internals.ConsumerCoordinatorTest > testOutdatedCoordinatorAssignment FAILED java.lang.IllegalArgumentException: Assigned partition test2-0 for non-subscribed topic; subscription is [test1] at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignFromSubscribed(SubscriptionState.java:195) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:249) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:410) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:344) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:340) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinatorTest.testOutdatedCoordinatorAssignment(ConsumerCoordinatorTest.java:466)

hachikuji

Thanks, the fix looks good. Just had a couple small comments.

hachikuji

LGTM. Thanks for the patch!

ijuma · 2019-02-23T22:45:58Z

Should this be cherry-picked to 2.2?

jsancio · 2019-02-25T17:59:27Z

@ijuma I don't think so. This was found during internal testing by waking up the consumer and while also changing the subscription.

@hachikuji What do you think?

* AK/trunk: (36 commits) KAFKA-7962: Avoid NPE for StickyAssignor (apache#6308) Address flakiness of CustomQuotaCallbackTest#testCustomQuotaCallback (apache#6330) KAFKA-7918: Inline generic parameters Pt. II: RocksDB Bytes Store and Memory LRU Caches (apache#6327) MINOR: fix parameter naming (apache#6316) KAFKA-7956 In ShutdownableThread, immediately complete the shutdown if the thread has not been started (apache#6218) MINOR: Refactor replica log dir fetching for improved logging (apache#6313) [TRIVIAL] Remove unused StreamsGraphNode#repartitionRequired (apache#6227) MINOR: Increase produce timeout to 120 seconds (apache#6326) KAFKA-7918: Inline generic parameters Pt. I: in-memory key-value store (apache#6293) MINOR: Fix line break issue in upgrade notes (apache#6320) KAFKA-7972: Use automatic RPC generation in SaslHandshake MINOR: Enable capture of full stack trace in StreamTask#process (apache#6310) KAFKA-7938: Fix test flakiness in DeleteConsumerGroupsTest (apache#6312) KAFKA-7937: Fix Flaky Test ResetConsumerGroupOffsetTest.testResetOffsetsNotExistingGroup (apache#6311) MINOR: Update docs to say 2.2 (apache#6315) KAFKA-7672 : force write checkpoint during StreamTask #suspend (apache#6115) KAFKA-7961; Ignore assignment for un-subscribed partitions (apache#6304) KAFKA-7672: Restoring tasks need to be closed upon task suspension (apache#6113) KAFKA-7864; validate partitions are 0-based (apache#6246) KAFKA-7492 : Updated javadocs for aggregate and reduce methods returning null behavior. (apache#6285) ...

Whenever the consumer coordinator sends a response that doesn't match the client consumer subscription, we should check the subscription to see if it has changed. If it has, we can ignore the assignment and request a rebalance. Otherwise, we can throw an exception as before. Testing strategy: create a mocked client that first sends an assignment response that doesn't match the client subscription followed by an assignment response that does match the client subscription. Reviewers: Jason Gustafson <jason@confluent.io>

jsancio added 2 commits February 21, 2019 16:34

KAFKA-7961: Ignore assignment for un-subscribed partitions

a9e38fa

jsancio commented Feb 22, 2019

View reviewed changes

Comment thread clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java Outdated

hachikuji self-assigned this Feb 22, 2019

hachikuji reviewed Feb 22, 2019

View reviewed changes

KAFKA-7961: Add LogContext to SubscriptionState

234f77d

hachikuji reviewed Feb 22, 2019

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java Outdated

Comment thread clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java Outdated

KAFKA-7961: Handle invalid response from leader

bf7b824

Throw an IllegalStateException if the leader sent and assignment that doesn't match for the subscription requested. Modify testInvalidCoordinatorAssignment and add testOutdatedCoordinatorAssignmetn tests for cover this condition.

jsancio commented Feb 22, 2019

View reviewed changes

KAFKA-7961: Fix logging to use "{}"

d31fc20

hachikuji reviewed Feb 22, 2019

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java Outdated

Comment thread clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java Outdated

KAFKA-7961: Use assertThrows

71eca2a

hachikuji approved these changes Feb 23, 2019

View reviewed changes

hachikuji merged commit eb8cc09 into apache:trunk Feb 23, 2019

jsancio deleted the kafka-7961 branch February 25, 2019 17:59

Conversation

jsancio commented Feb 22, 2019

Committer Checklist (excluded from commit message)

Uh oh!

jsancio commented Feb 22, 2019

Uh oh!

Uh oh!

hachikuji left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

jsancio left a comment

Choose a reason for hiding this comment

Uh oh!

jsancio Feb 22, 2019

Choose a reason for hiding this comment

Uh oh!

jsancio Feb 22, 2019

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

ijuma commented Feb 23, 2019

Uh oh!

jsancio commented Feb 25, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

hachikuji left a comment •

edited

Loading