...

• 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)

3 Public Interfaces

APIs 

• KafkaProducer.sendShareAcksToTransaction(acks, groupId) - for CTP 
  • This mirrors the existing KafkaProducer.sendOffsetsToTransaction(offsets, groupMetadata)

```

// When you have a producer and want atomic output + acks

producer.beginTransaction();

producer.send(output);

producer.sendShareAcksToTransaction(acks, groupId);

producer.commitTransaction();

```

• TransactionalShareAcknowledger - for standalone ack transactions

```

// When there's no producer (Flink source, Spark source, pure consumer)

public class TransactionalShareAcknowledger implements Closeable {

    public TransactionalShareAcknowledger(Map<String, Object> config);

    public void initTransactions();

    public void commitAcknowledgements(

        Map<TopicPartition, List<ShareAcknowledgement>> acks, String groupId);

    public void abortAcknowledgements();

    public void close();

}

```

Internally, TransactionalShareAcknowledger is a thin wrapper around a KafkaProducer (or its TransactionManager). It use the exact same RPCs - InitProducerId, AddShareAcksToTxn, TxnShareAcknowledge, EndTxn.

No new server-side infrastructure needed.

```

// TransactionalShareAcknowledger — internally just wraps a KafkaProducer 
TransactionalShareAcknowledger acknowledger = 
    new TransactionalShareAcknowledger(config);  // config has transactional.id
acknowledger.initTransactions();
acknowledger.commitAcknowledgements(acks, shareGroupId);  // begin+ack+commit in one call

```

Coordinator API

• Reuses the existing TransactionCoordinator
• new server-side component is a completeTransaction() method on ShareCoordinator, mirroring GroupCoordinator.completeTransaction().
• WriteTxnMarkers as the mechanism for the transaction coordinator to tell the group coordinator to complete transactional operations on __consumer_offsets
  • For consumer group we have => groupCoordinator.completeTransaction(partition, ...)
  • Similarly implement shareCoordinator.completeTransaction(partition, ...) for share group 

New Metrics

...

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

 }

3.5 New Metrics

...

...

...

...

Proposed Changes

1. Two-Phase Commit Protocol

...