Page History

...

KafkaProducer.sendShareAcksToTransaction(acks, groupId)

This mirrors the existing KafkaProducer.sendOffsetsToTransaction(offsets, groupMetadata)

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 Configuration

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

...