Corrected message.release() in PulsarMessageAck.java #2580
Conversation
|
@ozangunalp What do you think about enabling the pooling by default? What's more, we should probably release the messages on FailureHandlers which are not performing ACK. https://github.com/smallrye/smallrye-reactive-messaging/blob/main/smallrye-reactive-messaging-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/fault/PulsarContinue.java#L42 |
| @@ -41,7 +41,6 @@ public Uni<Void> handle(PulsarIncomingMessage<?> message) { | |||
| return Uni.createFrom().completionStage(() -> { | |||
| var txnMetadata = message.getMetadata(PulsarTransactionMetadata.class); | |||
| if (txnMetadata.isPresent()) { | |||
There was a problem hiding this comment.
It's already released in line #40
Ok for releasing the message on all failure handlers. But I think we should keep the message pooling to the Pulsar consumer default. I didn't hesitate to add |
Message is already released here:
smallrye-reactive-messaging/smallrye-reactive-messaging-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/ack/PulsarMessageAck.java
Line 40 in a9719f0