...

This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: Under Discussion Voting

Discussion thread: here 

JIRA: [KAFKA-19883](

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

1 Motivation

A recordToday, the Producer's side effects (output writes) and its source acknowledgment must be committed atomically. Either both succeed, or neither does.

1. Share consumer polls records -> broker sets them to ACQUIRED.
2. Framework processes and acknowledges records (implicit or explicit).
3. Checkpoint fails before sink outputs are committed.
4. Records are permanently ACKNOWLEDGED (not redelivered) -> data loss.
5. Note:
   1. Share records become terminal when acked: RecordState
   2. Ack path today is irreversible once ACKNOWLEDGED: SharePartition.acknowledge()

Goal: Enable exactly-once read semantics via transactional acknowledgements.

We will be following the similar existing pattern we have in Kafka Producer Transactions.

Use Cases

• Consume‑Transform‑Produce (CTP): bind acks to Kafka output transaction.
• Consume‑only frameworks: transactional acks independent of producer.

Frameworks:

- Apache Flink: Exactly-once checkpointing with Share Group sources
- Apache Spark: Structured Streaming with Share Group consumers
- Any coordinator-worker streaming framework requiring atomic acknowledgements

Although in this KIP we are refering Flink Stream processing engine few places but it is valid for any stream processing engine - almost all have same pattern of offset commits or replay when something crashes. 

1.1 Background: Share Groups

Share groups (KIP-932) allow multiple consumers to read from the same partition concurrently, with the broker controlling per-record delivery via an acquisition-lock mechanism. Each record passes through a state machine on the broker:

AVAILABLE → ACQUIRED → ACKNOWLEDGED (terminal)
                    → ARCHIVED (terminal, rejected or max delivery exceeded)
                    → AVAILABLE (released for redelivery)

Today, share groups support two acknowledgment modes:

• Implicit: Records from a previous poll() are automatically acknowledged on the next poll().
• Explicit: The application calls acknowledge(record, AcknowledgeType) for each record, then commitSync().

In both modes, acknowledgments are committed immediately and irrevocably. Once a record enters ACKNOWLEDGED state in the SharePartition (managed on the broker), it is never redelivered.

Relevant existing code:

• RecordState enum: kafka/server/src/main/java/org/apache/kafka/server/share/fetch/RecordState.java
• SharePartition.acknowledge(): kafka/core/src/main/java/kafka/server/share/SharePartition.java
• ShareConsumer interface: kafka/clients/src/main/java/org/apache/kafka/clients/consumer/ShareConsumer.java
• AcknowledgeType enum (ACCEPT/RELEASE/REJECT/RENEW): kafka/clients/src/main/java/org/apache/kafka/clients/consumer/AcknowledgeType.java

1.2 Why Existing Consumer-Group Exactly-Once Doesn't Apply

With traditional consumer groups, Flink avoids this problem by replaying from a saved offset:

1. KafkaSourceReader.snapshotState() saves offsets in Flink's state.
2. On failure recovery, consumer.seek(savedOffset) replays from the checkpoint.
3. Kafka offset commits (via sendOffsetsToTransaction()) are cosmetic — Flink state is the source of truth.

Share groups have no seek(). The broker controls which records are delivered. Once acknowledged, records are gone. Therefore, acknowledgment itself must become the transactional boundary, not just a cosmetic side-effect.

1.3 Existing Pattern: sendOffsetsToTransaction()

Kafka already solves the identical problem for consumer-group offsets via KafkaProducer.sendOffsetsToTransaction():

1. The producer includes consumer-group offsets in its ongoing transaction.
2. When the transaction commits, both output records and offset commits become visible atomically.
3. On abort, neither is visible — the consumer re-reads from the old offset.

This KIP applies the same pattern to share-group acknowledgments. Instead of committing to __consumer_offsets, we commit to __share_group_state.

Existing code this KIP mirrors:

• KafkaProducer.sendOffsetsToTransaction(): kafka/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
• AddOffsetsToTxnRequest.json: kafka/clients/src/main/resources/common/message/AddOffsetsToTxnRequest.json
• GroupCoordinator.completeTransaction(): handles WriteTxnMarkers for __consumer_offsets
• ShareCoordinatorShard.replayEndTransactionMarker(): already exists, handles transaction markers for __share_group_state

2. Use Cases

2.1 Consume-Transform-Produce (CTP)

An application reads from a share group, transforms records, and produces output to another Kafka topic. Both output and acknowledgments must commit atomically.

producer.beginTransaction();

ConsumerRecords<K,V> records = shareConsumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<K,V> record : records) {
    ProducerRecord<K,V> output = transform(record);
    producer.send(output);
}

// Bind share acks to this transaction (NEW API)
producer.sendShareAcksToTransaction(
    ShareAcknowledgements.fromRecords(records, AcknowledgeType.ACCEPT),
    shareConsumer.groupMetadata()
);

producer.commitTransaction();
// Output records AND share acks commit atomically

2.2 Flink / Spark Source (No Producer)

A streaming framework reads from a share group as a source. There is no Kafka producer in the pipeline — the output may go to a database, filesystem, or another system. The framework needs to commit share acks transactionally, coordinated with its own checkpointing.

TransactionalShareAcknowledger acknowledger = new TransactionalShareAcknowledger(props);
acknowledger.initTransactions();

// On checkpoint complete:
acknowledger.commitAcknowledgements(bufferedAcks, shareGroupId);
// Internally: beginTransaction → sendShareAcksToTransaction → commitTransaction

