...

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

1 Motivation

A record's side effects (output writes) and its source acknowledgment must be committed atomically. Either both succeed, or neither does.

...

- 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

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

...

• 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

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

1.3 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

2. Use Cases

2.1 Consume-Transform-Produce (CTP)

An application reads from a share group, transforms records, and produces output to another Kafka topic. Both output and acknowledgments must commit atomically.

producer.beginTransaction();

ConsumerRecords<K,V> records = shareConsumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<K,V> record : records) {
    ProducerRecord<K,V> output = transform(record);
    producer.send(output);
}

// Bind share acks to this transaction (NEW API)
producer.sendShareAcksToTransaction(
    ShareAcknowledgements.fromRecords(records, AcknowledgeType.ACCEPT),
    shareConsumer.groupMetadata()
);

producer.commitTransaction();
// Output records AND share acks commit atomically

2.2 Flink / Spark Source (No Producer)

A streaming framework reads from a share group as a source. There is no Kafka producer in the pipeline — the output may go to a database, filesystem, or another system. The framework needs to commit share acks transactionally, coordinated with its own checkpointing.

TransactionalShareAcknowledger acknowledger = new TransactionalShareAcknowledger(props);
acknowledger.initTransactions();

// On checkpoint complete:
acknowledger.commitAcknowledgements(bufferedAcks, shareGroupId);
// Internally: beginTransaction → sendShareAcksToTransaction → commitTransaction

2.3 Flink End-to-End Exactly-Once (Source + Sink)

When a Flink pipeline reads from a Kafka share group and writes to a Kafka sink topic, we achieve end-to-end exactly-once by binding both sink output and source acknowledgments to the same Kafka transaction:

Checkpoint lifecycle:
  prepareCommit()              → flush sink records, pre-commit Kafka txn
                               → include share acks in the same transaction
  snapshotState()              → save txn metadata + buffered acks
  notifyCheckpointComplete()   → commitTransaction() (acks + output atomically)
  On failure                   → abortTransaction() (acks + output both rolled back)

Public Interfaces

...