Current state: [One of "Under Discussion", "Accepted", "Rejected"]
Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]
JIRA: here [Change the link from KAFKA-1 to your own ticket]
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Several features added to the OffsetCommit API were never propagated to TxnOffsetCommit. OffsetCommit v10 uses topic IDs instead of topic names, v9+ returns `GROUP_ID_NOT_FOUND` when the group does not exist, and v9+ propagates `STALE_MEMBER_EPOCH` for the new consumer group protocol (KIP-848). TxnOffsetCommit still relies on topic names only, hardcodes `Uuid.ZERO_UUID` when persisting offsets, and maps both error cases to `ILLEGAL_GENERATION`. This KIP introduces TxnOffsetCommit v6 to close these gaps.
{
"apiKey": 28,
"type": "request",
"listeners": ["broker"],
"name": "TxnOffsetCommitRequest",
// Version 1 is the same as version 0.
//
// Version 2 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": [
{ "name": "TransactionalId", "type": "string", "versions": "0+",
"entityType": "transactionalId",
"about": "The ID of the transaction." },
{ "name": "GroupId", "type": "string", "versions": "0+",
"entityType": "groupId",
"about": "The ID of the group." },
{ "name": "ProducerId", "type": "int64", "versions": "0+",
"entityType": "producerId",
"about": "The current producer ID in use by the transactional ID." },
{ "name": "ProducerEpoch", "type": "int16", "versions": "0+",
"about": "The current epoch associated with the producer ID." },
{ "name": "GenerationId", "type": "int32", "versions": "3+",
"default": "-1",
"about": "The generation of the consumer." },
{ "name": "MemberId", "type": "string", "versions": "3+",
"default": "",
"about": "The member ID assigned by the group coordinator." },
{ "name": "GroupInstanceId", "type": "string", "versions": "3+",
"nullableVersions": "3+", "default": "null",
"about": "The unique identifier of the consumer instance provided by end user." },
{ "name": "Topics", "type" : "[]TxnOffsetCommitRequestTopic",
"versions": "0+",
"about": "Each topic that we want to commit offsets for.", "fields": [
// Updated
{ "name": "Name", "type": "string", "versions": "0-5",
"ignorable": true, "entityType": "topicName",
"about": "The topic name." },
// New
{ "name": "TopicId", "type": "uuid", "versions": "6+",
"ignorable": true,
"about": "The unique topic ID." },
{ "name": "Partitions",
"type": "[]TxnOffsetCommitRequestPartition", "versions": "0+",
"about": "The partitions inside the topic that we want to commit offsets for.",
"fields": [
{ "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." }
]}
]}
]
} |
{
"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": [
// Updated
{ "name": "Name", "type": "string", "versions": "0-5",
"ignorable": true, "entityType": "topicName",
"about": "The topic name." },
// New
{ "name": "TopicId", "type": "uuid", "versions": "6+",
"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+",
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." }
]}
]}
]
} |
Version 6 of the API supports the following new error code:
- `GROUP_ID_NOT_FOUND` - The group does not exist. Previously returned as `ILLEGAL_GENERATION`.
- `STALE_MEMBER_EPOCH` - The member epoch is stale under the new consumer group protocol. Previously returned as `ILLEGAL_GENERATION`.
- `UNKNOWN_TOPIC_ID` - The provided topic ID cannot be resolved to a known topic.
When the broker receives a v6 request, it resolves topic IDs to topic names and returns `UNKNOWN_TOPIC_ID` for unresolvable IDs. The resolved names are used for authorization, partition validation, and offset storage. The topic ID is persisted in the `OffsetCommitValue` record, replacing the hardcoded `Uuid.ZERO_UUID`. For v6+ requests, the broker returns `GROUP_ID_NOT_FOUND` and `STALE_MEMBER_EPOCH` directly instead of mapping them to `ILLEGAL_GENERATION` when the consumer rebalance protocol is used. Existing behavior for v0-5 is preserved.
The sendOffsetsToTransaction API remains unchanged. It accepts a Map<TopicPartition, OffsetAndMetadata> and a ConsumerGroupMetadata. When sendOffsetsToTransaction is called, the producer fetches metadata for the topics in the offsets map and waits for the response. This ensures topic IDs are available for topics the producer has not previously produced to. Once metadata is resolved, if all topics have a topic ID, the producer builds a v6 request. If any topic lacks a topic ID, it falls back to v5. The producer treats `STALE_MEMBER_EPOCH` and `GROUP_ID_NOT_FOUND` as abortable errors and `UNKNOWN_TOPIC_ID` as a fatal error.
The producer automatically negotiates the version based on broker support and topic ID availability. `LAST_STABLE_VERSION_BEFORE_TRANSACTION_V2 = 4` caps non-TV2 clients at v4, so v6 effectively requires Transaction V2 (KIP-890). No existing versions are deprecated.
The changes are covered by unit tests, integration tests, and system tests.
None