You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by Shekhar Prasad Rajak via dev <de...@kafka.apache.org> on 2026/03/01 17:39:38 UTC
Re: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups
Thank you Andrew for the review.
It took sometime to analyse the review comments and understand Kafka Producer 2PC implementation in depth.All the points are strongly valid and I have updated the KIP to adopt the RPC already exists and leverage the same methods for , CTP pattern (Consume-Transform-Produce) for kafka as sink or no producer usecase.
Please have a look: https://cwiki.apache.org/confluence/display/KAFKA/KIP-1289+Support+Hierarchical+Transactional+Acknowledgments+for+Share+Groups
Regards,Shekhar Prasad Rajak,
On Thursday 26 February 2026 at 03:32:36 am GMT+5:30, Andrew Schofield <sc...@apache.org> wrote:
Hi Shekhar,
Thanks for the KIP. Adding transactional support for acknowledgements is one of the future features I'd like to see for share groups, so I'm glad you've started this.
Here are some comments from an initial read.
AS1: Previously, Kafka transactions have always been a producer concept. You set a configuration called `transactional.id` in the producer configuration which gives authorization checking and helps with fencing. Then you use KafkaProducer methods like beginTransaction and commitTransaction to mark the start and end of the transaction. Only the producer is currently enabled for transactional operation, and the Kafka RPCs it uses include transactional ID as necessary to indicate operations which are transactional.
There will be differences for sure with share groups, but the overall model should probably still hold. Over the years, we've gradually tightened up the transaction protocol by adding features such as the producer epoch and two-phase commit. I would rather not re-invent everything. For example, WriteTxnMarkers is used by the transaction coordinator to tell the group coordinator to complete transactional operations on the __consumer_offsets topic. I would expect a similar kind of interaction with the share coordinator on the __share_group_state topic.
I suggest looking at how to adapt the existing Kafka protocol RPCs rather than defining new ones.
AS2: Why is there an acknowledge(ConsumerRecord<,>, AcknowledgeType, String) method? I would have expected that the transactional ID is a configuration, not a parameter of this method.
AS3: Have you considered how to write an application which uses a share group and producer, and coordinates their operations in the same transaction?
Thanks,
Andrew
On 2026/02/24 08:54:36 Shekhar Prasad Rajak via dev wrote:
> Hello team,
> In this discussion thread we will finalise the design for the KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups.
>
> Share Groups currently support only immediate acknowledgement modes (IMPLICIT/EXPLICIT). This causes data loss in distributed streaming frameworks:
>
> 1. Worker acknowledges records → Records removed from Kafka
> 2. Checkpoint fails before sink write
> 3. Records lost (acknowledged but never persisted)
>
> Goal: Enable exactly-once read semantics via transactional acknowledgements.
> We already have similar pattern in Kafka producer transaction
>
> KIP: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups - Apache Kafka - Apache Software Foundation
>
> Looking forward to community's feedback!
>
>
> Regards,Shekhar Prasad Rajak
>
>
Re: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups
Posted by Andrew Schofield <sc...@apache.org>.
Hi,
Sorry for the delay in responding.
I was interested to see how you attacked the problem of acknowledging transactionally. The direction is definitely along the right lines.
The tricky part of this is that the KafkaProducer is transactional but KafkaShareConsumer is not. For regular consumers, there is the similar distinction: KafkaProducer is transactional but KafkaConsumer is not. So, there are methods in KafkaProducer to add offsets to transactions, but the KafkaConsumer is entirely unaware of the use of transactions. KafkaConsumer does not keep much state beyond the current position of its assigned partitions, and the correct behaviour for handling the consumer when a transaction fails is to throw it away and make a new one. Even so, we find that developers who write transactional applications which use KafkaProducer and KafkaConsumer together don't always get the error handling right.
KafkaShareConsumer maintains a lot more state such as the acknowledgements map and updates to this need to be coordinated as part of the transaction. Also, the acknowledgements will require both the group ID and member ID to be sent to the broker.
I think there are two broad ways to achieve this:
a) We could make KafkaShareConsumer transactional in its own right. This is probably a bit controversial in terms of Kafka APIs. The code might look like this:
shareConsumer.beginTransaction()
shareConsumer.acknowledge(inputRecord)
shareConsumer.reply("outputTopic", outputRecord)
shareConsumer.commitTransaction()
It's basically introduced a very simple producer capability into the share consumer, and also made the share consumer transactional. Of course, in the no-producer case you described, the controversial reply() method would not be required.
b) We could somehow get KafkaProducer to call KafkaShareConsumer at transaction end in order to update the state correctly.
I think what you've done in the sequence diagrams is along the right lines. I think we need the KafkaShareConsumer to send the AddShareAcksToTxn to the share-partition leader on the instructions of the KafkaProducer, or at least assist in the building of it. Otherwise, the record states in the share-partition leader will not be updated and the record acquisition locks could time out.
Maybe the code for your CTP use case would look like the following:
producer.beginTransaction()
producer.send(outputRecord)
producer.sendShareAcknowledgementsToTransaction(shareConsumer)
producer.commitTransaction()
Or perhaps:
producer.beginTransaction()
producer.send(outputRecord)
txnAckData = shareConsumer.prepareAcknowledgementsForTransaction()
producer.sendShareAcknowledgementsToTransaction(txnAckData)
producer.commitTransaction()
The sendShareAcknowledgementsToTransaction method calls a method on the share consumer providing the transactional ID, producer ID and producer epoch which will be needed to participate in the Kafka transaction protocol. The share consumer would then call the share-partition leader, move the acknowledged records into a committing state, and then the SPL would call the share coordinator with a transactional variant of WriteShareGroupState.
Then the transaction coordinator would call the share coordinator during transaction commit to commit the transactional updates to the share-group state.
This is extending the transactional capabilities of Kafka quite considerably. I think the best path here is to flesh out the second of these options further. Let me know whether you'd like me to help with the next revision of the KIP.
Thanks,
Andrew
On 2026/03/01 17:39:38 Shekhar Prasad Rajak via dev wrote:
> Thank you Andrew for the review.
> It took sometime to analyse the review comments and understand Kafka Producer 2PC implementation in depth.All the points are strongly valid and I have updated the KIP to adopt the RPC already exists and leverage the same methods for , CTP pattern (Consume-Transform-Produce) for kafka as sink or no producer usecase.
> Please have a look: https://cwiki.apache.org/confluence/display/KAFKA/KIP-1289+Support+Hierarchical+Transactional+Acknowledgments+for+Share+Groups
>
>
>
> Regards,Shekhar Prasad Rajak,
>
>
>
>
>
>
> On Thursday 26 February 2026 at 03:32:36 am GMT+5:30, Andrew Schofield <sc...@apache.org> wrote:
>
> Hi Shekhar,
> Thanks for the KIP. Adding transactional support for acknowledgements is one of the future features I'd like to see for share groups, so I'm glad you've started this.
>
> Here are some comments from an initial read.
>
> AS1: Previously, Kafka transactions have always been a producer concept. You set a configuration called `transactional.id` in the producer configuration which gives authorization checking and helps with fencing. Then you use KafkaProducer methods like beginTransaction and commitTransaction to mark the start and end of the transaction. Only the producer is currently enabled for transactional operation, and the Kafka RPCs it uses include transactional ID as necessary to indicate operations which are transactional.
>
> There will be differences for sure with share groups, but the overall model should probably still hold. Over the years, we've gradually tightened up the transaction protocol by adding features such as the producer epoch and two-phase commit. I would rather not re-invent everything. For example, WriteTxnMarkers is used by the transaction coordinator to tell the group coordinator to complete transactional operations on the __consumer_offsets topic. I would expect a similar kind of interaction with the share coordinator on the __share_group_state topic.
>
> I suggest looking at how to adapt the existing Kafka protocol RPCs rather than defining new ones.
>
> AS2: Why is there an acknowledge(ConsumerRecord<,>, AcknowledgeType, String) method? I would have expected that the transactional ID is a configuration, not a parameter of this method.
>
> AS3: Have you considered how to write an application which uses a share group and producer, and coordinates their operations in the same transaction?
>
>
> Thanks,
> Andrew
>
> On 2026/02/24 08:54:36 Shekhar Prasad Rajak via dev wrote:
> > Hello team,
> > In this discussion thread we will finalise the design for the KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups.
> >
> > Share Groups currently support only immediate acknowledgement modes (IMPLICIT/EXPLICIT). This causes data loss in distributed streaming frameworks:
> >
> > 1. Worker acknowledges records → Records removed from Kafka
> > 2. Checkpoint fails before sink write
> > 3. Records lost (acknowledged but never persisted)
> >
> > Goal: Enable exactly-once read semantics via transactional acknowledgements.
> > We already have similar pattern in Kafka producer transaction
> >
> > KIP: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups - Apache Kafka - Apache Software Foundation
> >
> > Looking forward to community's feedback!
> >
> >
> > Regards,Shekhar Prasad Rajak
> >
> >
>
Re: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups
Posted by Andrew Schofield <sc...@apache.org>.
Hi Shekhar,
I will concentrate on reviewing KIP-1310 in preference to this one. I see what you're trying to do and it makes sense.
Thanks,
Andrew
On 2026/04/01 19:05:21 Shekhar Prasad Rajak via dev wrote:
> Hi team,
> To support works around transactions - I believe we need General Transaction Session - Please have a look & review : https://cwiki.apache.org/confluence/display/KAFKA/KIP-1310%3A+General+Transaction+Session
>
>
> Regards,
> Shekhar
>
>
>
>
>
> On Saturday 28 March 2026 at 01:30:05 pm GMT+5:30, Shekhar Prasad Rajak via dev <de...@kafka.apache.org> wrote:
>
>
> Hi Andrew,
>
> Thank you for the detailed breakdown, especially the distinction between making the ShareConsumer transactional (Option A) vs. the Producer-assisted model (Option B).
>
> I agree that Option B is the superior path for architectural consistency. By having the Producer "assist" the ShareConsumer, we can leverage the existing mechanisms.
>
> I am looking at implementing a prepareAcknowledgementsForTransaction() method on the KafkaShareConsumer. This would return a ShareGroupTxnContext containing the MemberID and the specific record acquisitions. The KafkaProducer would then use this context to trigger the AddShareAcksToTxn RPC.
> Regarding your points on KIP-1191 discussion thread, this Option B approach also allows us to solve the "Atomic DLQ" problem. If we include the DLQ write in the Producer's transaction, the Share Coordinator can wait for the WriteTxnMarker before finalising the move to the ARCHIVED state. This ensures the 1:1 audit trail remains intact even during broker failures.
> I have updated the KIP and will keep improving/rephrasing it to clarify implementation: https://cwiki.apache.org/confluence/x/J448G
> I am also starting to explore the broader challenge of Atomicity for cross-cluster scenarios. While standard transactions are intra-cluster, I am looking into "Acknowledgment Tokens" that could be recorded on a target cluster and mirrored back to the source coordinator to finalize state. I’d be interested to hear if the community sees value in laying the groundwork for that in this KIP.
>
>
>
> Regards,Shekhar Prasad Rajak,
>
>
>
>
>
>
> On Sunday 1 March 2026 at 11:09:59 pm GMT+5:30, Shekhar Prasad Rajak via dev <de...@kafka.apache.org> wrote:
>
> Thank you Andrew for the review.
> It took sometime to analyse the review comments and understand Kafka Producer 2PC implementation in depth.All the points are strongly valid and I have updated the KIP to adopt the RPC already exists and leverage the same methods for , CTP pattern (Consume-Transform-Produce) for kafka as sink or no producer usecase.
> Please have a look: https://cwiki.apache.org/confluence/display/KAFKA/KIP-1289+Support+Hierarchical+Transactional+Acknowledgments+for+Share+Groups
>
>
>
> Regards,Shekhar Prasad Rajak,
>
>
>
>
>
>
> On Thursday 26 February 2026 at 03:32:36 am GMT+5:30, Andrew Schofield <sc...@apache.org> wrote:
>
> Hi Shekhar,
> Thanks for the KIP. Adding transactional support for acknowledgements is one of the future features I'd like to see for share groups, so I'm glad you've started this.
>
> Here are some comments from an initial read.
>
> AS1: Previously, Kafka transactions have always been a producer concept. You set a configuration called `transactional.id` in the producer configuration which gives authorization checking and helps with fencing. Then you use KafkaProducer methods like beginTransaction and commitTransaction to mark the start and end of the transaction. Only the producer is currently enabled for transactional operation, and the Kafka RPCs it uses include transactional ID as necessary to indicate operations which are transactional.
>
> There will be differences for sure with share groups, but the overall model should probably still hold. Over the years, we've gradually tightened up the transaction protocol by adding features such as the producer epoch and two-phase commit. I would rather not re-invent everything. For example, WriteTxnMarkers is used by the transaction coordinator to tell the group coordinator to complete transactional operations on the __consumer_offsets topic. I would expect a similar kind of interaction with the share coordinator on the __share_group_state topic.
>
> I suggest looking at how to adapt the existing Kafka protocol RPCs rather than defining new ones.
>
> AS2: Why is there an acknowledge(ConsumerRecord<,>, AcknowledgeType, String) method? I would have expected that the transactional ID is a configuration, not a parameter of this method.
>
> AS3: Have you considered how to write an application which uses a share group and producer, and coordinates their operations in the same transaction?
>
>
> Thanks,
> Andrew
>
> On 2026/02/24 08:54:36 Shekhar Prasad Rajak via dev wrote:
> > Hello team,
> > In this discussion thread we will finalise the design for the KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups.
> >
> > Share Groups currently support only immediate acknowledgement modes (IMPLICIT/EXPLICIT). This causes data loss in distributed streaming frameworks:
> >
> > 1. Worker acknowledges records → Records removed from Kafka
> > 2. Checkpoint fails before sink write
> > 3. Records lost (acknowledged but never persisted)
> >
> > Goal: Enable exactly-once read semantics via transactional acknowledgements.
> > We already have similar pattern in Kafka producer transaction
> >
> > KIP: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups - Apache Kafka - Apache Software Foundation
> >
> > Looking forward to community's feedback!
> >
> >
> > Regards,Shekhar Prasad Rajak
> >
> >
>
Re: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups
Posted by Andrew Schofield <sc...@apache.org>.
Hi Shekhar Rajak,
I think that the overall direction looks pretty good. Some of the details about how the Kafka protocol would evolve are missing, so I've written a wiki page with my initial thoughts (https://cwiki.apache.org/confluence/spaces/KAFKA/pages/446071036/Design+notes). I also tried to simplify the application programming interface.
Let me know what you think.
Thanks,
Andrew
On 2026/08/04 13:26:37 Andrew Schofield wrote:
> Hi,
> Thanks for the comprehensive updates to the KIP and the supporting document. This is a great help for me to review the proposal. I will spend some time this week reviewing it in depth and providing feedback.
>
> Thanks,
> Andrew
>
> On 2026/08/03 17:00:05 Shekhar Prasad Rajak via dev 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
>
Re: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups
Posted by Andrew Schofield <sc...@apache.org>.
Hi,
Thanks for the comprehensive updates to the KIP and the supporting document. This is a great help for me to review the proposal. I will spend some time this week reviewing it in depth and providing feedback.
Thanks,
Andrew
On 2026/08/03 17:00:05 Shekhar Prasad Rajak via dev 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
Re: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups
Posted by Shekhar Prasad Rajak via dev <de...@kafka.apache.org>.
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 <sh...@ymail.com> 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
Re: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups
Posted by Shekhar Prasad Rajak via dev <de...@kafka.apache.org>.
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
Re: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups
Posted by Shekhar Prasad Rajak via dev <de...@kafka.apache.org>.
Hi Andrew, and the dev@ community,
Thank you for the thorough, constructive review.
Summary of what's been added or revised:
1. RPC Schemas — Section 3 (Public Interfaces > Wire Protocol)
2. Internal Topic Record Schemas — Section 3.2
- __transaction_state: schema unchanged
- __share_group_state: schema unchanged for initial implementation
code-grounded reasoning for the deferral
3. Sequence Diagrams — Section 3.3
4. Java Interface Changes — Section 3.4
5. Metrics — Section 5
6. Enablement and Rollout Plan — Section 6
- Reliability and availability guarantees during rollout
7. Corner Cases — Section 4
Read-process-write atomicity for Kafka destinations is the missing piece that
unlocks share-group adoption in any production EoS. This KIP is a foundational step toward that capability and toward proper integration with
external EoS-aware systems (sinks, connectors, stream processors). So let's expedite the execution of the idea.
I believe the KIP is now at good shape. I'd be grateful for a fresh pass on the wiki at your convenience.
Link to the wiki:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-1289+Support+Transactional+Acknowledgments+for+Share+Groups
Link to the reference rough implementation [The actual implementation will be phasewise with multiple smaller PRs]:
https://github.com/apache/kafka/pull/22357
Thanks,
Shekhar Rajak
From voting thread discussion
Hi,
Thanks very much for this KIP. I think it has promise and plan to spend the time required in the discussion. However, it seems to me that it's premature to start voting.
EoS support in Kafka is a very complicated area. I know how to make share groups and transactions work together. While this KIP is heading in the right direction, it is lacking a lot of vital details and there has not yet been proper engagement from the community to validate it. Personally, I have not had the bandwidth to get properly involved, especially given the rapidly approaching deadline for AK 4.4.
I would expect to see a specification for the work in the KIP including:
* RPC schemas for new and changed Kafka protocol APIs
* Schemas for any changes to the records on the internal topics
* Sequence diagrams for the RPC flows which explain all of the interactions, including recovery paths
* Java interface changes
* Metrics
* Enablement and rollout plan
KIP-932 is a huge document because it contains all of these things. This KIP needs to be a similar level of detail. It's going to be quite a large KIP.
I can help progress it, but only when I actually have significant time to dedicate to it.
Thanks,
Andrew
On Sunday 10 May 2026 at 12:51:16 pm GMT+5:30, Shekhar Prasad Rajak via dev <de...@kafka.apache.org> wrote:
Hi team,
Reducing the scope for the "KIP-1289 Support Transactional Acknowledgments for Share Groups" and updated the KIP document for "Kafka producer write AND consumes from a share group".
Lets start with the CTP pattern and later extend it for external systems like DB, Object store write.
Please review the updated KIP : KIP-1289 Support Transactional Acknowledgments for Share Groups - Apache Kafka - Apache Software Foundation
Thanks,
Shekharrajak
On Thursday 2 April 2026 at 12:35:43 am GMT+5:30, Shekhar Prasad Rajak via dev <de...@kafka.apache.org> wrote:
Hi team,
To support works around transactions - I believe we need General Transaction Session - Please have a look & review : https://cwiki.apache.org/confluence/display/KAFKA/KIP-1310%3A+General+Transaction+Session
Regards,
Shekhar
On Saturday 28 March 2026 at 01:30:05 pm GMT+5:30, Shekhar Prasad Rajak via dev <de...@kafka.apache.org> wrote:
Hi Andrew,
Thank you for the detailed breakdown, especially the distinction between making the ShareConsumer transactional (Option A) vs. the Producer-assisted model (Option B).
I agree that Option B is the superior path for architectural consistency. By having the Producer "assist" the ShareConsumer, we can leverage the existing mechanisms.
I am looking at implementing a prepareAcknowledgementsForTransaction() method on the KafkaShareConsumer. This would return a ShareGroupTxnContext containing the MemberID and the specific record acquisitions. The KafkaProducer would then use this context to trigger the AddShareAcksToTxn RPC.
Regarding your points on KIP-1191 discussion thread, this Option B approach also allows us to solve the "Atomic DLQ" problem. If we include the DLQ write in the Producer's transaction, the Share Coordinator can wait for the WriteTxnMarker before finalising the move to the ARCHIVED state. This ensures the 1:1 audit trail remains intact even during broker failures.
I have updated the KIP and will keep improving/rephrasing it to clarify implementation: https://cwiki.apache.org/confluence/x/J448G
I am also starting to explore the broader challenge of Atomicity for cross-cluster scenarios. While standard transactions are intra-cluster, I am looking into "Acknowledgment Tokens" that could be recorded on a target cluster and mirrored back to the source coordinator to finalize state. I’d be interested to hear if the community sees value in laying the groundwork for that in this KIP.
Regards,Shekhar Prasad Rajak,
On Sunday 1 March 2026 at 11:09:59 pm GMT+5:30, Shekhar Prasad Rajak via dev <de...@kafka.apache.org> wrote:
Thank you Andrew for the review.
It took sometime to analyse the review comments and understand Kafka Producer 2PC implementation in depth.All the points are strongly valid and I have updated the KIP to adopt the RPC already exists and leverage the same methods for , CTP pattern (Consume-Transform-Produce) for kafka as sink or no producer usecase.
Please have a look: https://cwiki.apache.org/confluence/display/KAFKA/KIP-1289+Support+Hierarchical+Transactional+Acknowledgments+for+Share+Groups
Regards,Shekhar Prasad Rajak,
On Thursday 26 February 2026 at 03:32:36 am GMT+5:30, Andrew Schofield <sc...@apache.org> wrote:
Hi Shekhar,
Thanks for the KIP. Adding transactional support for acknowledgements is one of the future features I'd like to see for share groups, so I'm glad you've started this.
Here are some comments from an initial read.
AS1: Previously, Kafka transactions have always been a producer concept. You set a configuration called `transactional.id` in the producer configuration which gives authorization checking and helps with fencing. Then you use KafkaProducer methods like beginTransaction and commitTransaction to mark the start and end of the transaction. Only the producer is currently enabled for transactional operation, and the Kafka RPCs it uses include transactional ID as necessary to indicate operations which are transactional.
There will be differences for sure with share groups, but the overall model should probably still hold. Over the years, we've gradually tightened up the transaction protocol by adding features such as the producer epoch and two-phase commit. I would rather not re-invent everything. For example, WriteTxnMarkers is used by the transaction coordinator to tell the group coordinator to complete transactional operations on the __consumer_offsets topic. I would expect a similar kind of interaction with the share coordinator on the __share_group_state topic.
I suggest looking at how to adapt the existing Kafka protocol RPCs rather than defining new ones.
AS2: Why is there an acknowledge(ConsumerRecord<,>, AcknowledgeType, String) method? I would have expected that the transactional ID is a configuration, not a parameter of this method.
AS3: Have you considered how to write an application which uses a share group and producer, and coordinates their operations in the same transaction?
Thanks,
Andrew
On 2026/02/24 08:54:36 Shekhar Prasad Rajak via dev wrote:
> Hello team,
> In this discussion thread we will finalise the design for the KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups.
>
> Share Groups currently support only immediate acknowledgement modes (IMPLICIT/EXPLICIT). This causes data loss in distributed streaming frameworks:
>
> 1. Worker acknowledges records → Records removed from Kafka
> 2. Checkpoint fails before sink write
> 3. Records lost (acknowledged but never persisted)
>
> Goal: Enable exactly-once read semantics via transactional acknowledgements.
> We already have similar pattern in Kafka producer transaction
>
> KIP: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups - Apache Kafka - Apache Software Foundation
>
> Looking forward to community's feedback!
>
>
> Regards,Shekhar Prasad Rajak
>
>
Re: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups
Posted by Shekhar Prasad Rajak via dev <de...@kafka.apache.org>.
Hi team,
Reducing the scope for the "KIP-1289 Support Transactional Acknowledgments for Share Groups" and updated the KIP document for "Kafka producer write AND consumes from a share group".
Lets start with the CTP pattern and later extend it for external systems like DB, Object store write.
Please review the updated KIP : KIP-1289 Support Transactional Acknowledgments for Share Groups - Apache Kafka - Apache Software Foundation
Thanks,
Shekharrajak
On Thursday 2 April 2026 at 12:35:43 am GMT+5:30, Shekhar Prasad Rajak via dev <de...@kafka.apache.org> wrote:
Hi team,
To support works around transactions - I believe we need General Transaction Session - Please have a look & review : https://cwiki.apache.org/confluence/display/KAFKA/KIP-1310%3A+General+Transaction+Session
Regards,
Shekhar
On Saturday 28 March 2026 at 01:30:05 pm GMT+5:30, Shekhar Prasad Rajak via dev <de...@kafka.apache.org> wrote:
Hi Andrew,
Thank you for the detailed breakdown, especially the distinction between making the ShareConsumer transactional (Option A) vs. the Producer-assisted model (Option B).
I agree that Option B is the superior path for architectural consistency. By having the Producer "assist" the ShareConsumer, we can leverage the existing mechanisms.
I am looking at implementing a prepareAcknowledgementsForTransaction() method on the KafkaShareConsumer. This would return a ShareGroupTxnContext containing the MemberID and the specific record acquisitions. The KafkaProducer would then use this context to trigger the AddShareAcksToTxn RPC.
Regarding your points on KIP-1191 discussion thread, this Option B approach also allows us to solve the "Atomic DLQ" problem. If we include the DLQ write in the Producer's transaction, the Share Coordinator can wait for the WriteTxnMarker before finalising the move to the ARCHIVED state. This ensures the 1:1 audit trail remains intact even during broker failures.
I have updated the KIP and will keep improving/rephrasing it to clarify implementation: https://cwiki.apache.org/confluence/x/J448G
I am also starting to explore the broader challenge of Atomicity for cross-cluster scenarios. While standard transactions are intra-cluster, I am looking into "Acknowledgment Tokens" that could be recorded on a target cluster and mirrored back to the source coordinator to finalize state. I’d be interested to hear if the community sees value in laying the groundwork for that in this KIP.
Regards,Shekhar Prasad Rajak,
On Sunday 1 March 2026 at 11:09:59 pm GMT+5:30, Shekhar Prasad Rajak via dev <de...@kafka.apache.org> wrote:
Thank you Andrew for the review.
It took sometime to analyse the review comments and understand Kafka Producer 2PC implementation in depth.All the points are strongly valid and I have updated the KIP to adopt the RPC already exists and leverage the same methods for , CTP pattern (Consume-Transform-Produce) for kafka as sink or no producer usecase.
Please have a look: https://cwiki.apache.org/confluence/display/KAFKA/KIP-1289+Support+Hierarchical+Transactional+Acknowledgments+for+Share+Groups
Regards,Shekhar Prasad Rajak,
On Thursday 26 February 2026 at 03:32:36 am GMT+5:30, Andrew Schofield <sc...@apache.org> wrote:
Hi Shekhar,
Thanks for the KIP. Adding transactional support for acknowledgements is one of the future features I'd like to see for share groups, so I'm glad you've started this.
Here are some comments from an initial read.
AS1: Previously, Kafka transactions have always been a producer concept. You set a configuration called `transactional.id` in the producer configuration which gives authorization checking and helps with fencing. Then you use KafkaProducer methods like beginTransaction and commitTransaction to mark the start and end of the transaction. Only the producer is currently enabled for transactional operation, and the Kafka RPCs it uses include transactional ID as necessary to indicate operations which are transactional.
There will be differences for sure with share groups, but the overall model should probably still hold. Over the years, we've gradually tightened up the transaction protocol by adding features such as the producer epoch and two-phase commit. I would rather not re-invent everything. For example, WriteTxnMarkers is used by the transaction coordinator to tell the group coordinator to complete transactional operations on the __consumer_offsets topic. I would expect a similar kind of interaction with the share coordinator on the __share_group_state topic.
I suggest looking at how to adapt the existing Kafka protocol RPCs rather than defining new ones.
AS2: Why is there an acknowledge(ConsumerRecord<,>, AcknowledgeType, String) method? I would have expected that the transactional ID is a configuration, not a parameter of this method.
AS3: Have you considered how to write an application which uses a share group and producer, and coordinates their operations in the same transaction?
Thanks,
Andrew
On 2026/02/24 08:54:36 Shekhar Prasad Rajak via dev wrote:
> Hello team,
> In this discussion thread we will finalise the design for the KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups.
>
> Share Groups currently support only immediate acknowledgement modes (IMPLICIT/EXPLICIT). This causes data loss in distributed streaming frameworks:
>
> 1. Worker acknowledges records → Records removed from Kafka
> 2. Checkpoint fails before sink write
> 3. Records lost (acknowledged but never persisted)
>
> Goal: Enable exactly-once read semantics via transactional acknowledgements.
> We already have similar pattern in Kafka producer transaction
>
> KIP: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups - Apache Kafka - Apache Software Foundation
>
> Looking forward to community's feedback!
>
>
> Regards,Shekhar Prasad Rajak
>
>
Re: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups
Posted by Shekhar Prasad Rajak via dev <de...@kafka.apache.org>.
Hi team,
To support works around transactions - I believe we need General Transaction Session - Please have a look & review : https://cwiki.apache.org/confluence/display/KAFKA/KIP-1310%3A+General+Transaction+Session
Regards,
Shekhar
On Saturday 28 March 2026 at 01:30:05 pm GMT+5:30, Shekhar Prasad Rajak via dev <de...@kafka.apache.org> wrote:
Hi Andrew,
Thank you for the detailed breakdown, especially the distinction between making the ShareConsumer transactional (Option A) vs. the Producer-assisted model (Option B).
I agree that Option B is the superior path for architectural consistency. By having the Producer "assist" the ShareConsumer, we can leverage the existing mechanisms.
I am looking at implementing a prepareAcknowledgementsForTransaction() method on the KafkaShareConsumer. This would return a ShareGroupTxnContext containing the MemberID and the specific record acquisitions. The KafkaProducer would then use this context to trigger the AddShareAcksToTxn RPC.
Regarding your points on KIP-1191 discussion thread, this Option B approach also allows us to solve the "Atomic DLQ" problem. If we include the DLQ write in the Producer's transaction, the Share Coordinator can wait for the WriteTxnMarker before finalising the move to the ARCHIVED state. This ensures the 1:1 audit trail remains intact even during broker failures.
I have updated the KIP and will keep improving/rephrasing it to clarify implementation: https://cwiki.apache.org/confluence/x/J448G
I am also starting to explore the broader challenge of Atomicity for cross-cluster scenarios. While standard transactions are intra-cluster, I am looking into "Acknowledgment Tokens" that could be recorded on a target cluster and mirrored back to the source coordinator to finalize state. I’d be interested to hear if the community sees value in laying the groundwork for that in this KIP.
Regards,Shekhar Prasad Rajak,
On Sunday 1 March 2026 at 11:09:59 pm GMT+5:30, Shekhar Prasad Rajak via dev <de...@kafka.apache.org> wrote:
Thank you Andrew for the review.
It took sometime to analyse the review comments and understand Kafka Producer 2PC implementation in depth.All the points are strongly valid and I have updated the KIP to adopt the RPC already exists and leverage the same methods for , CTP pattern (Consume-Transform-Produce) for kafka as sink or no producer usecase.
Please have a look: https://cwiki.apache.org/confluence/display/KAFKA/KIP-1289+Support+Hierarchical+Transactional+Acknowledgments+for+Share+Groups
Regards,Shekhar Prasad Rajak,
On Thursday 26 February 2026 at 03:32:36 am GMT+5:30, Andrew Schofield <sc...@apache.org> wrote:
Hi Shekhar,
Thanks for the KIP. Adding transactional support for acknowledgements is one of the future features I'd like to see for share groups, so I'm glad you've started this.
Here are some comments from an initial read.
AS1: Previously, Kafka transactions have always been a producer concept. You set a configuration called `transactional.id` in the producer configuration which gives authorization checking and helps with fencing. Then you use KafkaProducer methods like beginTransaction and commitTransaction to mark the start and end of the transaction. Only the producer is currently enabled for transactional operation, and the Kafka RPCs it uses include transactional ID as necessary to indicate operations which are transactional.
There will be differences for sure with share groups, but the overall model should probably still hold. Over the years, we've gradually tightened up the transaction protocol by adding features such as the producer epoch and two-phase commit. I would rather not re-invent everything. For example, WriteTxnMarkers is used by the transaction coordinator to tell the group coordinator to complete transactional operations on the __consumer_offsets topic. I would expect a similar kind of interaction with the share coordinator on the __share_group_state topic.
I suggest looking at how to adapt the existing Kafka protocol RPCs rather than defining new ones.
AS2: Why is there an acknowledge(ConsumerRecord<,>, AcknowledgeType, String) method? I would have expected that the transactional ID is a configuration, not a parameter of this method.
AS3: Have you considered how to write an application which uses a share group and producer, and coordinates their operations in the same transaction?
Thanks,
Andrew
On 2026/02/24 08:54:36 Shekhar Prasad Rajak via dev wrote:
> Hello team,
> In this discussion thread we will finalise the design for the KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups.
>
> Share Groups currently support only immediate acknowledgement modes (IMPLICIT/EXPLICIT). This causes data loss in distributed streaming frameworks:
>
> 1. Worker acknowledges records → Records removed from Kafka
> 2. Checkpoint fails before sink write
> 3. Records lost (acknowledged but never persisted)
>
> Goal: Enable exactly-once read semantics via transactional acknowledgements.
> We already have similar pattern in Kafka producer transaction
>
> KIP: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups - Apache Kafka - Apache Software Foundation
>
> Looking forward to community's feedback!
>
>
> Regards,Shekhar Prasad Rajak
>
>
Re: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups
Posted by Shekhar Prasad Rajak via dev <de...@kafka.apache.org>.
Hi Andrew,
Thank you for the detailed breakdown, especially the distinction between making the ShareConsumer transactional (Option A) vs. the Producer-assisted model (Option B).
I agree that Option B is the superior path for architectural consistency. By having the Producer "assist" the ShareConsumer, we can leverage the existing mechanisms.
I am looking at implementing a prepareAcknowledgementsForTransaction() method on the KafkaShareConsumer. This would return a ShareGroupTxnContext containing the MemberID and the specific record acquisitions. The KafkaProducer would then use this context to trigger the AddShareAcksToTxn RPC.
Regarding your points on KIP-1191 discussion thread, this Option B approach also allows us to solve the "Atomic DLQ" problem. If we include the DLQ write in the Producer's transaction, the Share Coordinator can wait for the WriteTxnMarker before finalising the move to the ARCHIVED state. This ensures the 1:1 audit trail remains intact even during broker failures.
I have updated the KIP and will keep improving/rephrasing it to clarify implementation: https://cwiki.apache.org/confluence/x/J448G
I am also starting to explore the broader challenge of Atomicity for cross-cluster scenarios. While standard transactions are intra-cluster, I am looking into "Acknowledgment Tokens" that could be recorded on a target cluster and mirrored back to the source coordinator to finalize state. I’d be interested to hear if the community sees value in laying the groundwork for that in this KIP.
Regards,Shekhar Prasad Rajak,
On Sunday 1 March 2026 at 11:09:59 pm GMT+5:30, Shekhar Prasad Rajak via dev <de...@kafka.apache.org> wrote:
Thank you Andrew for the review.
It took sometime to analyse the review comments and understand Kafka Producer 2PC implementation in depth.All the points are strongly valid and I have updated the KIP to adopt the RPC already exists and leverage the same methods for , CTP pattern (Consume-Transform-Produce) for kafka as sink or no producer usecase.
Please have a look: https://cwiki.apache.org/confluence/display/KAFKA/KIP-1289+Support+Hierarchical+Transactional+Acknowledgments+for+Share+Groups
Regards,Shekhar Prasad Rajak,
On Thursday 26 February 2026 at 03:32:36 am GMT+5:30, Andrew Schofield <sc...@apache.org> wrote:
Hi Shekhar,
Thanks for the KIP. Adding transactional support for acknowledgements is one of the future features I'd like to see for share groups, so I'm glad you've started this.
Here are some comments from an initial read.
AS1: Previously, Kafka transactions have always been a producer concept. You set a configuration called `transactional.id` in the producer configuration which gives authorization checking and helps with fencing. Then you use KafkaProducer methods like beginTransaction and commitTransaction to mark the start and end of the transaction. Only the producer is currently enabled for transactional operation, and the Kafka RPCs it uses include transactional ID as necessary to indicate operations which are transactional.
There will be differences for sure with share groups, but the overall model should probably still hold. Over the years, we've gradually tightened up the transaction protocol by adding features such as the producer epoch and two-phase commit. I would rather not re-invent everything. For example, WriteTxnMarkers is used by the transaction coordinator to tell the group coordinator to complete transactional operations on the __consumer_offsets topic. I would expect a similar kind of interaction with the share coordinator on the __share_group_state topic.
I suggest looking at how to adapt the existing Kafka protocol RPCs rather than defining new ones.
AS2: Why is there an acknowledge(ConsumerRecord<,>, AcknowledgeType, String) method? I would have expected that the transactional ID is a configuration, not a parameter of this method.
AS3: Have you considered how to write an application which uses a share group and producer, and coordinates their operations in the same transaction?
Thanks,
Andrew
On 2026/02/24 08:54:36 Shekhar Prasad Rajak via dev wrote:
> Hello team,
> In this discussion thread we will finalise the design for the KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups.
>
> Share Groups currently support only immediate acknowledgement modes (IMPLICIT/EXPLICIT). This causes data loss in distributed streaming frameworks:
>
> 1. Worker acknowledges records → Records removed from Kafka
> 2. Checkpoint fails before sink write
> 3. Records lost (acknowledged but never persisted)
>
> Goal: Enable exactly-once read semantics via transactional acknowledgements.
> We already have similar pattern in Kafka producer transaction
>
> KIP: KIP-1289 Support Hierarchical Transactional Acknowledgments for Share Groups - Apache Kafka - Apache Software Foundation
>
> Looking forward to community's feedback!
>
>
> Regards,Shekhar Prasad Rajak
>
>