2.3 Flink End-to-End Exactly-Once (Source + Sink)

When a Flink pipeline reads from a Kafka share group and writes to a Kafka sink topic, we achieve end-to-end exactly-once by binding both sink output and source acknowledgments to the same Kafka transaction:

Checkpoint lifecycle:
  prepareCommit()              → flush sink records, pre-commit Kafka txn
                               → include share acks in the same transaction
  snapshotState()              → save txn metadata + buffered acks
  notifyCheckpointComplete()   → commitTransaction() (acks + output atomically)
  On failure                   → abortTransaction() (acks + output both rolled back)

3 Public Interfaces

3.1 KafkaProducer API Addition

// In org.apache.kafka.clients.producer.KafkaProducer:

/**
 * Sends share-group acknowledgments as part of the current transaction.
 * Mirrors sendOffsetsToTransaction() but writes to __share_group_state
 * instead of __consumer_offsets.
 *
 * @param acknowledgements Map of TopicPartition to list of acknowledgment batches
 * @param groupMetadata    The share group metadata (group ID, member ID, generation)
 * @throws IllegalStateException if no transaction is in progress
 * @throws ProducerFencedException if the producer is fenced
 */
public void sendShareAcksToTransaction(
    Map<TopicPartition, ShareAcknowledgements> acknowledgements,
    ShareGroupMetadata groupMetadata
) throws ProducerFencedException;

This mirrors sendOffsetsToTransaction(). The reason a new method is needed (instead of reusing the existing one) is that:

• Different storage topic: acks go to __share_group_state, not __consumer_offsets.
• Different coordinator: the ShareCoordinator handles ack persistence, not GroupCoordinator.
• Different semantics: acks are per-record state transitions, not per-partition offsets.

3.2 TransactionalShareAcknowledger (Standalone)

For frameworks that do not use a KafkaProducer in the pipeline:

public class TransactionalShareAcknowledger implements Closeable {

    public TransactionalShareAcknowledger(Properties config);

    /** Initialize the internal transactional producer. Call once. */
    public void initTransactions();

    /**
     * Atomically commit share acknowledgments.
     * Internally executes: beginTransaction → sendShareAcksToTransaction → commitTransaction.
     * This is NOT a single RPC. It orchestrates the standard 2PC protocol.
     */
    public void commitAcknowledgements(
        Map<TopicPartition, ShareAcknowledgements> acks,
        String groupId
    );

    /** Abort any in-progress transactional acknowledgment. */
    public void abortAcknowledgements();

    public void close();

 }

Clarification: commitAcknowledgements() is a convenience wrapper. It internally calls three operations in sequence:

1. beginTransaction()
2. sendShareAcksToTransaction(acks, groupMetadata)
3. commitTransaction()

It does NOT introduce a new single-RPC path. It uses the standard 2PC protocol.

3.3 ShareGroupMetadata

public class ShareGroupMetadata {
    private final String groupId;
    private final String memberId;
    private final int generationId;
    // constructor, getters

 }

3.4 ShareAcknowledgements

public class ShareAcknowledgements {
    private final List<ShareAcknowledgementBatch> batches;

    public static ShareAcknowledgements fromRecords(
        ConsumerRecords<?, ?> records, AcknowledgeType type);

    // Each batch: firstOffset, lastOffset, acknowledgeType

 }

3.5 New Metrics

4 Proposed Changes

4.1 Reuse of Existing 2PC Protocol

This KIP reuses Kafka's existing two-phase commit protocol. No new coordinator type or consensus protocol is introduced.

Transaction lifecycle (identical to existing):

EMPTY → ONGOING → PREPARE_COMMIT → COMPLETE_COMMIT
                → PREPARE_ABORT  → COMPLETE_ABORT

What changes is which partitions are added to the transaction:

The WriteTxnMarkers request dispatches commit/abort markers to all partitions in the transaction set. If only sendShareAcksToTransaction() was called, markers go only to __share_group_state. If both sendOffsetsToTransaction() and sendShareAcksToTransaction() were called in the same transaction, markers go to both. The transaction coordinator does not distinguish between these — it just tracks partition sets.

4.2 Wire Protocol Changes

New request: AddShareAcksToTxnRequest (mirrors AddOffsetsToTxnRequest)

{
  "apiKey": TBD,
  "type": "request",
  "name": "AddShareAcksToTxnRequest",
  "validVersions": "0",
  "fields": [
    { "name": "TransactionalId", "type": "string", "versions": "0+" },
    { "name": "ProducerId", "type": "int64", "versions": "0+" },
    { "name": "ProducerEpoch", "type": "int16", "versions": "0+" },
    { "name": "GroupId", "type": "string", "versions": "0+" },
    { "name": "Topics", "type": "[]AddShareAcksToTxnTopic", "versions": "0+",
      "fields": [
        { "name": "Name", "type": "string", "versions": "0+" },
        { "name": "Partitions", "type": "[]int32", "versions": "0+" }
      ]
    }
  ]
}

Purpose: Tells the TransactionCoordinator to add the __share_group_state partition(s) for the given {groupId, topicPartition} pair(s) to the producer's ongoing transaction. The partition is determined by ShareCoordinator.partitionFor(groupId, topicPartition), where topicPartition refers to the original data topic partition being acknowledged, and the function maps it to the corresponding __share_group_state internal partition that stores the state for that share-group + data-partition combination.

New request: TxnShareAcknowledgeRequest (mirrors TxnOffsetCommitRequest)

