spring-projects/spring-amqp

Spring AMQP - support for Spring programming model with AMQP, especially but not limited to RabbitMQ

Gary Russell

@garyrussell

Bottom line is you cannot use the same queue as the other consumers, as I have been saying all along.

Bambus

@vlatk0o

Thank you Gary

I have set this in the application properties:
spring.rabbitmq.listener.simple.prefetch=0

and I in the rabbitMq i can see prefetch 1

Bambus

@vlatk0o

And as you said I receive every third message :)

Gary Russell

@garyrussell

Yeah - we coerce the prefetch to be at least 1; TBH, I am not sure what 0 means.
But why are you pursuing this? It's obvious the missing messages are going to the other consumers.

Bambus

@vlatk0o

Thank you a lot Gary

Patrik Mihalcin

@pmihalcin

@garyrussell
I simplified and rephrased my question and materialized it into SO question: https://stackoverflow.com/questions/60993877/rabbittemplates-setchanneltransacted-flag-causes-message-being-not-delivered-to
Feedback is more than welcome

Gary Russell

@garyrussell

The exchange is different in your second log [mer.cache.update.fanout].

Patrik Mihalcin

@pmihalcin

sorry.. it's typo.. I use the same

Gary Russell

@garyrussell

I can't explain it then; it works fine for me.

Artem Bilan

@artembilan

Huh? You have SO thread and you still chatting here?..

Gary Russell

@garyrussell

SO's not good for chatting.

Patrik Mihalcin

@pmihalcin

thanks a lot for the test.. I will have to dig even deeper and debug more

David Shapiro

@davyshaps12_twitter

Hi everyone, I was wondering if there was a way to manually trigger a RabbitListener shutdownTimeout. Specifically, I am looking to trigger a 20 minute shutdown on a specific RabbitListener when the listener has handled 500 messages. Apologies if I am unclear, not posting in the proper location, or if i need to provide some more context. Any help would be greatly appreciated. Thanks!

Gary Russell

@garyrussell

@davyshaps12_twitter It's the right place but it's not clear what you mean by "trigger a 20 minute shutdown". If you mean shut it down 20 minutes after the 500th message, then just schedule a task to shut it down at that time.

Gary Russell

@garyrussell

(Give the listener an id and use the RabbitListenerEndpointRegistry bean to get a reference to its listener container and call stop()).

David Shapiro

@davyshaps12_twitter

Thank you Gary, went with the solution you recommended using the RabbitListenerEndpointRegistry

Priyanshu Shukla

@priyanshus1

Hi Guys,
In case of RPC implementation, how to inform when a message is dropped from the queue due to TTL expiration.?

Gary Russell

@garyrussell

@priyanshus1 There is no mechanism available to do that.

Priyanshu Shukla

@priyanshus1

@garyrussell is it bounded by platform capabilities or we can implement such feature in spring amqp core ?

Artem Bilan

@artembilan

how to inform when a message is dropped from the queue due to TTL expiration.?

You can configure a DLX for those dropped messages and consume them from a bound queue.
See here: https://www.rabbitmq.com/dlx.html

Gary Russell

@garyrussell

Right; I thought (s)he was asking whether the broker would send some kind of event to the producer; of course, you can add a dead letter queue and get a notification that way.

Yuval Bonen

@yuvalprtn

Hi
We are looking for Shutdown-Gracefully option using Spring-AMQP
One of our main flows is based on a pipeline we created using RabbitMQ and we would like to support auto-scaling.
Growing is not the problem but reducing the number of machines while everything is working (or if it is a spot machine taken) is a bit of a problem.
Would like to know if there is a way to notify Spring-AMQP that it should stop receiving any new messages from RabbitMQ and just finish up the workload it is currently running.
Once done we will be able to shutdown this machine/pod/process gracefully.

Anything implemented in this direction or we need to tailor a solution ourselves?
Thanks

Gary Russell

@garyrussell

@yuvalprtn Just stop() the listener container(s) (or RabbitListenerEndpointRegistry if using @RabbitListener). Any prefetched messages will be processed. You will see Waiting for workers to finish.. You may need to increase the shutdownTimeout from its default 5000ms to avoid a force close with Workers not finished..

Olle Törnström

@olle

Hi everyone! I'm having trouble with recovery of a programmatically created listener, after broker connection lost, that is using an anonymous queue (excl., auto-delete). Tracking my issue here olle/query-response-spring-amqp#19. Any ideas are welcome, also anyone using anonymous/auto-delete queues and recovering connection failures successfully? Thanks!

Yuval Bonen

@yuvalprtn

Thanks @garyrussell will give it a spin!

Yuval Bonen

@yuvalprtn

@garyrussell do you know if calling the graceful shutdown https://docs.spring.io/spring-boot/docs/current-SNAPSHOT/reference/htmlsingle/#boot-features-graceful-shutdown is calling the RabbitListenerEndpointRegistryas well?

Gary Russell

@garyrussell

@yuvalprtn I am not familiar with it, but it doesn't sound like it does from the description; it only talks about the web containers. You would have to ask the Boot folks.

Yuval Bonen

@yuvalprtn

Thanks @garyrussell

Olle Törnström

@olle

Thanks @garyrussell, for help with my issue. Registering programmatically created broker resources with a GenericApplicationContextis important, and enables recovery by the RabbitAdmin. Hope someone finds this usefull too.

David Shapiro

@davyshaps12_twitter

