[jira] [Commented] (CAMEL-17592) concurrentConsumers URI parameter not working with aws2-sqs endpoint

["Narsi Reddy Nallamilli (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/CAMEL-17592?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17486435#comment-17486435 ] 

Narsi Reddy Nallamilli commented on CAMEL-17592:
------------------------------------------------

[~arnaudlbcn] The expected behavior which you are seeking is changed with https://issues.apache.org/jira/browse/CAMEL-16383. 

By default the consumer creates a defaultScheduler with only one thread which is why  the messages are consumed serially. Before the change the scheduler thread are created based on concurrentConsumers attribute value, that is why it used to work.

You need to specify your own scheduler with 'scheduledExecutorService' as shown in  [https://camel.apache.org/components/3.7.x/aws2-sqs-component.html] with required pool size to work the way you want.

However I have tried this and it is not honoring the attribute. The issue is in Sqs2Endpoint where it creates the consumer with default schedule. It should consider the provided custom schedule if exists.

 
{code:java}
public Consumer createConsumer(Processor processor) throws Exception {
    Sqs2Consumer sqsConsumer = new Sqs2Consumer(this, processor);
    this.configureConsumer(sqsConsumer);
    sqsConsumer.setMaxMessagesPerPoll(this.maxMessagesPerPoll);
    DefaultScheduledPollConsumerScheduler scheduler = new DefaultScheduledPollConsumerScheduler(); {code}
 

 

[~davsclaus]  can you please have a look?

> concurrentConsumers URI parameter not working with aws2-sqs endpoint
> --------------------------------------------------------------------
>
>                 Key: CAMEL-17592
>                 URL: https://issues.apache.org/jira/browse/CAMEL-17592
>             Project: Camel
>          Issue Type: Bug
>          Components: camel-aws
>    Affects Versions: 3.13.0
>         Environment: <dependency>
> <groupId>org.apache.camel.springboot</groupId>
> <artifactId>camel-aws2-sqs-starter</artifactId>
> <version>3.13.0</version>
> </dependency>
>            Reporter: Arnaud Level
>            Priority: Major
>
> The concurrentConsumers URI parameter is not taken into account by the endpoint.
> We can reproduce the issue with the following prototype:
> from("aws2-sqs://queuexxx?concurrentConsumers=5&amazonSQSClient=#sqsClient&
> waitTimeSeconds=20")
> .process(exchange -> {
>      System.out.println("Message received...");
> })
> .process(exchange -> {
>      try {
>         Thread.sleep(5000);
>      } catch (InterruptedException e) {
>         e.printStackTrace();
>      }});
> With the above queue, if we send 3 messages, we have to wait
> 5 seconds to see the second message ("Message received...") and 5 more
> seconds to see the third one. The expected behavior is to see the 3 messages
> consumed without delay since 3 threads will consume them in parallel.
> Turning on the Camel logs it seems that the next polling is done only after
> the Delete for the previous message is sent (which is with a delay of 5s).
> The messages are read 1 by 1 (=1 message per polling) so the above behavior is not seen because the message are consumed together by one polling and continue in the same thread.
> Note that, as commented by Narsi Reddy Nallamilli in the user mailing list, we obtain the expected behavior by adding a delay in the route as follow: 
> from("aws2-sqs://queuexxx?concurrentConsumers=5&amazonSQSClient=#sqsClient&
> waitTimeSeconds=20")
> .delay(1L)
> .process(exchange -> {
>     System.out.println("Message received...");
> })
> .process(exchange -> {
>     try {
>        Thread.sleep(5000);
>     } catch (InterruptedException e) {
>        e.printStackTrace();
>     }});



--
This message was sent by Atlassian Jira
(v8.20.1#820001)