sendOffsetsToTransaction(offsets, consumerGroupMetadata) allows EOS in read-process-write topologies that consume from

regular consumer groups. With KIP-932 introducing share groups, the equivalent capability is missing for share-group consumers.

This blocks share-group adoption in:

 1.  MirrorMaker and other Kafka-to-Kafka mirroring/forwarding pipelines.
 2.  Kafka Streams stateless topologies that want to use share groups for parallelism beyond partition count.

 3.  Atomic DLQ write in different connectors

The Goal: Atomic "Read-Process-Write"

Enable Exactly-Once Semantics (EOS) by making acknowledgments part of a Kafka transaction.

Either the output is produced AND the source record is acknowledged, or neither happens. This KIP is scopred for Kafka producer write AND consumes from a share group.

1.1 Background: Share Groups

Current State Machine:

• AVAILABLE: Ready for delivery.

• ACQUIRED: Locked by a consumer.

• RENEW

• ACKNOWLEDGED / ARCHIVED: Terminal states; records are never redelivered.

• AVAILABLE (Released): Returned to the pool for redelivery.

Current Acknowledgment Modes:

1. Implicit: Automatic ack on the subsequent poll().

2. Explicit: Manual ack via acknowledge(record, type) followed by commitSync()

2 Public Interfaces

Public API additions:

• Producer.sendShareAcknowledgementsToTransaction(Map<TopicIdPartition, List<AcknowledgementBatch>>, ShareGroupMetadata).
• ShareGroupMetadata class (groupId, memberId, memberEpoch, optional groupInstanceId).
• ShareConsumer.shareGroupMetadata() to obtain the above.

Producer (clients module)

public interface Producer<K, V> {

    /**
     * Sends a list of share-group acknowledgements to the consumer-coordinator and marks
     * them for atomic commit alongside the records produced in this transaction.
     *
     * The acknowledgements are staged on the broker in a TX_PENDING state until the
     * transaction is committed or aborted. On commit, ACCEPT records transition to
     * ACKNOWLEDGED and REJECT records transition to ARCHIVING/ARCHIVED. On abort,
     * all staged records revert to ACQUIRED and remain owned by the original consumer
     * until either the consumer re-acknowledges them or the acquisition lock expires.
     *
     * @param acknowledgements Per-partition list of acknowledgement batches. Only
     *                         AcknowledgeType.ACCEPT (1) and AcknowledgeType.REJECT (3)
     *                         are valid inside a transaction. RELEASE (2), RENEW (4),
     *                         and GAP (0) are rejected with InvalidRecordStateException.
     * @param groupMetadata    Snapshot of the share consumer's group identity, obtained
     *                         from ShareConsumer.shareGroupMetadata().
     *
     * @throws IllegalStateException if no transaction is in progress, or if the producer
     *                               is not transactional.
     * @throws ProducerFencedException if another producer with the same transactionalId
     *                                 has fenced this one.
     * @throws UnsupportedVersionException if the cluster does not advertise apiKey 93
     *                                     (TxnShareAcknowledge) via ApiVersions.
     * @throws GroupAuthorizationException if the configured principal cannot Write to
     *                                     the share group.
     * @throws InvalidProducerEpochException if the producer's epoch is stale.
     * @throws KafkaException for other non-fatal errors that may be retried by aborting
     *                        the transaction and retrying the entire read-process-write
     *                        loop.
     *
     * Threading: Returns immediately after enqueuing the request for the producer's
     *	background Sender thread. The broker has NOT processed it yet.
     * 	Any error is reported when commitTransaction() or abortTransaction() is later called.
     */
    void sendShareAcknowledgementsToTransaction(
        Map<TopicIdPartition, List<AcknowledgementBatch>> acknowledgements,
        ShareGroupMetadata groupMetadata
    ) throws ProducerFencedException;
}

ShareConsumer (clients module)

public interface ShareConsumer<K, V> {

    /**
     * Returns an immutable snapshot of this consumer's share-group identity for use
     * with Producer.sendShareAcknowledgementsToTransaction.
     *
     * The snapshot captures groupId, memberId, and memberEpoch atomically; if a
     * rebalance changes the memberEpoch between the snapshot and the producer call,
     * the broker will reject the staging request with STALE_MEMBER_EPOCH and the
     * user must abort the transaction and retry the read-process-write loop.
     *
     * @throws UnsupportedVersionException if the cluster does not support KIP-1289.
     * @throws TimeoutException if the snapshot cannot be obtained within the
     *                          configured default.api.timeout.ms.
     *
     * Threading: thread-safe; safe to call concurrently with poll() and acknowledge().
     */
    ShareGroupMetadata shareGroupMetadata();
}

ShareGroupMetadata (new class in clients module, package o.a.k.clients.consumer)

/* Thread-safe, Immutable, Concurrent with poll/acknowledge */
public final class ShareGroupMetadata {
    public ShareGroupMetadata(String groupId, String memberId, int memberEpoch);
    public String groupId();
    public String memberId();
    public int memberEpoch();
    @Override public boolean equals(Object other);
    @Override public int hashCode();
    @Override public String toString();
}

Wire protocol additions:

• New TxnShareAcknowledge RPC (apiKey 93 or next free).
• Schema mirrors TxnOffsetCommitRequest shape but carries AcknowledgementBatch instead of OffsetCommitInfo.

