Acknowledge outside of consumeMany and consumeOne

[tripletk]

Is there support to acknowledge outside of the consumeOne and consumeMany 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<String> 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);

[lhotari]

Is there support to acknowledge outside of the consumeOne and consumeMany of AdaptedReactiveMessageConsumer?

It is intentionally not supported. It probably needs a longer explanation, but the short explanation is that acknowledgement state must happen in the pipeline that is part of consumeOne or consumeMany. If you'd line to handle acknowledgements later in the pipeline, it should be possible to use data oriented programming to pass on the message results downstream so that in the end of the stream, the message results come out of the pipeline.

You would need to change the mindset for acknowledgement that it's not an imperative action, but "data", a result of the processing which the "runtime" takes care of handling towards the Pulsar broker.

There are multiple patterns of how to pass on such state downstream in a pipeline. If you explain more of the use case, we could find a solution together that meets your requirements.

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.

Why do you want acknowledgements to happen later? Answering that question could help find a solution that is aligned with the way how pulsar-client-reactive is designed to be used.

One detail about using ReactiveMessageConsumer directly is that it's very easy to misuse it. For most use cases, it's better to use the ReactiveMessagePipeline built with ReactiveMessagePipelineBuilder instead of using ReactiveMessageConsumer directly. With ReactiveMessagePipelineBuilder use can also configure concurrency and key-ordered message processing which are one of the main reasons to use pulsar-client-reactive since you can scale processing with high concurrency/parallelism while processing messages in key order.