[Enhancement] Expose Consumer in SinkContext for Transactional Message Acknowledgment
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before reporting
- [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.
### Motivation
I'm working on a Pulsar sink that needs to process incoming messages and produce to multiple output topics atomically. The key requirement is that the message acknowledgment should be part of the same transaction as the output operations, either everything succeeds together or everything fails together.
## Problem
When implementing custom Pulsar sinks with transactions to the cluster itself, there's no way to include the input message acknowledgment within the transaction scope. The Record.ack() method operates outside of any transaction.
### Example
```java
@Override
public void write(Record record) {
Transaction txn = client.newTransaction().build().get();
// Send to output topics within transaction
eventProducer.newMessage(txn).value(data).send();
txn.commit().get();
// This happens outside the transaction
record.ack();
}
```
### Proposed Solution
Expose the underlying consumer in SinkContext:
```java
public interface SinkContext {
...
Consumer getConsumer();
}
```
This would enable:
```java
@Override
public void write(Record record) {
Transaction txn = client.newTransaction().build().get();
eventProducer.newMessage(txn).value(data).send();
// Include acknowledgment in the transaction
Consumer consumer = sinkContext.getConsumer();
consumer.acknowledgeAsync(record.getMessageId(), txn).get();
txn.commit().get();
}
```
- [X] I'm willing to submit a PR!
Contributor guide
Research direction
Start with the SinkContext interface and the Consumer and Record APIs named in the issue, then trace how sink acknowledgments and transactions are currently handled. Done means exposing the consumer through SinkContext and allowing acknowledgeAsync(record.getMessageId(), txn) to participate in the same transaction as output operations.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend-api-design, distributed-systems
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100