New RPC: TxnShareAcknowledgeRequest (apiKey 93)
Listeners: broker.
Acknowledge type values: 0=Gap, 1=Accept, 2=Release, 3=Reject, 4=Renew.
Transactional constraint: only 1 (Accept) and 3 (Reject) are valid inside a transaction.
A batch containing any other value is rejected with INVALID_RECORD_STATE.

New RPC: TxnShareAcknowledgeResponse (apiKey 94, v0)

Top-level supported errors:
- GROUP_AUTHORIZATION_FAILED
- TOPIC_AUTHORIZATION_FAILED
- TRANSACTIONAL_ID_AUTHORIZATION_FAILED
- TRANSACTIONAL_ID_NOT_FOUND
- INVALID_PRODUCER_EPOCH / PRODUCER_FENCED
- INVALID_PRODUCER_ID_MAPPING
- INVALID_TXN_STATE
- UNKNOWN_MEMBER_ID
- STALE_MEMBER_EPOCH
- TRANSACTION_ABORTABLE (KIP-890)
- UNKNOWN_SERVER_ERROR

Per-partition supported errors:
- UNKNOWN_TOPIC_OR_PARTITION / UNKNOWN_TOPIC_ID
- NOT_LEADER_OR_FOLLOWER (with CurrentLeader populated)
- INVALID_RECORD_STATE
- INVALID_REQUEST
- KAFKA_STORAGE_ERROR

Existing RPCs NOT changed

- WriteTxnMarkers (apiKey 27): NO schema change. The broker hooks the
  existing marker arrival in KafkaApis.handleWriteTxnMarkersRequest and broadcasts
  to all SharePartition instances on that broker. Per-record fencing by
  (producerId, producerEpoch) filters out non-participants. This means the
  TransactionCoordinator does NOT need to track share-partitions as transaction
  participants explicitly — the broadcast is correct because the per-record
  state filter is the source of truth.

- AddPartitionsToTxn (apiKey 24): NO schema change. KIP-1289 does NOT add share
  partitions as transaction participants. The transactional output-write path
  continues to use AddPartitionsToTxn for the output topics; the share-group
  staging path uses TxnShareAcknowledge directly (TV2-only, single round trip).

- ApiVersions (apiKey 18): NO schema change, but the broker MUST advertise
  apiKey 93 only when feature-flag share.version >= 2 (see Enablement section).

- TxnOffsetCommit (apiKey 28): NO schema change. Independent path for regular
  consumer groups; share groups use the new RPC.

Internal Topic Record Schemas

__transaction_state: unchanged

share-partition participation in transactions is tracked via the existing TransactionMetadata.topicPartitions field

(mechanism: broker-internal AddPartitionsToTxn triggered by TxnShareAcknowledge handler, mirroring how __consumer_offsets-N is registered via AddOffsetsToTxn for regular consumer groups)

Flow — Regular Consumer Group 

• Producer writes to output topic
  • Source Component: TransactionManager (Client)
  • Action: Sends AddPartitionsToTxn("output-A-0") to TxnCoord.
  • State Change: TxnCoord records participants = {output-A-0} in __transaction_state.
