Status

Current state: Accepted

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 CodeWhen ReturnedPrior Behavior (v0-5)
GROUP_ID_NOT_FOUNDThe group does not exist and the request includes group membership information (`generationId >= 0`).Mapped to ILLEGAL_GENERATION.
STALE_MEMBER_EPOCHThe 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.

Proposed Changes

Broker-Side

Producer-Side

Compatibility, Deprecation, and Migration Plan

Test Plan

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

Rejected Alternatives