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

Today, the Producer's 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:

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 (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 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.
FieldTypeNotes
TransactionalIdstring (nullable)The producer's transactional.id.
GroupIdstring, entityType=groupIdThe share group ID.
ProducerIdint64, entityType=producerIdFor fencing.
ProducerEpochint16For fencing.
MemberIdstring, entityType=memberIdThe share group member ID.
MemberEpochint32For share-group fencing.
Topics[]TxnShareAcknowledgeTopicmapKey=true on TopicId.
Topics.TopicIduuid, mapKey=true
Topics.Partitions[]TxnShareAcknowledgePartitionmapKey=true on PartitionIndex.
Partitions.PartitionIndexint32, mapKey=true
Partitions.AcknowledgementBatches[]TxnShareAcknowledgeBatch
Batch.FirstOffsetint64Inclusive.
Batch.LastOffsetint64Inclusive.
Batch.AcknowledgeTypes[]int8Per-offset ack type byte. Size 1 = uniform type for whole range.
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
FieldTypeNotes
ThrottleTimeMsint32
ErrorCodeint16Top-level error.
Responses[]TxnShareAcknowledgeTopicResponse
Responses.TopicIduuid, mapKey=true
Responses.Partitions[]TxnShareAcknowledgePartitionResponse
Partitions.PartitionIndexint32
Partitions.ErrorCodeint16Per-partition error.
Partitions.ErrorMessagestring (nullable)
Partitions.CurrentLeaderLeaderIdAndEpoch (tagged)Populated on NOT_LEADER_OR_FOLLOWER.
NodeEndpoints[]NodeEndpoint (tagged)Top-level: address of any new leader referenced above.


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: schema

Flow — Regular Consumer Group 
Flow — Share Group (THIS KIP)

[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]

State machine additions:


Pseudo code


// Kafka-as-destination path (true EOS via TxnShareAcknowledge):
batch boundary (or per-record, depending on tx granularity):
    producer.beginTransaction()
    for each (record, output) in batch:
        producer.send(destinationTopic, output)
    producer.sendShareAcknowledgementsToTransaction(
        ackMap,                                  // record→ACCEPT for processed records
        shareConsumer.shareGroupMetadata()
    )
    producer.commitTransaction()
    // No separate consumer.acknowledge() — the ACK is in the transaction.

-------------------

// Pure-external-sink path (at-least-once + idempotent destination):[this KIP is not changin anything of this flow]
worker thread per record:
    output = process(record)
    externalSink.write(output, idempotencyKey)        // user task does this
    consumer.acknowledgeAsync(record, ACCEPT)
        .thenAccept(_ -> markComplete(record))


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)

2. TxnCoord Registration Fails (COORDINATOR_NOT_AVAILABLE, NOT_COORDINATOR)

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

5. Duplicate WriteTxnMarkers Delivery

6. Acquisition Lock Expiry While TX_PENDING

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

8. Stale Member Epoch (STALE_MEMBER_EPOCH, UNKNOWN_MEMBER_ID)

9. Mixed-Version Cluster During Rolling Upgrade

10. Invalid Acknowledge Type Inside Transaction

11. Partial Multi-Partition Stage Failure

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

13. Transaction Aborted Without Markers Reaching Broker

14. Concurrent Transactions From Same Producer (Different Epochs)

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

 Compatibility, Deprecation, and Migration Plan

Test Plan

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


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.