• Producer calls sendOffsetsToTransaction("my-group")
  • Source Component: TransactionManager (Client) & TransactionCoordinator (Broker Core)
  • Action: Sends AddOffsetsToTxn("my-group") to TxnCoord.
  • Execution: TxnCoord computes hash("my-group") = (let's say) 7.
  • State Change: Records participants = {output-A-0, __consumer_offsets-7} in __transaction_state .
• Producer commits
  • Source Component: TransactionCoordinator (Broker Core)
  • Execution: Reads participants from __transaction_state and resolves brokers via MetadataCache:
    • output-A-0  — > Broker 3
    • __consumer_offsets-7 ---→ Broker 4
  • Action: Sends WriteTxnMarkers to Broker 3 AND Broker 4.
• Each broker writes a control batch to its log
  • Source Component: KafkaApis & GroupCoordinator (Broker Core)
  • Execution:
    • Broker 3: Appends COMMIT batch to output-A-0 log (makes records visible to read_committed consumers).
    • Broker 4: Appends COMMIT batch to __consumer_offsets-7 log (makes staged offsets visible to GroupCoordinator).

Flow — Share Group (THIS KIP)

• producer.send("output-A", record)
  • Source Component: TransactionManager (Client)
  • Action: Sends AddPartitionsToTxnRequest(["output-A-0"]).
  • State Change: Updates local state to TransactionMetadata.topicPartitions = {output-A-0}.
• producer.sendShareAcknowledgementsToTransaction(acks, meta)
  • Source Component: TransactionManager (Client)  ---→  KafkaApis (Broker Core)
  • Action: Sends TxnShareAcknowledgeRequest to Broker 5 (the SharePartition leader - lets say it is broker 5).
  • Broker Execution: KafkaApis intercepts the request and performs two background operations:
    1. Resolves the state partition: shareCoordinatorPartition = (__share_group_state, partitionFor(groupId)).
    2. First call txnCoordinator.handleAddPartitionsToTransaction(transactionalId, producerId, producerEpoch, Set.of(shareCoordinatorPartition), callback, TV_2, requestLocal) 
       1. Then callback → Calls txnCoordinator.handleAddPartitionsToTransaction to dynamically register __share_group_state-N in __transaction_state.
       2. persists to __transaction_state via TransactionLogValue: topicPartitions = {output-A-0, __share_group_state-N}
  • State Change: TransactionMetadata now tracks both topicPartitions = {output-A-0, __share_group_state-N}.
  • Staging Phase: Only on success, SharePartitionManager stages the transactional acknowledgements, transitioning the internal InFlightState to TX_PENDING.
• producer.commitTransaction()
  • Source Component: Client  -------> TransactionCoordinator(Broker Core)
  • Action: Sends EndTxnRequest(COMMIT).
  • Execution: TxnCoord reads the topic partitions from state and resolves the physical leaders via MetadataCache:
    • output-A-0 --------> Broker 3
    • __share_group_state-N  -----→ Broker 5
  • Action: Dispatches WriteTxnMarkers to both brokers.
• Each broker handles the transaction markers
  • Execution on Broker 3 (Output Side):
    • Appends the COMMIT control batch to the output-A-0 log.
    • KIP-1289 hook runs (no-op since no TX_PENDING here)
  • Execution on Broker 5 (Share Side):
    • Appends the COMMIT control batch to the __share_group_state-N log.
    • Fires the KIP-1289 hook: Triggers sharePartitionManager.applyTxnMarker(COMMIT).
    • Scans the internal partitionCache, finds the matching TX_PENDING flight states, and commits them to ACKNOWLEDGED.
• Transaction Completion
  • Action: Both brokers return successful ACKs back to TxnCoord.
  • State Change: TxnCoord transitions the state to COMPLETE_COMMIT.
  • Result: The client-side blocking producer.commitTransaction() call returns successfully.

[Note: Only TV 2 supporting - broker-side auto-registration and proper epoch fencing, do not want to maintain TV1 code. TV2 is already default option since 4.0]

__share_group_state schema unchanged 

 no ShareCoordinatorShard changes; no snapshot replay changes; no compatibility risk.

Inter-broker protocol unchanged

 existing handleAddPartitionsToTransaction accepts arbitrary TopicPartition; no new code path on TxnCoord side

State machine additions:

• New transient state TX_PENDING 
• On WriteTxnMarkers commit: TX_PENDING(ACCEPT) → ACKNOWLEDGED; same for RELEASE and REJECT.
• On WriteTxnMarkers abort: TX_PENDING(*) → back to ACQUIRED (lock continues; consumer can retry the work).

Pseudo code

// One-time setup
shareConsumer.subscribe(List.of("source-topic"));
producer.initTransactions();   // exactly once per producer instance

// CTP loop
while (running) {
    ConsumerRecords<K, V> records = shareConsumer.poll(Duration.ofSeconds(5));
    if (records.isEmpty()) continue;

    try {
        producer.beginTransaction();

        Map<TopicIdPartition, Acknowledgements> acks = new HashMap<>();

        for (ConsumerRecord<K, V> record : records) {
            V output = process(record);                                          
            producer.send(new ProducerRecord<>("destination-topic", output));    
            acks.computeIfAbsent(
                    new TopicIdPartition(record.topicId(), 
                            new TopicPartition(record.topic(), record.partition())),
                    k -> Acknowledgements.empty()
                ).add(record.offset(), AcknowledgeType.ACCEPT);                  // accumulate
        }

        // Compress acks to AcknowledgementBatch wire form and stage in the txn.
        producer.sendShareAcknowledgementsToTransaction(
            toBatches(acks),  

{
  "apiKey": TBD,
  "type": "request",
  "name": "TxnShareAcknowledgeRequest",
  "validVersions": "0",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "0+" },
    { "name": "TransactionalId", "type": "string", "versions": "0+" },
    { "name": "ProducerId", "type": "int64", "versions": "0+" },
    { "name": "ProducerEpoch", "type": "int16", "versions": "0+" },
    { "name": "Topics", "type": "[]TxnShareAcknowledgeTopic", "versions": "0+",
      "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+" },
        { "name": "Partitions", "type": "[]TxnShareAcknowledgePartition", "versions": "0+",
          "fields": [
            { "name": "Partition", "type": "int32", "versions": "0+" },
            { "name": "AcknowledgementBatches", "type": "[]AcknowledgementBatch", "versions": "0+",
              "fields": [
   // Map<TopicIdPartition, List<AcknowledgementBatch>>
            shareConsumer.shareGroupMetadata()  { "name": "FirstOffset", "type": "int64", "versions": "0+" },
                       { "name": "LastOffset", "type": "int64", "versions": "0+" },
   
        );

     { "name": "AcknowledgeType", "type": "int8", "versions": "0+" }
      producer.commitTransaction();    // BLOCKS — true synchronisation point
        // ]
Records are now both produced AND acknowledged ATOMICALLY.
     }
   // Do NOT call shareConsumer.acknowledge(...) on these ]
records.
		//    if called mistakenly  }
      ]
    }
  ]
}

Purpose: Sent to the ShareCoordinator to write acknowledgments as pending (uncommitted) within the transaction. The acks become visible only when the transaction commits.

4.3 ShareCoordinator Changes

The ShareCoordinator must handle transaction completion, mirroring GroupCoordinator.completeTransaction():

// In ShareCoordinatorShard (NEW method):
public CoordinatorResult<Void, CoordinatorRecord> completeTransaction(then non-transactional acknowledge is rejected (TX_PENDING to ACKNOWLEDGED - InvalidRecordStateException); the transactional path works.

    } catch (ProducerFencedException | UnsupportedVersionException fatal) {
    long producerId,
   throw shortfatal; producerEpoch,
    TransactionResult result    // COMMIT or ABORT
) {
    if (result == TransactionResult.COMMIT) {
     // unrecoverable — kill the process
    //} Materializecatch pending(KafkaException transactionalabortable) acks{
 into share-group state
      producer.abortTransaction();  return applyPendingAcknowledgements(producerId, producerEpoch);
 // BLOCKS — }txn elsereverted; {
staged records revert to ACQUIRED on broker
  // Discard pending transactional acks
  // Records will be re-delivered on returnnext discardPendingAcknowledgementspoll(producerId, producerEpoch); retry naturally.
    }
}

The existing hook ShareCoordinatorShard.replayEndTransactionMarker() already exists for replaying transaction markers during log recovery. This KIP extends it to also handle live transaction completion.

4.4 State Storage

Transactional acks are written to __share_group_state using CoordinatorRuntime.scheduleTransactionalWriteOperation(). This follows the same pattern used by GroupCoordinator for transactional offset commits to __consumer_offsets.

Records in __share_group_state are written with the producer's producerId and producerEpoch, making them part of the transaction. They become readable by other consumers only after the transaction commits and the WriteTxnMarkers COMMIT marker is written.

5 Flow diagram 

CLIENT --> BROKER communication (over the Kafka wire protocol):

  Share Consumer -----> Share Group Coordinator
    ShareFetch, ShareAcknowledge, ShareGroupHeartbeat

  TransactionManager --> Transaction Coordinator
    InitProducerId, AddPartitionsToTxn, AddShareAcksToTxn(NEW), EndTxn

  TransactionManager --> Share Group Coordinator
    TxnShareAcknowledgeRequest (NEW)

  Producer ------------> Data Partition Leaders
    ProduceRequest


BROKER --> BROKER communication (internal, not client-visible):

  Transaction Coordinator --> Data Partition Leaders
    WriteTxnMarkers (COMMIT/ABORT control record)

  Transaction Coordinator --> Group Coordinator (via __consumer_offsets leader)
    WriteTxnMarkers (materializes transactional offset commits)

  Transaction Coordinator --> Share Group Coordinator (via __share_group_state leader) [NEW]
    WriteTxnMarkers (materializes transactional share acks)

Abort case: 

Image Removed

6. Corner Cases

Case 1: Crash BEFORE beginTransaction()

poll() -> records [0..9] ACQUIRED
<CRASH>

What happens:

• No transaction was started, no output produced, no acks sent
• ACQUIRED locks on records 0-9 expire (controlled by share.record.lock.duration.ms)
• Broker re-delivers records 0-9 to another member of the share group
• No data loss, no duplicates

Case 2: Crash AFTER send() but BEFORE commitTransaction()

poll() -> records [0..9] ACQUIRED
beginTransaction()
send(output for 0-9)       ← records written to broker, but transactional
<CRASH>

What happens:

• Output records are in the partition log but invisible (read_committed consumers skip uncommitted)
• No EndTxnRequest was sent
• TC detects transaction timeout (transaction.timeout.ms), auto-aborts
• TC sends WriteTxnMarkers(ABORT) -> output records get ABORT marker, permanently invisible
• Share group: no acks were sent, ACQUIRED locks expire, records 0-9 re-delivered
• No data loss, no duplicates

Case 3: Crash AFTER sendShareAcksToTransaction() but BEFORE commitTransaction()

poll() -> records [0..9] ACQUIRED
beginTransaction()
send(output)
sendShareAcksToTransaction()  ← acks written to SGC as PENDING
<CRASH>

What happens:

• Output records: in log, invisible (uncommitted)
• Share acks: stored in SGC as PENDING (not materialized)
• TC detects timeout, sends WriteTxnMarkers(ABORT) to both data partitions AND __share_group_state
• Output: ABORT marker written, records permanently invisible
• Share acks: SGC discards pending acks, locks expire, records 0-9 re-delivered
• No data loss, no duplicates -- this is the critical case that KIP-1289 solves

Case 4: Crash AFTER EndTxn(COMMIT) sent but BEFORE WriteTxnMarkers completes

poll() -> records [0..9] ACQUIRED
beginTransaction()
send(output)
sendShareAcksToTransaction()
commitTransaction()            ← EndTxn sent, PREPARE_COMMIT logged
<TC crashes or broker restart>

What happens:

• PREPARE_COMMIT is durable in __transaction_state log
• When TC recovers (or new leader elected for __transaction_state partition), it replays the log
• TC sees PREPARE_COMMIT, resumes sending WriteTxnMarkers(COMMIT) to all partitions
• Output records become visible, share acks materialized
• No data loss, no duplicates -- the two-phase commit guarantees completion

Case 5: ProducerFencedException (zombie detection)

Instance A: beginTransaction(), send(), ...
Instance B: initTransactions() with same transactional.id
            ← TC bumps epoch, A is now a zombie
Instance A: commitTransaction()
            ← TC rejects: ProducerFencedException

What happens:

• Instance A's transaction is aborted by the TC (epoch fenced)
• Any pending output and share acks from A are discarded
• Instance B takes over, re-processes the records
• No data loss, no duplicates -- fencing prevents split-brain

Case 6: abortTransaction() called explicitly

poll() -> records [0..9] ACQUIRED
beginTransaction()
send(output for 0-4)
record 5 fails validation
abortTransaction()             ← explicit abort

What happens:

• TC writes PREPARE_ABORT, sends WriteTxnMarkers(ABORT)
• Output records 0-4: ABORT marker, permanently invisible
• Share acks (if any sent): discarded
• ACQUIRED locks expire, all records 0-9 re-delivered
• Application gets fresh batch, can retry
• No data loss

Case 7: Partial produce failure (network error to one partition)

beginTransaction()
send("enriched-orders-0", rec1)   ← success
send("enriched-orders-1", rec2)   ← network error, KafkaException

What happens:

• Application catches KafkaException, calls abortTransaction()
• Same as Case 6: everything rolled back, records re-delivered
• No data loss

The Guarantee Matrix

The fundamental property: at no point can output records be visible while share acks are uncommitted, or vice versa. They are in the same transaction and share the same COMMIT/ABORT fate.

7. Compatibility, Deprecation, and Migration Plan

Impact on Existing Users

• If older broker doesn’t support TxnShareAcknowledgeRequest, fallback to current explicit/implicit ack with warning.
• Clients must fail fast on unsupported brokers
• No breaking changes for existing Share Group users
• `transactional` mode is opt-in via `share.acknowledgement.mode` config
• Existing `implicit` and `explicit` modes continue to work unchanged

Migration Path

1. Phase 1: Add `transactional` acknowledgement mode (backward compatible)
2. Phase 2: Streaming frameworks (Flink, Spark) implement transactional sources
3. Phase 3: Documentation and best practices for exactly-once semantics

Deprecation

- No deprecation of existing modes planned
- `transactional` mode recommended for exactly-once use cases

8. Test Plan

Unit Tests

- Transaction state machine transitions
- Idempotency of all transaction operations
- Timeout handling and auto-abort
- Record state transitions (ACQUIRED → ACKNOWLEDGED on commit, ACQUIRED → AVAILABLE on abort/timeout)

Integration Tests

- End-to-end transaction commit/abort flows
- Multi-consumer transactions within same group
- Coordinator failure and recovery
- Network partition scenarios

System Tests

- Flink checkpoint integration with transactional Share Groups
- Exactly-once delivery verification under failures
- Performance benchmarks comparing transactional vs non-transactional modes
- Stress testing with concurrent transactions

Chaos Tests

- Broker failure during PREPARE state
- Coordinator crash before/after prepare
- Consumer crash during transaction
- Network partitions between coordinator and brokers

Include exactly‑once recovery tests similar to Flink e2e:

}

Corner cases 

Here are your corner cases formatted using the same clean, un-phased, human-style layout. The wording, titles, technical specifications, and internal error codes remain exactly as provided.

1. Broker Crash During TX_PENDING (in-memory state)

• Scenario: broker stages TX_PENDING then crashes before marker arrives.
• Behavior: TX_PENDING is in-memory only (__share_group_state schema unchanged in v1); records replay as ACQUIRED/AVAILABLE after broker restart and are redelivered.
• Guarantee: no data loss; at-least-once on broker-crash window (existing share-group baseline).
• Future: persistent TX_PENDING via extending ShareSnapshotValue.DeliveryState deferred to a follow-up KIP.

2. TxnCoord Registration Fails (COORDINATOR_NOT_AVAILABLE, NOT_COORDINATOR)

• Scenario: broker-internal AddPartitionsToTxn returns retriable error.
• Behavior: broker does NOT stage; returns error to producer; producer enters abortable state.
• Recovery: user catches KafkaException → abortTransaction() → retries.

3. Producer Fenced Mid-Stage (PRODUCER_FENCED, INVALID_PRODUCER_EPOCH)

• Scenario: zombie producer A (epoch 1) sends TxnShareAcknowledge after producer B (epoch 2) has been initialized.
• Behavior: TxnCoord registration rejects with fencing error → no staging happens → producer A becomes fatal.
• Late-arriving marker safety: per-InFlightState (stagedProducerId, stagedProducerEpoch) filter rejects mismatched markers; idempotent.

5. Duplicate WriteTxnMarkers Delivery

• Scenario: TxnCoord retries marker after network failure; broker receives same marker twice.
• Behavior: second invocation finds state != TX_PENDING (already resolved) → no-op.
• Guarantee: idempotent by state filter — verified in InFlightStateTxnTest.

6. Acquisition Lock Expiry While TX_PENDING

• Scenario: transaction stages records but stalls; lock duration expires before commit/abort.
• Behavior: lock timer detects TX_PENDING with expired lock → reverts to ACQUIRED → re-acquirable; subsequent marker for the stale staging is filtered out (epoch mismatch).
• Configuration: group.share.record.lock.duration.ms (default 30s; per share group).

7. Transaction Timeout While TX_PENDING (transaction.timeout.ms)

• Scenario: producer crashes after staging; TxnCoord detects abandoned transaction via timeout.
• Behavior: TxnCoord auto-sends WriteTxnMarkers(ABORT); broker reverts TX_PENDING → ACQUIRED → AVAILABLE on next lock cycle. No data loss; eventual redelivery.

8. Stale Member Epoch (STALE_MEMBER_EPOCH, UNKNOWN_MEMBER_ID)

• Scenario: rebalance changes memberEpoch between shareGroupMetadata() snapshot and sendShareAcksToTransaction() call.
• Behavior: broker compares request memberEpoch with current SharePartition owner; mismatch → returns STALE_MEMBER_EPOCH; no staging.
• Recovery: abortable; user retries the entire read-process-write loop with a fresh snapshot.

9. Mixed-Version Cluster During Rolling Upgrade

• Scenario: some brokers lack apiKey 93 (TxnShareAcknowledge).
• Behavior: producer probes ApiVersions; if apiKey 93 absent on any broker → UnsupportedVersionException thrown synchronously, no bytes on wire.
• Recovery: user must abort the transaction (output-side AddPartitionsToTxn still needs cleanup).
• Gating: share.version=2 finalized feature flag controls broker advertisement.

10. Invalid Acknowledge Type Inside Transaction

• Scenario: producer sends AcknowledgeType.RELEASE (2), RENEW (4), or GAP (0) inside sendShareAcksToTransaction.
• Behavior: broker rejects with INVALID_RECORD_STATE; no staging.
• Allowed values: only ACCEPT (1) and REJECT (3) — release/renew/gap have no transactional semantics.

11. Partial Multi-Partition Stage Failure

• Scenario: producer stages on partitions A and B; A succeeds, B fails.
• Behavior: per-partition 2PC rollback — already-staged records on A are reverted to ACQUIRED before the response returns (commit 8 on the branch).
• Result: all-or-nothing per call; user sees error → aborts → retries.
• Guarantee: no partial commit; clean rollback semantics.

12. Leader Change for __share_group_state-N Between Stage and Commit

• Scenario: share-coordinator partition leadership moves from broker 5 to broker 7 between staging and WriteTxnMarkers dispatch.
• Behavior: TxnCoord re-resolves leader at commit time via MetadataCache → marker goes to broker 7; broker 7's in-memory partitionCache does NOT contain the original TX_PENDING state.
• Limitation in this intial version: TX_PENDING lost on share-coordinator leadership change (same root cause as Corner Case 1); lock-timeout reverts records; redelivery.
• Future: persistent TX_PENDING closes this case as well.

13. Transaction Aborted Without Markers Reaching Broker

• Scenario: transaction aborts but broker hosting TX_PENDING is partitioned from TxnCoord.
• Behavior: lock timeout (Corner Case 6) eventually reverts; subsequent late-arriving marker is no-op (state already AVAILABLE or epoch mismatched).
• Convergence guaranteed.

14. Concurrent Transactions From Same Producer (Different Epochs)

• Scenario: producer rapidly bumps epoch; old in-flight TxnShareAcknowledge arrives with stale epoch.
• Behavior: TxnCoord rejects with PRODUCER_FENCED; broker never stages. Per-record fencing on the state machine prevents any stale marker from resolving newer staging.

...

Note: We can persist the record state and get rid of case 1 in follow up KIP.

Metrics

All new metrics follow the existing Kafka metric conventions (Yammer for broker JMX, KafkaMetric for client-side).

Metric names mirror the established kafka.server:type=group-coordinator-metrics,name=... and kafka.server:type=share-coordinator-metrics,name=... patterns.

Backward-compatible: no existing metrics renamed or removed.

Compatibility, Deprecation, and Migration Plan

Enablement and Rollout Plan

Feature Gating

opt-in, online-upgradable, and zero-disruption for existing workloads - No new feature flag introduced. Uses existing kafka-features.sh machinery.

Rolling Upgrade (Online; No Traffic Disruption)

Phase 1 — Software upgrade (rolling)
  Upgrade brokers one at a time to the binary containing KIP-1289.
  KIP-1289 code is dormant (share.version still 1).
  Standard Kafka rolling-restart

Phase 2 — Soak timing
  Verify cluster health

Phase 3 — Finalise feature
  kafka-features.sh upgrade --feature share.version --version 2
  - Online metadata propagation (update of  __cluster_metadata)
  - No restart, no rebalance, no leader change, no socket disruption
  - Brokers begin advertising apiKey 93

Phase 4 — Client onboarding
  Roll out applications calling sendShareAcknowledgementsToTransaction.
  Add the alerts (esp. TxnPendingRecordsCount) and relevant metrics in dashboard & observability.

Cross-Version Client/Broker Matrix

Downgrade

1. Drain in-flight TX_PENDING:
   - Halt producers calling sendShareAcknowledgementsToTransaction
   - Wait for transaction.timeout.ms (default 60s) for any abandoned txns to clear
2. kafka-features.sh downgrade --feature share.version --version 1
Broker rejects the downgrade RPC if any in-memory TX_PENDING exists, preventing data inconsistency.
Software downgrade (binary): must follow feature downgrade; standard rolling restart.

Test Plan

Rough implementation [The actual implementation will be phasewise with multiple smaller PRs]:
    https://github.com/apache/kafka/pull/22357

The verification strategy focuses on state machine integrity and fault tolerance under high-concurrency and failure scenarios.

• Unit Tests: Validates state transitions (e.g., ACQUIRED to ACKNOWLEDGED on commit vs. AVAILABLE on abort), idempotency of operations, and transaction timeout/auto-abort logic.

• Integration Tests: Focuses on end-to-end commit/abort flows, coordinator recovery, and multi-consumer behavior within a single group during network partitions.

• System & Performance Tests: Benchmarks transactional vs. non-transactional modes and verifies exactly-once delivery.

• Chaos Tests: Simulates broker and coordinator crashes specifically during critical phases like PREPARE_COMMIT to ensure protocol durability.

Follow-up KIP (deferred)

• Persistent TX_PENDING in __share_group_state for crash-resilient EOS on broker failover during staging (addresses Corner Cases).
• Per-share-partition metric for TX_PENDING residency time— would add a histogram of "time spent in TX_PENDING" useful for diagnosing slow producers; can be added in the persistence KIP without compatibility concerns.
• External processing engines or database as 2PC participants for write/sink records
• Kill after sendShareAcksToTransaction but before commit.
• Kill after commit but before notifyCheckpointComplete.
Ensure acks are either replayed or visible, never lost
• .

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.