...

- 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

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

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.

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

Public Interfaces

...