Status

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).

Motivation

Several features added to the The OffsetCommit API were never propagated has received several enhancements that were not applied to TxnOffsetCommit. :

• OffsetCommit v10 uses topic IDs instead of topic names

...

• (KIP-848).
• OffsetCommit v9+ 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-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.

This 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`:

{
  "apiKey": 28,
 

{
  "apiKey": 28,
  "type": "request",
   "listeners": ["broker"],
   "name": "TxnOffsetCommitRequest",
   // ...
 Version 1// isVersion the6 sameadds assupport version 0.
 for topic IDs and removes support for
  //
  // Version 2 adds the committed leader epoch.
  //
  topic names. It also returns GROUP_ID_NOT_FOUND when the group
  // Versiondoes 3not addsexist the member.id, group.instance.id and generation.id.
  //
 and STALE_MEMBER_EPOCH when the member epoch is
  // Versionstale 4 adds support forunder the new errorconsumer codegroup TRANSACTION_ABORTABLEprotocol (KIP-8901319).
  //
  // 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": [
    "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 producerepoch IDassociated inwith use by the transactionalproducer ID." },
      { "name": "ProducerEpochGenerationIdOrMemberEpoch", "type": "int16int32",  // CHANGED
      "versions": "03+",
      "aboutdefault": "The current-1",
      "about": "The 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 member ID generationassigned ofby 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+",
      "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." },                              
     -5",                                  // CHANGED
          "entityType": "topicName", "ignorable": true,
          "about": "The topic name." },
        { "name": "TopicId", "type": "uuid",                  // NEW
          "versions": "6+", "ignorable": true,
          "about": "The topic ID." },
        { "name": "Partitions",
          "type": "[]TxnOffsetCommitRequestPartition",
          "versions": "0+",
          "about": "The partitions inside the topic that we want to commit offsets for.",
          "fields": [
            { "name": "PartitionsPartitionIndex",
        "type": "[]TxnOffsetCommitRequestPartitionint32", "versions": "0+" },
         "about": "The partitions inside the topic that we want to commit offsets for.",
        "fields": [
        { "name": "PartitionIndexCommittedOffset", "type": "int32int64", "versions": "0+" },
          "about": "The index of the partition within the topic." },
        { "name": "CommittedOffsetCommittedLeaderEpoch", "type": "int64int32",
              "versions": "02+",
          "about"default": "The message offset to be committed." },
       -1", "ignorable": true },
            { "name": "CommittedLeaderEpochCommittedMetadata", "type": "int32string",
              "versions": "20+",
          "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." }
      ]}
    ]}
  ]
}

...

 "nullableVersions": "0+" }
          ]
        }
      ]
    }
  ]
}

Key changes:

• Name bounded to 0-5, TopicId added at 6+  This follows the clean break pattern used by OffsetCommit v10, Fetch v13, and Produce v13.
• GenerationId renamed to GenerationIdOrMemberEpoch. Aligns with OffsetCommitRequest. Source-level only; no wire format change.

TxnOffsetCommit Response (v6)

The response mirrors the request: v6 returns topic IDs instead of topic names.

{
   "apiKey": 28,
   "type": "response",
   "name": "TxnOffsetCommitResponse",
   // Starting...
 in version// 1,Version on6 quotaadds violation,support brokersfor sendtopic outIDs responses
  // before throttling.
  //
 and removes support for
  // Versiontopic 2names. isIt thecan samealso 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
  //return GROUP_ID_NOT_FOUND and
  // STALE_MEMBER_EPOCH (KIP-1319).
  "validVersions": "0-6",                                    // CHANGED
  "flexibleVersions": "3+",
  // Supported errors:
  // ...
  // - 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+)                         // Updated NEW
  "fields": [
      { "name": "NameThrottleTimeMs", "type": "stringint32", "versions": "0-5+",         },
       { "ignorablename": true"Topics",
      "entityTypetype": "topicName[]TxnOffsetCommitResponseTopic",               
        "versions": "0+",
      "about": "The responses for each topic name." },
   "fields": [
   // New
      { "name": "TopicIdName", "type": "uuidstring", "
          "versions": "6+0-5",            
        "ignorable": true,                                        //  CHANGED
             "aboutentityType": "topicName"The unique topic ID." },                          
     , "ignorable": true,
          "about": "The topic name." },
        { "name": "PartitionsTopicId",
        "type": "[]TxnOffsetCommitResponsePartitionuuid", "versions": "0+",
        "about": "The responses for each partition in the topic.",
        "fields": [
        { "name": "PartitionIndex", "type // NEW
          "versions": "int326+", "versionsignorable": "0+"true,
               "about": "The partitiontopic indexID." },
            { "name": "ErrorCodePartitions",
          "type": "int16[]TxnOffsetCommitResponsePartition",
          "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.

Proposed Changes

Broker

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.

Producer

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.

Compatibility, Deprecation, and Migration Plan

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.

Test Plan

The changes are covered by unit tests, integration tests, and system tests.

Rejected Alternatives

...

 "fields": [
            { "name": "PartitionIndex", "type": "int32", "versions": "0+" },
            { "name": "ErrorCode", "type": "int16", "versions": "0+" }
          ]
        }
      ]
    }
  ]
}

New error codes:

For v0-5, all existing error mappings are preserved.

Proposed Changes

Broker-Side

• Topic ID resolution. For v6 requests, the broker resolves topic IDs to names via the metadata cache before authorization and partition validation. This follows the same pattern as OffsetCommit v10. The per-partition error code returned for an invalid topic is determined in this order:
  • If the topic ID cannot be resolved by the broker, return UNKNOWN_TOPIC_ID for all partitions of that topic.
  • If the client is not authorized for the resolved topic name, return TOPIC_AUTHORIZATION_FAILED for all partitions of that topic.
  • If a requested partition does not exist, return UNKNOWN_TOPIC_OR_PARTITION for that partition only.
• Error mapping. For v6+, GROUP_ID_NOT_FOUND and STALE_MEMBER_EPOCH are returned directly instead of being mapped to ILLEGAL_GENERATION. For v0-5, the existing mapping is preserved.
• Persisting topic IDs. The broker populates the existing topicId tagged field in OffsetCommitValue v4 with the real topic ID instead of Uuid.ZERO_UUID. No new record schema is introduced.

Producer-Side

• sendOffsetsToTransaction API unchanged. The public API does not change.
• Topic ID resolution. When sendOffsetsToTransaction  is called and the broker supports v6, the producer adds the topics from the offset map to its metadata cache and waits for a refresh (bounded by max.block.ms) before building the request with the resolved topic IDs. No explicit cleanup is performed: the producer's metadata cache already evicts topics that have been idle for longer than metadata.max.idle.ms (default 5 minutes), which gives transient semantics for free. Topics that are committed repeatedly stay cached, which avoids a metadata wait on subsequent commits.
• Fallback. If the broker supports v6 but any topic in the offset map has an unavailable topic ID, the producer falls back to v5 for that request (all-or-nothing, same pattern as OffsetCommit's canUseTopicIds). In practice, this should never trigger since Transaction V2 requires KRaft, which always assigns topic IDs.
• Error handling. GROUP_ID_NOT_FOUND and STALE_MEMBER_EPOCH are treated as abortable errors. The application must abort the transaction. UNKNOWN_TOPIC_ID is treated as a retriable error consistent with UNKNOWN_TOPIC_OR_PARTITION. Since UnknownTopicIdException is a RetriableException, it falls into the existing retriable error handling path: the partition stays in pendingTxnOffsetCommits and the handler re-enqueues the request.
• Response mapping. For v6 responses (keyed by topic ID), the handler uses the topic ID to name mapping built during request construction to correlate responses back to TopicPartition.

Compatibility, Deprecation, and Migration Plan

• Version negotiation. v0-4 require no Transaction V2. v5 requires Transaction V2. v6 requires Transaction V2 and uses topic IDs with a fallback to v5.
• Rolling upgrades. No new record schema is introduced. The existing OffsetCommitValue v4 topicId tagged field (`ignorable: true`) is populated with real topic IDs by new brokers and ZERO_UUID by old brokers.
• Backward compatibility. Clients using v0-5 are unaffected. Brokers that do not support v6 negotiate a lower version.

Test Plan

The changes are covered by unit tests, integration tests, and system tests.

Rejected Alternatives

• Splitting Into Two Versions (Error Codes + Topic IDs). OffsetCommit split error codes (v9) and topic IDs (v10) because v9 could be used without KRaft. For TxnOffsetCommit, v5+ already requires Transaction V2 (which requires KRaft), so any broker
  supporting v6 necessarily supports both capabilities. Splitting would add protocol complexity without practical benefit.
• Keeping Topic Names Alongside Topic IDs in v6. The established Kafka convention s a clean break: Name in versions 0-N, TopicId in N+1 onward. Sending both adds redundancy and ambiguity.