Concurrency issue with SEDA component

[vknfh3 <vi...@gmail.com>]

Hi,

I have been facing concurrency issues with SEDA component when using it for
parallely processing 200 or more JSON Messages. Our application should be
able to process and save at least 200 JSONS at a time from each user
simultaneously. The SEDA component route seems to be working fine when only
one user is issuing 200 JSON requests(which will be enclosed in a
JSONArray). I'm running into issues when another user is hitting the same
SEDA route at the same time when the old request from other user is still in
progress. Due to this the output JSON messages are being duplicated and in
some instances getting more JSON messages back than expected!!

Before using SEDA component I tried using split and parallellProcessing and
I ran into similar kind of concurrency issues.
Here is what I have:

from("seda:savesupplyitemsplitter?timeout=0&waitForTaskToComplete=Always&concurrentConsumers=200")
        		.routeId("SaveSupplyItemSplitter")
				.setProperty(SupplyItemRoutesConstants.JSON_ARRAY_MSG, body())
				.process(new Processor() {
					@Override
					public void process(Exchange exchange) throws Exception { 
						List<JSONObject> supplyItemList = (List<JSONObject>)
exchange.getIn().getBody(List.class);
						exchange.setProperty(SupplyItemRoutesConstants.JSON_ARRAY_MSG,
supplyItemList);
						if(!supplyItemList.isEmpty() || supplyItemList.size()!=0){
							JSONObject sItem = supplyItemList.get(0);   //Get the first element
from the List
							exchange.getOut().setBody(sItem); //set the supplyItem as body to the
exchange
						}
					}
				})
				.process(new SupplierNumberProcessor()) 
				.enrich(SupplyItemRoutesConstants.GET_SUPPLIER_GLN_ENDPOINT, new
SupplyItemGLNAggregator())
				.split(simple(SupplyItemRoutesConstants.BODY_STRING), new
SupplyItemDataAggregator())
				.to("direct:savesupplyitem")
           .end();

from("direct:savesupplyitem")    //complex route with lot of external
service calls and calls to other camel enpoints. This route also has lot
processors and aggregators. All the routes that this endpoint calls are
"direct" end points
    . to(some complex processing and saving to database)
.end();   

I have uploaded the code for the above route (direct:savesupplyitem) as
separate file for your reference:
SaveSupplyItemRoutes.txt
<http://camel.465427.n5.nabble.com/file/n5750407/SaveSupplyItemRoutes.txt>  

Do I need to do anything to make it work for multiple users (more than one
user will call the seda endpoint simultaneously)? Am I missing something
here ? Please suggest.

Thanks



--
View this message in context: http://camel.465427.n5.nabble.com/Concurrency-issue-with-SEDA-component-tp5750407.html
Sent from the Camel Development mailing list archive at Nabble.com.

Re: Concurrency issue with SEDA component

[Willem Jiang <wi...@gmail.com>]

I just have a quick look of your camel route, it looks like you store the message body into the exchange property some place. If you do the same thing before the splitter,(splitter need to copy the exchange for you) you may have some trouble as the exchanges are sharing some message body across different thread.

--  
Willem Jiang

Red Hat, Inc.
Web: http://www.redhat.com
Blog: http://willemjiang.blogspot.com (English)
http://jnn.iteye.com (Chinese)
Twitter: willemjiang  
Weibo: 姜宁willem



On April 22, 2014 at 7:31:39 AM, vknfh3 (vinaykumar258@gmail.com) wrote:
> Hi,
>  
> I have been facing concurrency issues with SEDA component when using it for
> parallely processing 200 or more JSON Messages. Our application should be
> able to process and save at least 200 JSONS at a time from each user
> simultaneously. The SEDA component route seems to be working fine when only
> one user is issuing 200 JSON requests(which will be enclosed in a
> JSONArray). I'm running into issues when another user is hitting the same
> SEDA route at the same time when the old request from other user is still in
> progress. Due to this the output JSON messages are being duplicated and in
> some instances getting more JSON messages back than expected!!
>  
> Before using SEDA component I tried using split and parallellProcessing and
> I ran into similar kind of concurrency issues.
> Here is what I have:
>  
> from("seda:savesupplyitemsplitter?timeout=0&waitForTaskToComplete=Always&concurrentConsumers=200")  
> .routeId("SaveSupplyItemSplitter")
> .setProperty(SupplyItemRoutesConstants.JSON_ARRAY_MSG, body())
> .process(new Processor() {
> @Override
> public void process(Exchange exchange) throws Exception {
> List supplyItemList = (List)
> exchange.getIn().getBody(List.class);
> exchange.setProperty(SupplyItemRoutesConstants.JSON_ARRAY_MSG,
> supplyItemList);
> if(!supplyItemList.isEmpty() || supplyItemList.size()!=0){
> JSONObject sItem = supplyItemList.get(0); //Get the first element
> from the List
> exchange.getOut().setBody(sItem); //set the supplyItem as body to the
> exchange
> }
> }
> })
> .process(new SupplierNumberProcessor())
> .enrich(SupplyItemRoutesConstants.GET_SUPPLIER_GLN_ENDPOINT, new
> SupplyItemGLNAggregator())
> .split(simple(SupplyItemRoutesConstants.BODY_STRING), new
> SupplyItemDataAggregator())
> .to("direct:savesupplyitem")
> .end();
>  
> from("direct:savesupplyitem") //complex route with lot of external
> service calls and calls to other camel enpoints. This route also has lot
> processors and aggregators. All the routes that this endpoint calls are
> "direct" end points
> . to(some complex processing and saving to database)
> .end();
>  
> I have uploaded the code for the above route (direct:savesupplyitem) as
> separate file for your reference:
> SaveSupplyItemRoutes.txt
>  
>  
> Do I need to do anything to make it work for multiple users (more than one
> user will call the seda endpoint simultaneously)? Am I missing something
> here ? Please suggest.
>  
> Thanks
>  
>  
>  
> --
> View this message in context: http://camel.465427.n5.nabble.com/Concurrency-issue-with-SEDA-component-tp5750407.html  
> Sent from the Camel Development mailing list archive at Nabble.com.
>