...

Transaction States Transitions

- EMPTY -> ONGOING -> PREPARE -> COMMITTED (successexisting Kafka 2PC (TransactionState: ONGOING→PREPARE_COMMIT→COMPLETE_COMMIT)

- EMPTY -> ONGOING -> PREPARE -> ABORTED (abort after prepare)

...

Share acks are written transactionally to __share_group_state using CoordinatorRuntime.scheduleTransactionalWriteOperation().

Failure Handling

existing TransactionCoordinator recovery

3. Wire Protocol Details

TxnShareAcknowledgeRequest

...

mirrors AddOffsetsToTxnRequest; This tells the TransactionCoordinator to add __share_group_state partitions to the producer's transaction.

Consume-Transform-Produce Pattern

This is the exactly-once guarantee: either both output AND acks commit, or neither does: 

```

KafkaShareConsumer<K, V> consumer = new KafkaShareConsumer<>(props);  // NOT transactional
KafkaProducer<K, V> producer = new KafkaProducer<>(props);            // transactional.id

producer.initTransactions();

while (true) {
    ConsumerRecords<K, V> records = consumer.poll(...);  // ← just reads, not transactional
    
    producer.beginTransaction();
    for (var record : records) {
        producer.send(new ProducerRecord<>("output", transform(record)));
    }
    producer.sendShareAcksToTransaction(buildAcks(records), "my-share-group");  //  transactional write
    producer.commitTransaction();
    // Atomically: output records committed + share acks committed
    // If abort: output rolled back + acks discarded → records return to AVAILABLE
}

```

In the existing consumer group CTP pattern, the same thing happens (in the above code instead of KafkaShareConsumer, use KafkaConsumer and Consumer offsets to __consumer_offsets).

If abortTransaction() is called (or the transaction times out):

• Output records get ABORT markers -> consumers with read_committed skip them
• Share acks are discarded -> records stay ACQUIRED -> acquisition lock timeout -> records return to AVAILABLE -> re-delivered to another consumer

Compatibility, Deprecation, and Migration Plan

...