Status

Current state: Under Discussion

Discussion thread: here 

JIRA: here 

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

The OffsetCommit API has received several enhancements that were not applied to TxnOffsetCommit:

Meanwhile, TxnOffsetCommit still relies on topic names only, hardcodes Uuid.ZERO_UUID when persisting offsets in OffsetCommitValue records, and maps both GROUP_ID_NOT_FOUND and STALE_MEMBER_EPOCH to ILLEGAL_GENERATION.

This KIP introduces TxnOffsetCommit v6 to close these gaps.

Public Interfaces

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,
  "type": "request",
  "listeners": ["broker"],
  "name": "TxnOffsetCommitRequest",
  // ...
  // Version 6 adds 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": "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": "GenerationIdOrMemberEpoch", "type": "int32",  // CHANGED
      "versions": "3+", "default": "-1",
      "about": "The generation of the group if using the classic group protocol or the member epoch if using the consumer protocol." },
    { "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": [
        { "name": "Name", "type": "string",
          "versions": "0-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": "PartitionIndex", "type": "int32", "versions": "0+" },
            { "name": "CommittedOffset", "type": "int64", "versions": "0+" },
            { "name": "CommittedLeaderEpoch", "type": "int32",
              "versions": "2+", "default": "-1", "ignorable": true },
            { "name": "CommittedMetadata", "type": "string",
              "versions": "0+", "nullableVersions": "0+" }
          ]
        }
      ]
    }
  ]
}

Key changes:

TxnOffsetCommit Response (v6)

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

{
  "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:
  // ...
  // - GROUP_ID_NOT_FOUND (version 6+)                       // NEW
  // - STALE_MEMBER_EPOCH (version 6+)                       // NEW
  // - UNKNOWN_TOPIC_ID (version 6+)                         // NEW
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+" },
    { "name": "Topics",
      "type": "[]TxnOffsetCommitResponseTopic", "versions": "0+",
      "about": "The responses for each topic.", "fields": [
        { "name": "Name", "type": "string",
          "versions": "0-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": "[]TxnOffsetCommitResponsePartition",
          "versions": "0+", "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:

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

None