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:


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;
}


Wire protocol additions:

Broker / coordinator additions:

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


3. Compatibility, Deprecation, and Migration Plan

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