Hi, Good morning! I am currently using 'IntegrationFlows.imapIdleAdapter' to listen to messages coming in from a single incoming gmail inbox. I was wondering how to listen to MULTIPLE inboxes using the same or another IMAP related class, instead of the single gmail inbox I am currently listening to. Let me know if there is any code or context that I can provide to make it more clear what I am looking for. Thank you!

2 replies

David Shapiro

@davyshaps12_twitter

Hi! I one other question. I am running this code to listen to emails

return IntegrationFlows
        .from(Mail.imapIdleAdapter(imapUrl())
                  .autoStartup(true)
                  .javaMailProperties(p -> p.put("mail.imap.ssl.trust", "*")
                                            .put("mail.imap.ssl.enable", "true"))
                  .shouldMarkMessagesAsRead(false)
                  .shouldDeleteMessages(false)
                  .shouldReconnectAutomatically(true)
                  .autoCloseFolder(false))
        .handle(m-> emailReceiverService.createAndSaveEmail((MimeMessage)m.getPayload()))
        .get();

4 replies

Occasionally I am getting the follow exception in a higher environment

javax.mail.FolderClosedException: * BYE JavaMail Exception: javax.net.ssl.SSLException: Connection timed out (Read failed)
    at com.sun.mail.imap.IMAPFolder.handleIdle(IMAPFolder.java:3316)

I was wondering how you would recommend handling this issue? by poking the gmail server with a 'mail.timeout' property? Or is there another recommended method? Thanks again!

ladh

@leeadh

Hi all I am trying to connect spring cloud data flow to RabbitMQ. I keep hitting a Caused by: java.net.ConnectException: Connection refused (Connection refused)

I thought it was because my RabbitMQ is deployed in dockers. So i tried a local installation on my mac using brew instead but still am hitting the same issue. Does anyone know why?

2 replies

Olle Törnström

@olle

@leeadh Sounds like credentials mismatch. I'm sorry but I don't have any experience with the cloud data setup. The default credentials in Spring AMQP are tuned to just work with a most vanilla RabbitMQ. Here's my go-to docker-compose.yaml

version: '3'
services:
  rabbitmq:
    image: "rabbitmq:3-management"
    ports:
      - "15672:15672"
      - "5672:5672"

@leeadh properties and credentials are basically localhost:5672 and guest/guest.

Priyanshu Shukla

@priyanshus1

HI All ,
is it a bad idea of have static numbers of channels min and max concurrency = n. Does it make performance bad ?

8 replies

Florian Beaufumé

@fbeaufume

Hi guys, in a Spring Boot 2.2 application, I need to consume messages from Azure Service Bus. That bus supports AMQP 1.0. I did not see the AMQP versions supported by Spring AMQP (I suppose 0-9-1 only since that's the version natively supported by RabbitMQ). So my questions are :

What are the AMQP versions supported by Spring AMQP ?
Can Spring AMQP alone be used to consume messages from Azure Service Bus ?
If not, is there a recommended portable (i.e. not dependant to Azure or Microsoft) solution ?

Thank you for your help.

Artem Bilan

@artembilan

Spring AMPQ is mostly based on the AMQP 0.9.1 protocol.

just because AMQP 1.0 is fully compatible with JMS spec, there is no reason to do something in Spring AMQP to support it.

Consider to use whatever JMS library is there to consume that Azure Service Bus

Florian Beaufumé

@fbeaufume

This message was deleted

Gary Russell

@garyrussell

Take a look at Apache QPID Proton J

Florian Beaufumé

@fbeaufume

Great, thank you guys, is one of these suggestion nicely integrated to Spring (Boot) ?

macportter

@macportter

Hi,
To achieve better throughput in Spring AMQP, is it better to use a larger number of concurrent consumers or to scale horizontally with fewer consumers?
Example:
5 instances with 20-30 concurrent consumers each
10 instances with 10 concurrent consumers each
Thanks

5 replies

Vinh

@KagariSan

Hi everyone, I trying to build up a system that require listen to large number of queue (about 10k queue) defined on database. Message will come in queue and has to be execute one by one.
I configured DirectMessageListenerContainer to ConsumersPerQueue = 1 and PrefetchCount = 1. Is that ok?
Btw when listen on queue (before it created) it's throw error queue not exists. There are anyway to create queue on fly if queue not exist then create it?

5 replies

maglioneangelo

@maglioneangelo

Hi everyone,
I'm having an issue with Spring-Rabbit 2.x setting the AutoStartup to false.
During the initialization of the application, a 3th part library is doing an applicationContext.refresh.

Starting from spring-Rabbit 2.x

private void startIfNecessary(MessageListenerContainer listenerContainer) {
if (this.contextRefreshed || listenerContainer.isAutoStartup()) {
listenerContainer.start();
}
}
the listener is started also in case the autostartup is set to false. Shouldn't be an && instead of || ? Is it a regression?
Thanks in advance

17 replies

Patrik Mihalcin

@pmihalcin

Hi all.. how can I get number of unacknowledged messages for given rabbitmq queue? I tried to use /api/queues/%2F/QUEUE_NAME, but it always returns 0

my test scenario:
send e.g. 10 messages to headers exchange with correct header
consume message with consumers created by Amqp.inboundAdapter
wait in loop while unacknowledged == 0

the problem is first http call for number of unacknowledged returns 0

2 replies

Where communities thrive

People

Repo info

Activity