...

• 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

...

Config                           Default      Description
share.acknowledgement.mode       explicit     Values: implicit, explicit, transactional
share.transaction.timeout.ms     60000        Maximum transaction duration before auto-abort
share.transaction.max.pending    5            Maximum concurrent transactions per group

New Metrics

Metric                                Type        Description
share-transaction-active              Gauge       Active transactions
share-transaction-prepare-time-ms     Histogram   Prepare latency
share-transaction-commit-time-ms      Histogram   Commit latency
share-transaction-abort-total         Counter     Aborted transactions
share-transaction-timeout-total       Counter     Timed-out transactions

...

Handled by existing WriteTxnMarkersRequest (inter-broker) and  EndTxnRequest(ABORT)

AddShareAcksToTxnRequest

• mirrors AddOffsetsToTxnRequest; This tells the TransactionCoordinator to add __share_group_state partitions to the producer's transaction.
• AddShareAcksToTxn computes TopicPartition(SHARE_GROUP_STATE_TOPIC_NAME, shareCoordinator.partitionFor(shareGroupId, topicPartition)) and  adds it to the producer's TransactionMetadata.topicPartitions set via TransactionCoordinator.handleAddPartitionsToTransaction().
• This is what causes WriteTxnMarkers to dispatch markers to __share_group_state partitions during commit.

```

AddShareAcksToTxnRequest => TransactionalId ProducerId ProducerEpoch GroupId [Topics]

  TransactionalId => STRING

  ProducerId => INT64

  ProducerEpoch => INT16

  GroupId => STRING

  Topics => TopicName [Partitions]

    Partitions => INT32

```

Consume-Transform-Produce Pattern

...

• 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

no-producer (Flink/Spark) use case 

```

// Flink ShareGroupSource connector — on checkpoint complete

class FlinkShareGroupSourceReader implements SourceReader<...> {

    private TransactionalShareAcknowledger acknowledger;

    private List<ShareAcknowledgement> pendingAcks;

    void open() {

        acknowledger = new TransactionalShareAcknowledger(config);

        acknowledger.initTransactions();

    }

    void snapshotState(long checkpointId) {

        // Save pending acks to checkpoint state

    }

    void notifyCheckpointComplete(long checkpointId) {

        // Atomically commit acks for this checkpoint

        acknowledger.commitAcknowledgements(pendingAcks, shareGroupId);

    }

}

```

Compatibility, Deprecation, and Migration Plan

...

- Transaction state machine transitions
- Idempotency of all transaction operations
- Timeout handling and auto-abort
- Record state transitions (ACQUIRED → LOCKED → ACKNOWLEDGEDACKNOWLEDGED on commit, ACQUIRED → AVAILABLE on abort/timeout)

Integration Tests

- End-to-end transaction commit/abort flows
- Multi-consumer transactions within same group
- Coordinator failure and recovery
- Network partition scenarios

...