[EI v6.6.0] Messages are accumulating in MB Queues

[PasinduGunarathne]

We have encountered an issue with the EI v6.6.0 MB profile that the published messages are accumulating in the MB queues. According to the investigation this behaviour caused when using a Spring AMQP consumer client to consume messages from MB queues.

Steps to reproduce

1. Use a sample Spring AMQP client consumer code [1]
2. Download EI v6.6.0.181 and run the MB profile
3. Publish some messages(~10000) to the MB queue( you can use the Spring client to publish messages as well)
4. Use Spring client to consume messages
5. Run the Spring client for sometime and check the MS queues
6. Observe the MB logs for the following error logs

TID: [-1] [] [2024-04-01 13:06:02,211] ERROR {org.wso2.andes.kernel.subscription.OutBoundMessageTracker} - Message reference has been already cleared for message id 109699249755652096. Acknowledge or Nak is already received
TID: [-1] [] [2024-04-01 13:06:02,237]  WARN {org.wso2.andes.server.queue.QueueEntryImpl} - Requesting rejection by null subscriber:org.wso2.andes.server.queue.QueueEntryImpl@3631c166
TID: [-1] [] [2024-04-01 13:06:02,218] ERROR {org.wso2.andes.kernel.disruptor.LogExceptionHandler} - [ Sequence: 11725 ] Exception occurred while processing inbound events.Event type: ACKNOWLEDGEMENT_EVENT java.lang.NullPointerException
	at org.wso2.andes.kernel.AndesAckEvent.addTraceLog(AndesAckEvent.java:134)
	at org.wso2.andes.kernel.AndesAckEvent.setMetadataReference(AndesAckEvent.java:80)
	at org.wso2.andes.kernel.disruptor.inbound.AckHandler.ackReceived(AckHandler.java:107)
	at org.wso2.andes.kernel.disruptor.inbound.AckHandler.processAcknowledgements(AckHandler.java:89)
	at org.wso2.andes.kernel.disruptor.inbound.AckEventBatchHandler.onEvent(AckEventBatchHandler.java:99)
	at org.wso2.andes.kernel.disruptor.inbound.AckEventBatchHandler.onEvent(AckEventBatchHandler.java:32)
	at com.lmax.disruptor.BatchEventProcessor.processEvents(BatchEventProcessor.java:168)
	at com.lmax.disruptor.BatchEventProcessor.run(BatchEventProcessor.java:125)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)

Even though we use invalid consumer to consume messages from MB and its giving errors due that invalid client, the MB itself should be able to handle those kind of situations.

Inorder to find more information regarding the Spring AMQP client implementation you can refer to the documentation [2]

[1] https://github.com/spring-projects/spring-amqp-samples/tree/main
[2] https://docs.spring.io/spring-amqp/docs/3.1.0/reference/html/#message-listener

Thank you,
Pasindu G.