Hi All, I've searched for this topic now a fair bit and I'm not quite getting a conclusive answer to my issue.
What we've noticed the last while is that in some cases under high enough load with multiple consumers on the Aggregator route some messages get overwritten in the aggregation result with a following message. Seems like a very random and intermittent event and I'm not quite sure how to replicate that with a test case as I'm not 100% sure of the mechanics behind how the Aggregator is treating multiple consumers. Since we've started seeing this issue there was a synchronized block put around the aggregation result list modifications and that seems to have done absolutely nothing in terms of reducing or eliminating the items overwriting each other from time to time. Here is my setup: RabbitMQ endpoint configuration. inputEndpoint: rabbitmq://batchExchange?connectionFactory=rabbitConnectionFactory&autoDelete=false&queue=batchrouteQueue&exchangeType=topic&autoAck=false&bridgeEndpoint=true&concurrentConsumers=15&threadPoolSize=15&channelPoolMaxSize=15&prefetchCount=1&prefetchEnabled=true,vmsys301.vhihealthcare.net:5671 The route (goes through an abstract route config for retry and exception handling). from("direct:" + getServiceName()) .log("Receiving ${body}") .process(this::process) //sets the aggregation header .choice() .when(header(AGGREGATIONKEY).contains("ONLINE ")) .aggregate(header(AGGREGATIONKEY)) .aggregationStrategyRef(aggregationStrategy) .aggregationRepositoryRef(batchOnlineRepo) .completionSize(batchSize) .completionTimeout(onlineBatchTimeout) .log("Sending out Online ${body}") .to(outputEndpoint).endChoice() .otherwise() .aggregate(header(AGGREGATIONKEY)) .aggregationStrategyRef(aggregationStrategy) .aggregationRepositoryRef(batchRepo) .completionSize(batchSize) .completionTimeout(batchTimeout) .log("Sending out ${body}") .to(outputEndpoint).endChoice().end(); The AggregationStrategy public class BatchAggregationStrategy implements AggregationStrategy { final Logger logger = LoggerFactory.getLogger(this.getClass()); public Exchange aggregate(Exchange oldExchange, Exchange newExchange) { String aggregationkey = newExchange.getIn().getHeader("aggregationkey", String.class); if (oldExchange == null) { Object documentPayload = newExchange.getIn().getBody(Object.class); if (documentPayload instanceof DocumentPayloadBatch) { logger.debug("Skip aggregation ["+aggregationkey+"](this is a recovered exchange, just return it): " + newExchange.getIn()); return newExchange; } else { DocumentPayloadBatch documentPayloadList = new DocumentPayloadBatch(); synchronized(documentPayloadList) { documentPayloadList.add((DocumentPayload) documentPayload); newExchange.getIn().setBody(documentPayloadList); logger.debug("First item of aggregation ["+aggregationkey+"](old exchange =null): " + newExchange.getIn()); } return newExchange; } } DocumentPayload documentPayload = newExchange.getIn().getBody(DocumentPayload.class); DocumentPayloadBatch documentPayloadList = oldExchange.getIn().getBody(DocumentPayloadBatch.class); synchronized(documentPayloadList) { documentPayloadList.add(documentPayload); logger.debug("Payload with uuid " + documentPayload.getUuid() + " added to list for batch sending. Have " + documentPayloadList.size() + " payloads in list ["+aggregationkey+"] now ."); } //for some reason documentPayloadList doesn't get updated with the latest entry sometimes (1 in 5000 requests or so on average) return oldExchange; } } Given the above aggregation strategy code, should I be also synchronizing the oldExchange.getIn().getBody(DocumentPayloadBatch.class) call? Would the Aggregator be even compatible with multiple consumers like in the above setup? I can reduce it down to a single consumer, it's a fairly quick aggregation anyway, though I would prefer to keep my 15 consumers of course. Due to the sporadic and intermittent nature of the issue I'm leaning to think it's to do with the consumers racing each other to update the aggregation result, but then the resulting list updates are already in synchronized blocks. The remaining part of the aggregation that's outside of said block is the exchange updates and reads themselves. The updated exchange with the new item hasn't been returned yet before another consumer grabs a version of it and starts its own update (not sure if that's even possible though). Thanks, Valdis
Vhi Group DAC (Vhi) is a holding company for insurance and healthcare services, which include Vhi Healthcare DAC, Vhi Insurance DAC, Vhi Health Services DAC and Vhi Investments DAC. Vhi Healthcare DAC trading as Vhi Healthcare and Vhi Insurance DAC trading as Vhi Insurance are regulated by the Central Bank of Ireland. Vhi Healthcare is tied to Vhi Insurance DAC for health insurance in Ireland which is underwritten by Vhi Insurance DAC. Vhi Healthcare is tied to Zurich Life Assurance plc for Vhi Life Term Insurance and Vhi Mortgage Protection which are underwritten by Zurich Life Assurance plc. Vhi Healthcare is tied to Collinson Insurance Solutions Europe Limited for MultiTrip Travel Insurance and Vhi Dental Insurance which are underwritten by Great Lakes Insurance SE and for Vhi International Health Insurance which is underwritten by Collinson Insurance Europe Limited. For more information about the Vhi Group please go to: https://www.vhi.ie/about-vhi. Tá Vhi Group DAC (Vhi) ina chuideachta sealbhaíochta le haghaidh seirbhísí árachais agus seirbhísí cúram sláinte, lena n-áirítear Vhi Healthcare DAC, Vhi Insurance DAC, Vhi Health Services DAC agus Vhi Investments DAC. Déanann Banc Ceannais na hÉireann rialáil ar Vhi Healthcare DAC, ag trádáil dó mar Vhi Healthcare, agus ar Vhi Insurance DAC, ag trádáil dó mar Vhi Insurance. Tá Vhi Healthcare ceangailte le Vhi Insurance DAC le haghaidh árachas sláinte in Éirinn, rud atá frithgheallta ag Vhi Insurance DAC. Tá Vhi Healthcare ceangailte le Zurich Life Assurance plc le haghaidh Árachais Saoil de chuid Vhi agus Árachas Cosanta Morgáiste de chuid Vhi atá frithgheallta ag Zurich Life Assurance plc. Tá Vhi Healthcare ceangailte le Collinson Insurance Solutions Europe Limited le haghaidh Árachas Taistil Ilturais agus Árachas Fiaclóireachta de chuid Vhi atá frithgheallta ag Great Lakes Insurance SE agus le haghaidh Árachas Sláinte Idirnáisiúnta de chuid Vhi atá frithgheallta ag Collinson Insurance Europe Limited. Chun tuilleadh faisnéise a fháil faoi Ghrúpa Vhi, tabhair cuairt ar: https://www.vhi.ie/about-vhi. This e-mail and any files transmitted with it contain information which may be confidential and which may also be privileged and is intended solely for the use of the individual or entity to whom it is addressed. Unless you are the intended recipient you may not copy or use it, or disclose it to anyone else. Any opinions expressed are that of the individual and not necessarily that of the Vhi Group. If you have received this e-mail in error please notify the sender by return.