Our grail application is using following RabbitMQ client and using XML configuration
spring-integration-amqp:3.0.8.RELEASE
spring-rabbit:1.4.5.RELEASE
Following is the XML configuration for one of the queue that is giving issue:
<integration:poller id="default" default="true" fixed-rate="2000"/>
<rabbit:connection-factory id="rabbitConnectionFactory"
host="${rabbitmq.host}"
username="${rabbitmq.username}"
password="${rabbitmq.password}"
requested-heartbeat="30"/>
<rabbit:admin connection-factory="rabbitConnectionFactory"/>
<rabbit:template id="amqpTemplate"
connection-factory="rabbitConnectionFactory"/>
<rabbit:queue id="batchUnspscQueue" name="${unspsc.batch.queue}"/>
<integration:channel id="fromUnspscBatch">
<integration:queue capacity="10" />
</integration:channel>
<amqp:inbound-channel-adapter channel="fromUnspscBatch" concurrent-consumers="1" queue-names="${unspsc.batch.queue}" acknowledge-mode="NONE" channel-transacted="false" connection-factory="rabbitConnectionFactory" />
<integration:service-activator
id="erpUnspscBatchSubscriber"
ref="erpMessageHandlerService"
method="performUnspscBatchUpdate"
input-channel="fromUnspscBatch"
output-channel="nullChannel"/>
In production we are noticing that when there is a heavy load on 'unspsc.batch.queue' (i.e. input messages are more than the channel internal queue capacity i.e. 10), suddenly consumers start disappearring from the Queue (Noticed using RabbitMQ Management UI)
Looks like connection is getting reset but strange thing is consumers for all other queues attached to the same connection received shutdown message and then restarted but consumers attached to this specific queue which was busy didn't restarted; messages stopped getting processed for this queue and started getting queued in the broker queue.
For now, whenever we are in this situation we need to restart the application. Some time we need to restart application multiple times because consumers process some messages and then again disappear.
Noticed following log messages for other queue consumers which restarted fine:
02.02-07:10:21.202 [SimpleAsyncTaskExecutor-3] INFO org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer - Restarting Consumer: tags=[{amq.ctag-OaL0P3G4j0VQKGQ16lGlBg=background-worker.queue}], channel=Cached Rabbit Channel: AMQChannel(amqp://wluser@10.6.10.4:5672/,44), acknowledgeMode=NONE local queue size=0
...
02.02-07:10:21.200 [SimpleAsyncTaskExecutor-2] WARN org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer - Consumer raised exception, processing can restart if the connection factory supports itcom.rabbitmq.client.ShutdownSignalException: connection error
We are able to replicate this in dev env and when we change the capacity to a huge number then it do not give us this issue. Because consumer is able to bring all the queued messages and then process then in its own pace. But this will have its own side effect on the application cpu /ram and if server restarts in-between all messages will be lost as ACK mode is set to NONE.
<integration:channel id="fromUnspscBatch">
<integration:queue capacity="100000" />
</integration:channel>
Any help or insight on this will be very appreciated.
1: Is there any configuration we are missing to handle a scenario i.e. when capacity is kept less (e.g. 10) and there are good amount of messages in the broker queue (say 10000)?
On https://docs.spring.io/spring-integration/docs/1.0.x/reference/html/ch03s02.html noticed followign line. Is it having any side effect when channel internal queue reached its capacity?
If the queue has reached capacity, then the sender will block until room is available
2: Also any idea why consumers for all other queues attached to the same connection received shutdown message and then restarted but consumers attached to this specific queue which was busy didn't restarted?