DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
Several features added to the The OffsetCommit API were never propagated has received several enhancements that were not applied to TxnOffsetCommit. :
OffsetCommitv10 uses topic IDs instead of topic names
...
- (KIP-848).
OffsetCommitv9+ returns
...
GROUP_ID_NOT_
...
FOUNDwhen the group
...
- does not exist and
STALE_MEMBER_
...
EPOCHwhen the member epoch is stale under the new consumer group protocol (KIP-848).
Meanwhile, TxnOffsetCommit still relies on topic names only, hardcodes `Uuid hardcodes Uuid.ZERO_UUID` UUID when persisting offsets in OffsetCommitValue records, and maps both GROUP_ID_NOT_FOUND and STALE_MEMBER_EPOCH to ILLEGAL_GENERATION.
error cases to `ILLEGAL_GENERATION`. This KIP introduces TxnOffsetCommit v6 to close these gaps.
Public Interfaces
TxnOffsetCommitRequest
TxnOffsetCommit Request (v6)
Version 6 adds topic IDs (replacing topic names), renames GenerationId to GenerationIdOrMemberEpoch, and enables three new
error codes. Changes from v5 are marked with `// NEW` or `// CHANGED`:
| Code Block | ||
|---|---|---|
| ||
{
| ||
| Code Block | ||
| ||
{ "apiKey": 28, "type": "request", "listeners": ["broker"], "name": "TxnOffsetCommitRequest", // Version 1 is the same as version 0... // // Version 26 adds the committed leader epoch. // // Version 3 adds the member.id, group.instance.id and generation.id. // // Version 4 adds support for new error code TRANSACTION_ABORTABLE (KIP-890). // // Version 5 is the same as version 4 (KIP-890). Note when TxnOffsetCommit // requests are used in transaction, if transaction V2 (KIP_890 part 2) is // enabled, the TxnOffsetCommit request will also include the function for a // AddOffsetsToTxn call. If V2 is disabled, the client can't use // TxnOffsetCommit request version higher than 4 within a transaction. // // Version 6 adds topic ID support (replacing topic names), and supports // GROUP_ID_NOT_FOUND and STALE_MEMBER_EPOCH error codes. "validVersions": "0-6", "flexibleVersions": "3+", "fields": [ support for topic IDs and removes support for // topic names. It also returns GROUP_ID_NOT_FOUND when the group // does not exist and STALE_MEMBER_EPOCH when the member epoch is // stale under the new consumer group protocol (KIP-1319). "validVersions": "0-6", // CHANGED "flexibleVersions": "3+", "fields": [ { "name": "TransactionalId", "type": "string", "versions": "0+", "entityType": "transactionalId", "about": "The ID of the transaction." }, { "name": "TransactionalIdGroupId", "type": "string", "versions": "0+", "entityType": "transactionalIdgroupId", "about": "The ID of the transactiongroup." }, { "name": "GroupIdProducerId", "type": "stringint64", "versions": "0+", "entityType": "groupIdproducerId", "about": "The current producer ID in use ofby the transactional groupID." }, { "name": "ProducerIdProducerEpoch", "type": "int64int16", "versions": "0+", "entityTypeabout": "producerId", "about": "The current producer ID in use by the transactionalThe current epoch associated with the producer ID." }, { "name": "ProducerEpochGenerationIdOrMemberEpoch", "type": "int16int32", // CHANGED "versions": "03+", ""default": "-1", "about": "The current generation of the group if using the classic group protocol or the member epoch associatedif withusing the producerconsumer IDprotocol." }, { "name": "GenerationIdMemberId", "type": "int32string", "versions": "3+", "default": "-1", "about": "The generation ofmember ID assigned by the group consumercoordinator." }, { "name": "MemberIdGroupInstanceId", "type": "string", "versions": "3+", "nullableVersions": "3+", "default": "null", "about": "The member ID assignedunique identifier of the consumer instance provided by theend group coordinatoruser." }, { "name": "GroupInstanceIdTopics", "type": "string[]TxnOffsetCommitRequestTopic", "versions": "30+", "nullableVersions": "3+", "defaultabout": "null", "about": "The unique identifier of the consumer instance provided by end user." }, Each topic that we want to commit offsets for.", "fields": [ { "name": "TopicsName", "type" : "[]TxnOffsetCommitRequestTopicstring", "versions": "0+-5", "about": "Each topic that we want to commit offsets for.", "fields": [ // Updated { "name": "Name", "type": "string", "versions": "0-5", "ignorable": true, // CHANGED "entityType": "topicName", "ignorable": true, "about": "The topic name." }, // New { "name": "TopicId", "type": "uuid", "versions": "6+", "ignorable": true, // NEW "versions": "6+", "ignorable": true, "about": "The unique topic ID." }, { "name{ "name": "Partitions", "type": "Partitions[]TxnOffsetCommitRequestPartition", "type": "[]TxnOffsetCommitRequestPartition", "versions": "0+", "about": "The partitions inside the topic that we want to commit offsets for.", "fields": [ { "name": " { "name": "PartitionIndex", "type": "int32", "versions": "0+" }, "about": "The index of the partition within the topic." }, { "name": "CommittedOffset", "type": "int64", "versions": "0+" }, "about": "The message offset to be committed." }, { "name": "CommittedLeaderEpoch", "type": "int32", "versions": "2+", "default": "-1", "ignorable": true }, "about": "The leader epoch of the last consumed record." }, { "name": "CommittedMetadata", "type": "string", "versions": "0+", "nullableVersions": "0+", } "about": "Any associated metadata the client wants to keep." } ]} ]} ] } ] } ] } |
TxnOffsetCommitResponse
Key changes:
Namebounded to0-5,TopicIdadded at6+This follows the clean break pattern used byOffsetCommitv10,Fetchv13, andProducev13.GenerationIdrenamed toGenerationIdOrMemberEpoch. Aligns withOffsetCommitRequest. Source-level only; no wire format change.
TxnOffsetCommit Response (v6)
The response mirrors the request: v6 returns topic IDs instead of topic names.
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 28,
"type": "response",
"name": "TxnOffsetCommitResponse",
// ...
// Version 6 adds support for topic IDs and removes support for
// topic names. It can also return GROUP_ID_NOT_FOUND and
// STALE_MEMBER_EPOCH (KIP-1319).
"validVersions": "0-6", // CHANGED
"flexibleVersions": "3+",
// Supported errors:
// ...
// - | ||
| Code Block | ||
| ||
{ "apiKey": 28, "type": "response", "name": "TxnOffsetCommitResponse", // Starting in version 1, on quota violation, brokers send out responses // before throttling. // // Version 2 is the same as version 1. // // Version 3 adds illegal generation, fenced instance id, and unknown // member id errors. // // Version 4 adds support for new error code TRANSACTION_ABORTABLE // (KIP-890). // // Version 5 is the same with version 3 (KIP-890). // // Version 6 adds topic ID support (replacing topic names), and supports // GROUP_ID_NOT_FOUND, STALE_MEMBER_EPOCH, and UNKNOWN_TOPIC_ID error // codes. "validVersions": "0-6", "flexibleVersions": "3+", "fields": [ { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." }, { "name": "Topics", "type": "[]TxnOffsetCommitResponseTopic", "versions": "0+", "about": "The responses for each topic.", "fields": [ (version 6+) // NEW // - STALE_MEMBER_EPOCH (version 6+) // NEW // - UNKNOWN_TOPIC_ID (version 6+) // UpdatedNEW "fields": [ { "name": "NameThrottleTimeMs", "type": "stringint32", "versions": "0-5+" }, "ignorable": true, "entityType{ "name": "Topics", "type": "[]TxnOffsetCommitResponseTopic", "versions": "topicName0+", "about": "The topic name." }, "about": "The responses for each topic.", "fields": [ // New { "name": "TopicIdName", "type": "uuidstring", "versions": "6+0-5", "ignorable": true, // "about": "The unique topic ID." }, { "name": "Partitions", "type": "[]TxnOffsetCommitResponsePartition", "versions": "0+", "about": "The responses for each partition in the topic.", "fields": [ { "name": "PartitionIndex", "type": "int32", "versions": "0+", CHANGED "entityType": "topicName", "ignorable": true, "about": "The topic name." }, { "name": "TopicId", "type": "uuid", // NEW "versions": "6+", "ignorable": true, "about": "The partitiontopic indexID." }, { "name": "ErrorCodePartitions", "type": "int16[]TxnOffsetCommitResponsePartition", "versions": "0+", "about": "The error code, or 0 if there was no error." } ]} ]} ] } "fields": [ { "name": "PartitionIndex", "type": "int32", "versions": "0+" }, { "name": "ErrorCode", "type": "int16", "versions": "0+" } ] } ] } ] } |
New error codes:
| Error Code | When Returned | Prior Behavior (v0-5) |
|---|---|---|
| `GROUP_ID_NOT_FOUND` | The group does not exist and the request includes group membership information (`generationId >= 0`). | Mapped to `ILLEGAL_GENERATION`. |
| `STALE_MEMBER_EPOCH` | The member epoch is stale under the new consumer group protocol. | Mapped to `ILLEGAL_GENERATION`. |
| `UNKNOWN_TOPIC_ID` | The topic ID cannot be resolved by the broker. | N/A (topic names used). |
For v0-5, all existing error mappings are preserved.
Version 6 of the API supports the following new error code:
...