Thanks Andrew/dev team for sharing the design note and discussion.
Here is the observation at my end:
1. We must try to decouple the producer from the share consumer to make it work
for distributed stream engine like flink, spark streaming, etc2. Txn Commit
marker will apply commit to all the participants so share consumer need not to
ack separately for CTP usecase.
TL;DR of the google doc:
https://docs.google.com/document/d/1sWMZ1c3j_rwg1jQByZ4rW-66GGoswkRI3DnSUiJPJ94/edit?usp=sharing
```
1. ShareConsumer polls source records.
2. Application decides one final outcome per record:
ACCEPT or REJECT.
3. ShareConsumer.acknowledge(record, decision) stores decisions locally.
4. ShareConsumer creates:
ShareAcknowledgements
-> source topic/partition + offset-range decisions
ShareGroupMetadata
-> group ID + member ID + member epoch
5. Producer already has an open transaction and calls:
sendShareAcknowledgementsToTransaction(
ShareAcknowledgements,
ShareGroupMetadata
)
6. Producer TransactionManager adds its own identity:
transactional ID + producer ID + producer epoch.
7. Producer sends this RPC to each affected source-partition leader:
TxnShareAcknowledge(
group/member metadata,
producer identity,
source topic-partition acknowledgement batches
)
Broker staging flow
8. Source-partition leader validates:
- member ID and member epoch;
- member owns the source offsets;
- offsets are ACQUIRED;
- acknowledgement types are valid.
9. It maps each source share partition to its durable state partition:
orders-2 -> __share_group_state-7
orders-5 -> __share_group_state-3
10. It registers those internal state partitions with the transaction:
AddPartitionsToTxn(
transactional ID,
producer ID / epoch,
__share_group_state-7,
__share_group_state-3
)
11. It writes each source offset range as TX_PENDING in durable
__share_group_state, including producer ID/epoch and ACCEPT/REJECT.
How affected share-state leaders learn about commit
12. Producer calls EndTxn(COMMIT) on its transaction coordinator.
13. Transaction coordinator reads its durable participant list:
output-orders-0
__share_group_state-7
__share_group_state-3
14. Transaction marker sender resolves the current leader for every
participant partition and sends this RPC:
WriteTxnMarkers(
producer ID,
producer epoch,
coordinator epoch,
COMMIT,
participant partitions
)
15. The leaders of __share_group_state-7 and __share_group_state-3
receive their marker request.
16. KafkaApis detects the internal share-state partition and calls:
ShareCoordinatorService.completeTransaction(...)
17. The matching ShareCoordinatorShard finds TX_PENDING batches for
that producer ID/epoch and finalizes them:
ACCEPT -> ACKNOWLEDGED
REJECT -> ARCHIVED
```
Regards,
Shekhar Prasad Rajak,
On Monday 3 August 2026 at 10:30:05 pm GMT+5:30, Shekhar Prasad Rajak
<[email protected]> wrote:
Hi team,
I’ve spent the last few months diving deep into our share group mechanics,
specifically focusing on distributed acknowledgements, fault tolerance, and
recovery protocols. To move us forward, I’ve mapped out a detailed design and
architectural diagrams here:
https://docs.google.com/document/d/1sWMZ1c3j_rwg1jQByZ4rW-66GGoswkRI3DnSUiJPJ94/edit?tab=t.0
Public APIs and High level overview KIP:
https://cwiki.apache.org/confluence/x/J448G
Please review the proposals so we can align on the implementation details.
Next, I will outline the integration plan with Flink to support queue semantics
(as a proof that our KIP changes will going to help stream processing engines
like Flink) where consumer elasticity matters more than throughput, and topics
with severe partition skew.
Regards,Shekhar Rajak
