...

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

Motivation

Ensure atomicity between share‑group acknowledgments and downstream A record's side effects (e.g., Kafka transactions) so a record is either both processed and acknowledged or neitheroutput 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()

...

- 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

Public Interfaces

...