...

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

1 Motivation

In the current KIP-932 implementation, Share Group acknowledgments are immediate and irrevocable.

If a processing framework (Flink, Spark, etc.) crashes after acknowledging a record but before committing its own state, that record is lost.

• Current State: ACKNOWLEDGED is a terminal state; there is no way to revert to AVAILABLE for redelivery.

• The Risk: Permanent data loss during processing failures.

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.

1.1 Background: Share Groups

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:

1. Implicit: Automatic ack on the subsequent poll().

2. Explicit: Manual ack via acknowledge(record, type) followed by commitSync()

1.2 Why Existing Consumer-Group Exactly-Once Doesn't Apply

Traditional Consumer Groups rely on replayability, which Share Groups lack:

• Consumer Groups: Frameworks like Flink save offsets in their own state. On failure, they use seek(offset) to replay data. In this model, Kafka offset commits are "cosmetic" (non-critical) because Flink is the source of truth.

• Share Groups: There is no seek() functionality. The broker manages delivery; once a record is acknowledged, it is removed from the delivery pipeline.

• The Conflict: Because records cannot be replayed, the acknowledgment itself must be the transactional boundary. It must stay "pending" until the entire processing transaction is confirmed.

1.3 Existing Pattern: sendOffsetsToTransaction()

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)

3 Public Interfaces

3.1 KafkaProducer API Addition

// In org.apache.kafka.clients.producer.KafkaProducer:

/**
 * Sends share-group acknowledgments as part of the current transaction.
 * Mirrors sendOffsetsToTransaction() but writes to __share_group_state
 * instead of __consumer_offsets.
 *
 * @param acknowledgements Map of TopicPartition to list of acknowledgment batches
 * @param groupMetadata    The share group metadata (group ID, member ID, generation)
 * @throws IllegalStateException if no transaction is in progress
 * @throws ProducerFencedException if the producer is fenced
 */
public void sendShareAcksToTransaction(
    Map<TopicPartition, ShareAcknowledgements> acknowledgements,
    ShareGroupMetadata groupMetadata
) throws ProducerFencedException;

This mirrors sendOffsetsToTransaction(). The reason a new method is needed (instead of reusing the existing one) is that:

• Different storage topic: acks go to __share_group_state, not __consumer_offsets.
• Different coordinator: the ShareCoordinator handles ack persistence, not GroupCoordinator.
• Different semantics: acks are per-record state transitions, not per-partition offsets.

3.2 TransactionalShareAcknowledger (Standalone)

For frameworks that do not use a KafkaProducer in the pipeline:

public class TransactionalShareAcknowledger implements Closeable {

    public TransactionalShareAcknowledger(Properties config);

    /** Initialize the internal transactional producer. Call once. */
    public void initTransactions();

    /**
     * Atomically commit share acknowledgments.
     * Internally executes: beginTransaction → sendShareAcksToTransaction → commitTransaction.
     * This is NOT a single RPC. It orchestrates the standard 2PC protocol.
     */
    public void commitAcknowledgements(
        Map<TopicPartition, ShareAcknowledgements> acks,
        String groupId
    );

    /** Abort any in-progress transactional acknowledgment. */
    public void abortAcknowledgements();

    public void close();

 }

Clarification: commitAcknowledgements() is a convenience wrapper. It internally calls three operations in sequence:

1. beginTransaction()
2. sendShareAcksToTransaction(acks, groupMetadata)
3. commitTransaction()

It does NOT introduce a new single-RPC path. It uses the standard 2PC protocol.

3.3 ShareGroupMetadata

public class ShareGroupMetadata {
    private final String groupId;
    private final String memberId;
    private final int generationId;
    // constructor, getters

 }

3.4 ShareAcknowledgements

public class ShareAcknowledgements {
    private final List<ShareAcknowledgementBatch> batches;

    public static ShareAcknowledgements fromRecords(
        ConsumerRecords<?, ?> records, AcknowledgeType type);

    // Each batch: firstOffset, lastOffset, acknowledgeType

...

Ensures that output records and source acknowledgments are committed as a single atomic unit within a Kafka-to-Kafka pipeline.

• The Flow: beginTransaction() → send(output) → sendShareAcksToTransaction() → commitTransaction().

• Result: Output is visible and source records are finalized only if both operations succeed.

2.2 Source-Only Frameworks (No Kafka Producer)

For applications writing to external systems (Databases, S3) that require transactional acknowledgments coordinated with their own internal checkpoints.

• Tool: KIP-1302 General Transaction Session

• The Flow: Aggregates acks and commits them via a background transaction once the external sink confirms the write.

2.3 End-to-End Exactly-Once (Flink/Spark/OLAP)

Integrates Share Groups into the two-phase commit (2PC) lifecycle of streaming engines.

• Pre-commit: Sink records are flushed and share acks are added to the transaction.

• Snapshot: Transactional metadata is saved to the framework state.

• Commit: On checkpoint completion, the transaction is finalized.

• Recovery: If the framework fails, the transaction is aborted; output is rolled back, and the broker automatically reverts share records to AVAILABLE for redelivery.

3 Public Interfaces

A new method is added to KafkaProducer to mirror traditional offset commits:

• Method: sendShareAcksToTransaction(Map<TopicPartition, ShareAcknowledgements>, ShareGroupMetadata)

• Why a new method? Unlike offsets, share acks target __share_group_state (not __consumer_offsets) and are managed by the ShareCoordinator (not the GroupCoordinator).

KIP-1310: General Transaction Session#4.5Custom2PCCoordinators

// Phase 1: Prepare
kafkaSession.beginTransaction();
producer.send(records);
database.prepareTransaction(dbTxnId); 
kafkaSession.prepareTransaction();    

// Phase 2: Commit (Recovery-friendly)
TransactionSession resumed = TransactionSession.resume(txnId, pid, epoch, configs);
resumed.commitTransaction();          
database.commitTransaction(dbTxnId);

3.5 New Metrics

...