KAFKA-14690: Add topic IDs to the OffsetCommit API version 9 by Hangleton · Pull Request #13240 · apache/kafka

Hangleton · 2023-02-13T16:30:33Z

This change introduces topic ids in OffsetCommit requests and responses. The approach chosen in the PR is to support references to topics by either their ID or name (but not both at the same time).

Topic IDs are not surfaced to the Kafka consumer APIs.

On the consumer-side, topic IDs and names are resolved using the client-side cached metadata. On brokers, the metadata cache is used. If a topic ID cannot be resolved on the broker, the corresponding topic will be assigned an UNKNOWN_TOPIC_ID in the OffsetCommit response.

This PR is written as part of KIP-848: The Next Generation of the Consumer Rebalance Protocol.

Out of scope

The admin client continues to use OffsetCommit version 8, without topic ids.
Support of topic ids for offset commits in the new component CommitRequestManager will be added in KAFKA-14777.

Request path on the consumer

Request path on the broker

Requests version >= 9 use topic ids exclusively. Brokers resolve the topic name in the request handler based on cached metadata. The request is then propagated to the group coordinator which constructs the response with topic names as of time of this PR. The resolution of topic name back to topic ID is performed when the response from the group coordinator is merged with the partial response for topic with error (if any).

Committer Checklist (excluded from commit message)

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

Hangleton · 2023-02-13T17:01:57Z

Hello David (@dajac), still working on this but opening a draft if you wish to start reviewing at your convenience.

dajac · 2023-02-13T17:50:49Z

@Hangleton Thanks. I will take a look later this week.

Hangleton · 2023-02-14T15:34:01Z

Thanks for the review, David. I am working on adding unit tests for OffsetCommitResponse and the server-side handling of the request/response, and fix the bugs you have identified.

…cNameOrNull; remove <p></p>; nullable -> if/else; revert RequestQuotaTest

…ents in OffsetCommitRequestTest

…dleOffsetCommitRequestTopicsAndPartitionsValidation the same

…est.

…r.OffsetCommitRequestTest.

…torTest.

Hangleton · 2023-05-25T08:00:45Z

Hi David (@dajac), thanks for the review and apologies for the delayed reply. Thanks to Christo's help, I believe most of your comments have been addressed. I have one question regarding the behaviour of the offset commit consumer API that you identified here. Thanks!

Hangleton · 2023-06-01T12:47:18Z

Hello David (@dajac), I was discussing this with Christo today as part of his work on the OffsetFetch API. Would you like this PR on OffsetCommit to be split to make the review easier and reduce risks?

dajac · 2023-06-12T12:41:20Z

+                    if (topicName == null) {
+                        // Could only happen if the broker replied with an ID which was not in the request and
+                        // unknown by this client. This would be a bug.
+                        log.warn("Ignoring invalid topic ID found in OffsetCommit response: " + topic.topicId());
+                        continue;
+                    }


As a second thought, I wonder if we should complete the future with an exception here. Being defensive would help us to catch bugs early one. What do you think?

dajac

@Hangleton I am back one this one. I looked at the files in clients package and left a few minor comments.

dajac · 2023-06-12T12:43:20Z

 *   - {@link Errors#INVALID_COMMIT_OFFSET_SIZE}
 *   - {@link Errors#TOPIC_AUTHORIZATION_FAILED}
 *   - {@link Errors#GROUP_AUTHORIZATION_FAILED}
+ *   - {@link Errors#STALE_MEMBER_EPOCH}


Should we remove this one for now as it is not implemented yet?

dajac · 2023-06-12T12:44:42Z

+  // Version 9 adds TopicId field and can return STALE_MEMBER_EPOCH, UNKNOWN_MEMBER_ID
+  // and UNKNOWN_TOPIC_ID errors (KIP-848).


Should we remove STALE_MEMBER_EPOCH and UNKNOWN_MEMBER_ID for now?

dajac · 2023-06-12T13:00:26Z

+            assertFalse(coordinator.commitOffsetsSync(offsets, time.timer(timeoutMs)));
+        } else {
+            AtomicBoolean callbackInvoked = new AtomicBoolean();
+            coordinator.commitOffsetsAsync(offsets, (inputOffsets, exception) -> {


My understanding is that we don't retry when commitOffsetsAsync is used. Is it correct? If it is, it may be better to split the test in two. It is really misleading otherwise.

dajac · 2023-06-12T13:01:50Z

+        }
+    }
+
+    private Map<TopicPartition, OffsetAndMetadata> testRetryCommitWithUnknownTopicIdSetup() {


nit: It may be better to name this one prepare.....

dajac · 2023-06-12T13:01:56Z

+        }
+    }
+
+    private Map<TopicPartition, OffsetAndMetadata> testRetryCommitWithUnknownTopicIdSetup() {


nit: It may be better to name this one prepare.....

dajac · 2023-06-12T13:05:09Z

+        Map<TopicIdPartition, Long> byTopicNameOffsets = new HashMap<>();
+        byTopicNameOffsets.put(ti1p, 100L);
+        byTopicNameOffsets.put(ti2p, 200L);
+        byTopicNameOffsets.put(unknownTopicIdPartition, 300L);


Is this used anywhere?

dajac · 2023-06-12T13:12:15Z

+        // The following offset commit response defines a topic incorrectly. The coordinator ignores the topic,
+        // and the group authorization failure is therefore not propagated.
+        client.prepareResponse(offsetCommitResponse(topicName, Uuid.ZERO_UUID, Errors.GROUP_AUTHORIZATION_FAILED));
+        assertTrue(coordinator.commitOffsetsSync(offsets, time.timer(Long.MAX_VALUE)));


I think that the method should return false if any mismatched topic id. If I commit foo-topic-id and bar-topic-id, the method should not succeed if we don't get a response for any of them, right?

dajac · 2023-06-12T13:13:25Z

+        // The following offset commit responses defines a topic incorrectly. The coordinator ignores the topic,
+        // and the group authorization failure is therefore not propagated.
+        asserter.accept(
+            offsetCommitResponse(topicName, Uuid.ZERO_UUID, Errors.GROUP_AUTHORIZATION_FAILED),
+            null);


This case is not correct as well in my opinion. The caller should get an exception in this case.

dajac · 2023-06-12T13:32:08Z

+        return topic;
    }

+    public static final class NameAndId {


nit: TopicNameAndId?

It is a bit weird to have this class defined here but I cannot think of a better place for now. Thoughts?

dajac · 2023-06-12T13:33:43Z

-        ).build(version);
+                                )))
+            ),
+                true


