KAFKA-14950: implement assign() and assignment() #13797

philipnee · 2023-06-01T17:15:36Z

In this PR: I implemented assign() and assignment(). Ported the original tests from the KafkaConsumerTest.java

Different from the original Implementation:

We will explicitly send an assignment change event to the background thread to invoke auto-commit if the group.id is configured. After updating the subscription state, a NewTopicsMetadataUpdateRequestEvent will also be sent to the background thread to update the metadata.

unsubscribe add tests Update DefaultBackgroundThreadTest.java Bad test...

junrao

@philipnee : Thanks for the PR. Left a few comments.

.../java/org/apache/kafka/clients/consumer/internals/events/MetadataUpdateApplicationEvent.java

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

junrao · 2023-06-07T16:50:05Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

+        /*
+                this.coordinator.onLeavePrepare();
+                this.coordinator.maybeLeaveGroup("the consumer unsubscribed from all topics");
+         */


Should we uncomment the code?

Removed those lines as it's for unsubscribe which isn't baked enough for this PR.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

junrao · 2023-06-07T16:56:18Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

+
+        log.info("Assigned to partition(s): {}", Utils.join(partitions, ", "));
+        if (this.subscriptions.assignFromUser(new HashSet<>(partitions)))
+           updateMetadata(time.milliseconds());


The existing consumer seems to only update the metadata for new topics. updateMetadata doesn't make it clear that it's for new topics. Could we make it clear?

Does the latest rename and refactor address your concerns?

junrao · 2023-06-07T16:56:37Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

