Page History

...

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 

...