KAFKA-16194: Do not return records from poll if group metadata unknown #15369

cadonna · 2024-02-14T10:32:44Z

Due to the asynchronous nature of the async consumer, it might happen that on the application thread the group metadata is not known after the first poll returns records. If the offsets of those records are then send to a transaction with

txnProducer.sendOffsetsToTransaction(offsetsToCommit, groupMetadata);

and then the transaction is committed, the group coordinator will raise an error saying that the member is not known since the member in groupMetadata is still -1 before the metadata is updated.

This commit avoids this error by not returning any records from poll() until the group metadata is updated, i.e., the member ID and the generation ID (a.k.a. member epoch) are known. This check is only done if group management is used.

Additionally, this commit resets the group metadata when the consumer unsubscribes.

Committer Checklist (excluded from commit message)

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

lucasbru

Thanks for the PR! I left some comments.
(I did not check the tests in detail, because I'm not sure about the current main implementation)

lucasbru · 2024-02-20T13:18:46Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

@@ -98,6 +99,8 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;


lucasbru · 2024-02-20T13:25:15Z

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

@@ -727,6 +732,17 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
        }
    }

+    private boolean isGenerationKnown() {


How about this one-liner

return groupMetadata.filter(g -> g.generationId() != JoinGroupRequest.UNKNOWN_GENERATION_ID).isPresent();

I am not sure this is correct, because I can use user assigned partitions but still use a group ID to manage the offsets broker-side, if I remember correctly.

Probably, I can do something like

if (subscriptions.hasAutoAssignedPartitions()) { return groupMetadata.filter(g -> g.generationId() != JoinGroupRequest.UNKNOWN_GENERATION_ID).isPresent(); } return true;

Makes sense. But then I'm a bit confused by the method name, because it just returns true when there is no generation...

lucasbru · 2024-02-20T13:26:29Z

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

@@ -1467,6 +1483,7 @@ public void unsubscribe() {
                } catch (TimeoutException e) {
                    log.error("Failed while waiting for the unsubscribe event to complete");
                }
+                groupMetadata = initializeGroupMetadata(groupMetadata.get().groupId(), Optional.empty());


This looks like an independent fix? Maybe mention in the PR description

lucasbru · 2024-02-20T13:27:58Z

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

-                if (!fetch.isEmpty()) {
-                    if (fetch.records().isEmpty()) {
-                        log.trace("Returning empty records from `poll()` "
+                if (isGenerationKnown()) {


If this check is false... where do we block? It seems like we busy loop with 100% CPU, but maybe I'm wrong

The busy loop should not be too bad since the group metadata update event is added to the queue when also the new assignment is set in the membership manager.

Moreover, this is kind of a temporary solution until https://issues.apache.org/jira/browse/KAFKA-16285 is done.

lucasbru

If you want, you could improve the name of isGenerationKnown, but LGTM.

jolshan · 2024-02-22T22:05:33Z

Hmm after this was merged, I'm seeing failed builds on every master build.

[2024-02-22T21:19:13.914Z] Execution failed for task ':streams:test'.
[2024-02-22T21:19:13.914Z] > The following test methods could not be retried, which is unexpected. Please file a bug report at https://github.com/gradle/test-retry-gradle-plugin/issues
[2024-02-22T21:19:13.914Z]      org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStoreTest#shouldLogAndMeasureExpiredRecords[org.apache.kafka.streams.state.internals.WindowKeySchema@71154f21]
[2024-02-22T21:19:13.914Z]      org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStoreTest#shouldLogAndMeasureExpiredRecords[org.apache.kafka.streams.state.internals.SessionKeySchema@5b69d40d]
[2024-02-22T21:19:13.914Z]      org.apache.kafka.streams.state.internals.RocksDBTimestampedSegmentedBytesStoreTest#shouldLogAndMeasureExpiredRecords[org.apache.kafka.streams.state.internals.WindowKeySchema@41c07648]
[2024-02-22T21:19:13.914Z]      org.apache.kafka.streams.state.internals.RocksDBTimestampedSegmentedBytesStoreTest#shouldLogAndMeasureExpiredRecords[org.apache.kafka.streams.state.internals.SessionKeySchema@52d10fb8]

I see this build was fine so not sure if there was some other issue or if the issue is external.

jolshan · 2024-02-22T22:32:05Z

Looks like it was actually this one: #15396. I will follow up there.

apache#15369) Due to the asynchronous nature of the async consumer, it might happen that on the application thread the group metadata is not known after the first poll returns records. If the offsets of those records are then send to a transaction with txnProducer.sendOffsetsToTransaction(offsetsToCommit, groupMetadata); and then the transaction is committed, the group coordinator will raise an error saying that the member is not known since the member in groupMetadata is still -1 before the metadata is updated. This commit avoids this error by not returning any records from poll() until the group metadata is updated, i.e., the member ID and the generation ID (a.k.a. member epoch) are known. This check is only done if group management is used. Additionally, this commit resets the group metadata when the consumer unsubscribes. Reviewer: Lucas Brutschy <lbrutschy@confluent.io>

cadonna force-pushed the AK16194-fix_groupmetadata branch from a064c7e to bc0f1d3 Compare February 19, 2024 09:36

cadonna marked this pull request as ready for review February 19, 2024 09:46

cadonna requested a review from lucasbru February 19, 2024 09:46

cadonna changed the title ~~WIP~~ KAFKA-16194: Do not return records from poll if group metadata unknown Feb 19, 2024

cadonna requested a review from dajac February 19, 2024 10:12

cadonna force-pushed the AK16194-fix_groupmetadata branch from abbcde2 to 1699673 Compare February 20, 2024 09:00

lucasbru reviewed Feb 20, 2024

View reviewed changes

cadonna force-pushed the AK16194-fix_groupmetadata branch from fc64170 to 9e9f895 Compare February 21, 2024 08:18

lucasbru approved these changes Feb 21, 2024

View reviewed changes

cadonna added 8 commits February 21, 2024 11:55

WIP

680ac6a

First solution

a80c3f8

Tidy up code

823276a

Use Runnable instead of Supplier

9a5e912

Fix checkstyle issues

c1e2804

Including feedback

6264b29

Rename verification method

39158bc

Fix unit test

bd4ecee

cadonna force-pushed the AK16194-fix_groupmetadata branch from b3f5edb to bd4ecee Compare February 21, 2024 11:02

Fix unit test differently

b9f16b7

cadonna merged commit 02ebfc6 into apache:trunk Feb 21, 2024
1 check failed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-16194: Do not return records from poll if group metadata unknown #15369

KAFKA-16194: Do not return records from poll if group metadata unknown #15369

cadonna commented Feb 14, 2024 •

edited

lucasbru left a comment

lucasbru Feb 20, 2024

lucasbru Feb 20, 2024

cadonna Feb 20, 2024

cadonna Feb 20, 2024

lucasbru Feb 21, 2024

lucasbru Feb 20, 2024

cadonna Feb 20, 2024

lucasbru Feb 20, 2024

cadonna Feb 20, 2024 •

edited

lucasbru left a comment

jolshan commented Feb 22, 2024

jolshan commented Feb 22, 2024

KAFKA-16194: Do not return records from poll if group metadata unknown #15369

KAFKA-16194: Do not return records from poll if group metadata unknown #15369

Conversation

cadonna commented Feb 14, 2024 • edited

Committer Checklist (excluded from commit message)

lucasbru 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

cadonna Feb 20, 2024 • edited

Choose a reason for hiding this comment

lucasbru left a comment

Choose a reason for hiding this comment

jolshan commented Feb 22, 2024

jolshan commented Feb 22, 2024

cadonna commented Feb 14, 2024 •

edited

cadonna Feb 20, 2024 •

edited