apache / apache/pulsar-client-reactive

Acknowledge outside of consumeMany and consumeOne

Open
#220 1 comment 0 reactions 1 assignee Claimed by @lhotari View on GitHub
Dominant language
Java
Stars
67
Forks
23
Avg merge
22h
Merged PRs (30d)
2

Description

Is there support to acknowledge outside of the [consumeOne](https://github.com/apache/pulsar-client-reactive/blob/6170ebe7f3b6d7ee75071d9b6fb3d6a750a622e7/pulsar-client-reactive-adapter/src/main/java/org/apache/pulsar/reactive/client/internal/adapter/AdaptedReactiveMessageConsumer.java#L63) and [consumeMany](https://github.com/apache/pulsar-client-reactive/blob/6170ebe7f3b6d7ee75071d9b6fb3d6a750a622e7/pulsar-client-reactive-adapter/src/main/java/org/apache/pulsar/reactive/client/internal/adapter/AdaptedReactiveMessageConsumer.java#L216) of AdaptedReactiveMessageConsumer?

My usecase is I want to consume the messages but don't want the acknowledgement to happen within the consumeOne, consumeMany. I want to separately acknowledge the pulsar message later on in the reactive pipeline after consumeOne, consumeMany.

e.g.
```
ReactiveMessageConsumer messageConsumer=
reactivePulsarClient.messageConsumer(Schema.STRING)
.topic(topicName)
.subscriptionName("sub")
.build();
messageConsumer.consumeMany(messageFlux ->
messageFlux.map(message ->
doSomeProcessing(message)))
.flatMap(msgResult -> {
MessageId msgId = msgResult.getMessageId();
// Acknowledge here outside of consumeMany or consumeOne
handleAcknowledgement(msgId);
})
// for demonstration
.subscribe(System.out::println);
```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.