apache / apache/pulsar

[Enhancement] Expose Consumer in SinkContext for Transactional Message Acknowledgment

Open
#24,588 11 comments 0 reactions 0 assignees View on GitHub
type/enhancement
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.