nit: this seems to be misaligned.

dajac · 2023-09-29T21:32:07Z

Closing this PR for now as the topic id work will be done later. We can re-open it when we resume the work.

Hangleton commented Feb 13, 2023

View reviewed changes

Hangleton force-pushed the offset-commit-api-version-9 branch 2 times, most recently from d09095b to 568cbe5 Compare February 13, 2023 17:00

dajac self-requested a review February 13, 2023 17:50

dajac added the KIP-848 The Next Generation of the Consumer Rebalance Protocol label Feb 13, 2023

Hangleton force-pushed the offset-commit-api-version-9 branch 6 times, most recently from 75b997e to 3b634e9 Compare February 14, 2023 13:35

dajac reviewed Feb 14, 2023

View reviewed changes

Comment thread core/src/main/scala/kafka/server/KafkaApis.scala Outdated

dajac reviewed Feb 14, 2023

View reviewed changes

Comment thread core/src/main/scala/kafka/server/KafkaApis.scala Outdated

dajac reviewed Feb 14, 2023

View reviewed changes

Comment thread core/src/main/scala/kafka/server/KafkaApis.scala Outdated

Hangleton force-pushed the offset-commit-api-version-9 branch 13 times, most recently from 00b0e10 to f4f7dcd Compare February 20, 2023 15:00

clolov added 10 commits May 24, 2023 10:45

Remove BiMap; getTopicIdOrZero -> topicIdOrZero; getTopicName -> topi…

642f063

…cNameOrNull; remove <p></p>; nullable -> if/else; revert RequestQuotaTest

Revert builder test changes

1d26721

Checkstyle changes

5811933

Revert refactoring changes in KafkaApisTest; address some of the comm…

3120b9c

…ents in OffsetCommitRequestTest

Split unit test into two

8ebd8e8

Simplify partition method

d5785a4

Parameterise test

20f57e3

Fix testHandleOffsetCommitRequestTopicsAndPartitionsValidation

da744e9

Fix testHandleOffsetCommitRequest and formatting

72f7dd6

Make offsetCommitRequest in testHandleOffsetCommitRequest and testHan…

892dd25

…dleOffsetCommitRequestTopicsAndPartitionsValidation the same

Hangleton force-pushed the offset-commit-api-version-9 branch from d2a813a to 892dd25 Compare May 24, 2023 09:46

Hangleton added 4 commits May 24, 2023 11:56

Fix indent and unused import.

bd4d35c

Consolidate definition of topic-partition in ConsumerCoordinatorTest.

254a6ec

Split unit test on consumer coordinator to provide higher cohesion.

c3639ec

Remove unncesseary tearDown() override in server.OffsetCommitRequestT…

0b0e649

…est.

Hangleton force-pushed the offset-commit-api-version-9 branch from 1496b9e to 0b0e649 Compare May 24, 2023 13:35

Hangleton added 6 commits May 24, 2023 14:52

Use parameterized test to expose OffsetCommitRequest version in serve…

d1de0ae

…r.OffsetCommitRequestTest.

Simplify server.OffsetCommitRequestTest by using DTO directly.

cf91d4e

Remove redundant test of alterConsumerGroupOffsets.

3a4e22d

Indentation.

7abdece

Remove request captor for the OffsetCommitRequest in ConsumerCoordina…

9207f5a

…torTest.

Identation.

ffbc262

Hangleton requested a review from dajac May 25, 2023 07:58

clolov mentioned this pull request May 30, 2023

KAFKA-14691; [1/N] Add new fields to OffsetFetchRequest and OffsetFetchResponse - WIP #13764

Closed

dajac reviewed Jun 12, 2023

View reviewed changes

dajac added the stale Stale PRs label Sep 5, 2023

		// Version 9 adds TopicId field and can return STALE_MEMBER_EPOCH, UNKNOWN_MEMBER_ID
		// and UNKNOWN_TOPIC_ID errors (KIP-848).

Conversation

Hangleton commented Feb 13, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Request path on the consumer

Request path on the broker

Committer Checklist (excluded from commit message)

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Hangleton commented Feb 13, 2023

Uh oh!

dajac commented Feb 13, 2023

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Hangleton commented Feb 14, 2023

Uh oh!

Hangleton commented May 25, 2023

Uh oh!

Hangleton commented Jun 1, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac commented Sep 29, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Hangleton commented Feb 13, 2023 •

edited

Loading