RabbitListener's returning Mono<Void> doesn't ack the message

[johanhaleby]

As discussed on stackoverflow a RabbitListener returning Mono<Void> will not ack the message. For example:

@RabbitListener
public Mono<Void> myListener(MyMessage myMessage) {
    Mono<Void> mono = myService.doSomething(myMessage);
    return mono;
}

The reason is that spring-rabbit doesn't seem to handle empty streams.

[jndietz]

This still appears to be a problem.

    @RabbitListener(queues = "audit")
    Mono<Void> listen(CustomerActivityEventWrapper wrapper) {
        log.debug("Received message {}", wrapper.toString())
        strategies
                .find { it.handles(wrapper.payload) }
                .handleEvent(wrapper.payload)
                      // 👆 returns a Mono<Void>
    }

The API call happening within my strategy occurs, but I still see this in the console:

2022-08-04 12:15:20.056 DEBUG 21028 --- [ntContainer#0-2] com.somecompany.listener.EventListener          : Received message com.somecompany.model.rabbit.event.CustomerActivityEventWrapper@150f7163
2022-08-04 12:15:20.184  WARN 21028 --- [ntContainer#0-2] .a.r.l.a.MessagingMessageListenerAdapter : Container AcknowledgeMode must be MANUAL for a Mono<?> return type; otherwise the container will ack the message immediately
2022-08-04 12:15:20.669 DEBUG 21028 --- [ntContainer#0-2] o.s.w.r.f.client.ExchangeFunctions       : [386278] HTTP POST https:/some-api.somecompany.com/api/v1/customer-events
2022-08-04 12:15:22.303 DEBUG 21028 --- [ctor-http-nio-2] o.s.http.codec.json.Jackson2JsonEncoder  : [386278] Encoding [com.somecompany.model.jsite.event.JsiteAuditEvent@4aac9d27]
2022-08-04 12:15:22.408 DEBUG 21028 --- [ctor-http-nio-2] o.s.w.r.f.client.ExchangeFunctions       : [386278] [1d990f1a-1, L:/10.5.26.18:53692 - R:somecompany.somecompany.com/10.72.93.31:443] Response 201 CREATED
2022-08-04 12:15:22.418 ERROR 21028 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory       : Shutdown Signal: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 1, class-id=60, method-id=80)
2022-08-04 12:15:22.774  INFO 21028 --- [ntContainer#0-2] o.s.a.r.l.SimpleMessageListenerContainer : Restarting Consumer@64f3ca6: tags=[[amq.ctag-Zwr_bRDdqroucKk0JDDOcQ]], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,3), conn: Proxy@22c8ee48 Shared Rabbit Connection: SimpleConnection@7f93f4c [delegate=amqp://guest@127.0.0.1:5672/, localPort= 53656], acknowledgeMode=AUTO local queue size=0

[artembilan]

@jndietz ,

Sorry, what the problem are you observing?
You log doesn't show any errors.

[jndietz]

@artembilan There is an ERROR line:

2022-08-04 12:15:22.418 ERROR 21028 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory       : Shutdown Signal: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 1, class-id=60, method-id=80)

My thought is that it isn't acking the message when it should be.

[artembilan]

unknown delivery tag 1,

I guess that's because you didn't follow the recommendation from the previous log:

Container AcknowledgeMode must be MANUAL for a Mono<?> return type; otherwise the container will ack the message immediately

[jndietz]

I guess that's because you didn't follow the recommendation from the previous log:

Yep... you're right. I guess I mixed up how MANUAL is supposed to work with reactor. I assumed I was supposed to also perform the acknowledgement, but I guess that isn't the case. This worked well for me:

    @RabbitListener(queues = "audit", ackMode = "MANUAL") // 👈 no actual acknowledgement needed in the code below
    Mono<Void> listen(CustomerActivityEventWrapper wrapper) {
        log.debug("Received message {}", wrapper.toString())
        strategies
                .find { it.handles(wrapper.payload) }
                .handleEvent(wrapper.payload)
    }

Thanks @artembilan

[garyrussell]

I suppose the error message emitted by the amqp-client might be a bit confusing:

reply-text=PRECONDITION_FAILED - unknown delivery tag 1

In this case, it means delivery tag 1 has already been acknowledged (and therefor the second ack used an "unknown" tag.