@@ -537,7 +568,9 @@ public void subscribe(Pattern pattern) {

    @Override
    public void unsubscribe() {
-        throw new KafkaException("method not implemented");
+        // fetcher.clearBufferedDataForUnassignedPartitions(Collections.emptySet());


Is this still needed?

Backed out the unsubscribe logic as it wasn't implemented in any meaningful way.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

junrao · 2023-06-07T20:04:51Z

...s/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java

@@ -166,7 +208,7 @@ private DefaultBackgroundThread mockBackgroundThread() {
                applicationEventsQueue,
                backgroundEventsQueue,
                this.errorEventHandler,
-                processor,
+            applicationEventProcessor,


indentation

Cleanup

Clean up to minimize diffs

philipnee · 2023-07-06T15:45:19Z

@kirktrue thanks for taking over this. @junrao - would you be up for another round of review?

junrao · 2023-07-07T22:44:28Z

@philipnee Was the following comment addressed? Kirk mentioned to take a closer look.

#13797 (comment)

junrao

@philipnee : Thanks for the updated PR. A couple of more comments.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

philipnee · 2023-07-19T18:18:14Z

@junrao - thanks for the review. I've addressed the two issues you pointed out. Would you be able to take another look at it?

junrao

@philipnee : Thanks for the updated PR. The PR LGTM. Are the 23 test failures related?

philipnee · 2023-07-20T22:50:43Z

Thanks, @junrao -The failing tests should be fixed in the latest commit: The integration test failed because of missing subscription state dependency, which was added in one of the subsequent PR.

The failing tests are irrelevant, and these are:

Build / JDK 8 and Scala 2.12 / testOffsetTranslationBehindReplicationFlow() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsWithCustomForwardingAdminIntegrationTest
44s
Build / JDK 11 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
12s
Build / JDK 20 and Scala 2.13 / [1] tlsProtocol=TLSv1.2, useInlinePem=false – org.apache.kafka.common.network.SslTransportLayerTest
14s
Build / JDK 20 and Scala 2.13 / testSyncTopicConfigs() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsWithCustomForwardingAdminIntegrationTest
1m 56s

junrao

@philipnee : Thanks for the updated PR. LGTM

junrao · 2023-07-21T18:41:16Z

@philipnee : Could you update the description of the PR (to be included in the commit message) before I merge it? The description still has the following and I am not sure if it's still relevant.

Missing:

Fetcher to clear the buffer

philipnee · 2023-07-21T19:39:03Z

Hey @junrao - Thanks again. I updated the description to reflect the updated code more accurately. Here I just quoted the paragraph I made the change to:

We will explicitly send an assignment change event to the background thread to invoke auto-commit if the group.id is configured. After updating the subscription state, a NewTopicsMetadataUpdateRequestEvent will also be sent to the background thread to update the metadata.

We will explicitly send an assignment change event to the background thread to invoke auto-commit if the group.id is configured. After updating the subscription state, a NewTopicsMetadataUpdateRequestEvent will also be sent to the background thread to update the metadata. Co-authored-by: Kirk True <kirk@kirktrue.pro> Reviewers: Jun Rao <junrao@gmail.com>

…resolve some conflicts (#32) * cherry-pick * KAFKA-14950: implement assign() and assignment() (apache#13797) We will explicitly send an assignment change event to the background thread to invoke auto-commit if the group.id is configured. After updating the subscription state, a NewTopicsMetadataUpdateRequestEvent will also be sent to the background thread to update the metadata. Co-authored-by: Kirk True <kirk@kirktrue.pro> Reviewers: Jun Rao <junrao@gmail.com> * cherry-pick conflict * Update CommitRequestManager.java Update ApplicationEventProcessor.java --------- Co-authored-by: Kirk True <kirk@kirktrue.pro>

We will explicitly send an assignment change event to the background thread to invoke auto-commit if the group.id is configured. After updating the subscription state, a NewTopicsMetadataUpdateRequestEvent will also be sent to the background thread to update the metadata. Co-authored-by: Kirk True <kirk@kirktrue.pro> Reviewers: Jun Rao <junrao@gmail.com>

KAFKA-14950: implement assign() and assignment()

2799139

unsubscribe add tests Update DefaultBackgroundThreadTest.java Bad test...

philipnee mentioned this pull request Jun 1, 2023

KAFKA-14950: implement assign() and assignment() #13605

Closed

philipnee added KIP-848 ctr labels Jun 1, 2023

Update DefaultBackgroundThreadTest.java

728e45e

junrao reviewed Jun 7, 2023

View reviewed changes

kirktrue added 5 commits June 8, 2023 12:17

Updates based on PR feedback

cf3cb45

Update DefaultBackgroundThreadTest.java

5d54fd7

Cleanup

Update DefaultBackgroundThreadTest.java

c36f4ce

Clean up to minimize diffs

Removed as-yet unused UNSUBSCRIBE event type

2d99782

Merge branch 'apache:trunk' into consumer-refactor-assignment

d0b1205

philipnee added 4 commits July 14, 2023 10:45

Ensure autocommit is probbed when invoking assignment change

4ea15e9

Trigger autocommit without reseting timer

1ef62b5

updating auto commit timer

632b80d

Testing assignment change event on the background thread

d524453

junrao reviewed Jul 17, 2023

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java Outdated Show resolved Hide resolved

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java Show resolved Hide resolved

Updated pr comment.

2eac051

junrao reviewed Jul 20, 2023

View reviewed changes

Adding subscription state from the previous commit

bc7024a

junrao approved these changes Jul 21, 2023

View reviewed changes

junrao merged commit 1656591 into apache:trunk Jul 21, 2023
1 check was pending

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-14950: implement assign() and assignment() #13797

KAFKA-14950: implement assign() and assignment() #13797

philipnee commented Jun 1, 2023 •

edited

junrao left a comment

junrao Jun 7, 2023

kirktrue Jun 8, 2023

junrao Jun 7, 2023

kirktrue Jun 8, 2023

junrao Jun 7, 2023

kirktrue Jun 8, 2023

junrao Jun 7, 2023

kirktrue Jun 8, 2023

philipnee commented Jul 6, 2023

junrao commented Jul 7, 2023

junrao left a comment

philipnee commented Jul 19, 2023

junrao left a comment

philipnee commented Jul 20, 2023

junrao left a comment

junrao commented Jul 21, 2023

philipnee commented Jul 21, 2023 •

edited

KAFKA-14950: implement assign() and assignment() #13797

KAFKA-14950: implement assign() and assignment() #13797

Conversation

philipnee commented Jun 1, 2023 • edited

junrao 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

philipnee commented Jul 6, 2023

junrao commented Jul 7, 2023

junrao left a comment

Choose a reason for hiding this comment

philipnee commented Jul 19, 2023

junrao left a comment

Choose a reason for hiding this comment

philipnee commented Jul 20, 2023

junrao left a comment

Choose a reason for hiding this comment

junrao commented Jul 21, 2023

philipnee commented Jul 21, 2023 • edited

philipnee commented Jun 1, 2023 •

edited

philipnee commented Jul 21, 2023 •

edited