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.
Current state: Under Discussion
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).
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.
Current State Machine:
AVAILABLE: Ready for delivery.
ACQUIRED: Locked by a consumer.
ACKNOWLEDGED / ARCHIVED: Terminal states; records are never redelivered.
AVAILABLE (Released): Returned to the pool for redelivery.
Current Acknowledgment Modes:
Implicit: Automatic ack on the subsequent poll().
Explicit: Manual ack via acknowledge(record, type) followed by commitSync()
// 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)) |
Zero Breaking Changes: Current implicit and explicit modes remain the default and continue to function as-is.
Opt-in Requirement: Users must explicitly enable the new mode.
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.
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.