Hi Narsi, I've just created the Jira: CAMEL-17592 <https://issues.apache.org/jira/browse/CAMEL-17592>
On Wed, Feb 2, 2022 at 11:35 PM Narsi Reddy Nallamilli < narsi.nallami...@gmail.com> wrote: > Hi Arnaud, > > Yup, It used to work before because I use it in my project and was not > aware of the bug till you reported. > > Can you open up a jira for the same? > > On Thu, 3 Feb, 2022, 01:00 Arnaud Level, <arn...@toro-intl.com> wrote: > > > Hi again Narsi, > > > > I read more about the *delay* component and you are right! Using it with > a > > custom (or even the default) thread pool profile we can have the expected > > behavior without the thread pool being exhausted as I was thinking. This > is > > a workaround to the concurrentConsumer bug not being taken into account. > > I still think we should open an issue as this behavior is supposed to be > > supported by the concurrentConsumer URI parameter of the SQS component. > > What do you think? > > > > Thanks for your help again! > > Arnaud > > > > > > > > > > > > On Wed, Feb 2, 2022 at 6:42 PM Arnaud Level <arn...@toro-intl.com> > wrote: > > > > > Hi Narsi, > > > > > > Thank you for the reply! Indeed it works but not exactly as expected. I > > am > > > not sure to understand the behavior because with this delay it is not > > > limited to the number of concurrent consumers. If I set > > > concurrentConsumers=3 and I send 5 messages (which are read 1 by 1 > with 5 > > > pollings) I will see the 5 messages being consumed immediately even > > though > > > they all are entering in a sleep of 5 secs each. > > > The expected behavior would be to see 3 immediately and then the next 2 > > > when two of the first 3 wakes up (=after the 5 s sleep). > > > > > > From a behavior perspective, it looks like a new thread is created > every > > > time a message is available on the queue but without any limit. If this > > is > > > the case we can easily end up with an exhausted thread pool. I had a > > quick > > > look at the delay component and it is by default async which would > > explain > > > the behavior we are seeing. What do you think ? > > > > > > Regarding the concurrentConsumers parameter, if this is a bug, I > believe > > > it is a regression because according to this article > > > < > > > https://blog.christianposta.com/camel/very-fast-camels-and-cloud-messaging/ > > > > it > > > has worked in the past. Shouldn't we open an issue ? > > > > > > Thanks again! > > > Arnaud > > > > > > On Wed, Feb 2, 2022 at 5:52 PM Narsi Reddy Nallamilli < > > > narsi.nallami...@gmail.com> wrote: > > > > > >> Hi Arnaud, > > >> > > >> Yes, you are correct, 'concurrentConsumers; uri attribute is not > working > > >> as > > >> expected and this is some bug. > > >> > > >> However I found when you use delay in the route it works as expected. > > >> > > >> > > >> > > > 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(); > > >> }}); > > >> > > >> try it and let me know. > > >> > > >> On Mon, Jan 31, 2022 at 11:48 PM Arnaud Level <arn...@toro-intl.com> > > >> wrote: > > >> > > >> > Hi Larry, > > >> > > > >> > Thank you for your message! Your reply makes sense to me and I've > > tried > > >> > what you suggested to test it with a queue like: > > >> > > > >> > > > >> > > > from("aws2-sqs://aramark-notifications?maxMessagesPerPoll=1&concurrentConsumers=5&amazonSQSClient=#sqsClient") > > >> > but it does not change, I still see the messages with a delay of 5 > > >> seconds > > >> > between each of them. > > >> > > > >> > If I turn on the trace with > > >> > (logging.level.org.apache.camel.component.aws2.sqs=TRACE). I see > that > > >> the > > >> > next polling is only triggered after > > >> > the Delete message of a consumed message is sent. And in the logs I > > see > > >> "1 > > >> > message received" but no polling will occur until the Delete message > > is > > >> > sent which occurs only after the sleep of 5 secs. Only 1 message is > > read > > >> > per polling and I should see at least 5 of them since > > >> > concurrentConsumers=5. > > >> > It looks to me there is still only 1 concurrent consumer and I don't > > >> > understand why. > > >> > > > >> > Arnaud > > >> > > > >> > On Mon, Jan 31, 2022 at 3:27 PM Larry Shields < > > larry.shie...@gmail.com> > > >> > wrote: > > >> > > > >> > > Hi Arnaud, > > >> > > > > >> > > I think what may be happening is that you first consumer is > grabbing > > >> all > > >> > of > > >> > > the messages from the queue that are available. The default > message > > >> poll > > >> > > size is unlimited so your other 4 consumers are polling for > nothing > > on > > >> > the > > >> > > queue. What you might want to try is setting the > maxMessagesPerPoll > > >> to a > > >> > > value. This will reduce the number of messages a consumer > retrieves > > >> for > > >> > > each poll. > > >> > > > > >> > > The other thing I have done is set Greedy=true when using > concurrent > > >> > > consumers if you're looking to increase throughput. This will > cause > > >> the > > >> > > consumer to make another poll immediately without waiting for the > > next > > >> > > delay if the previous poll returned at least 1 message. > > >> > > > > >> > > Example: > > >> > > > > >> > > > > >> > > > >> > > > aws-sqs://my-queue?greedy=true&maxMessagesPerPoll=10&concurrentConsumers=5 > > >> > > > > >> > > So in this example, each consumer will pull no more than 10 > messages > > >> per > > >> > > poll to the SQS Queue. So if there are less than 10 messages on > > your > > >> > > queue, the other consumers aren't going to really get an messages. > > >> > > > > >> > > On Mon, Jan 31, 2022 at 7:15 AM Arnaud Level < > arn...@toro-intl.com> > > >> > wrote: > > >> > > > > >> > > > Hi Narsi, > > >> > > > It's a fifo but I actually tried the simple code I've posted on > a > > >> > > standard > > >> > > > one too and observed the same behavior. > > >> > > > > > >> > > > On Mon, Jan 31, 2022 at 1:11 PM Narsi Reddy Nallamilli < > > >> > > > narsi.nallami...@gmail.com> wrote: > > >> > > > > > >> > > > > Hi Arnaud, > > >> > > > > > > >> > > > > Is your AWS queue type fifo or standard? > > >> > > > > > > >> > > > > On Mon, 31 Jan, 2022, 17:31 Arnaud Level, < > arn...@toro-intl.com > > > > > >> > > wrote: > > >> > > > > > > >> > > > > > Hi, > > >> > > > > > > > >> > > > > > (Camel version: camel-aws2-sqs-starter: 3.12.0) > > >> > > > > > > > >> > > > > > I am trying to use and understand concurentConsumers with a > > SQS > > >> > > queue: > > >> > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > > 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 I send 3 messages at the same > time, I > > >> have > > >> > > to > > >> > > > > wait > > >> > > > > > 5 seconds to see the second message ("Message received...") > > and > > >> 5 > > >> > > more > > >> > > > > > seconds to see the third one. My understanding of > > >> > concurentConsumers > > >> > > is > > >> > > > > > that with a value of 5 I would see the 3 messages at the > same > > >> time > > >> > > > since > > >> > > > > 3 > > >> > > > > > threads will consume them in parallel. If I add the > > Thread.sleep > > >> > in a > > >> > > > > seda > > >> > > > > > route, I'm having this behavior (= The 3 messages are read > at > > >> the > > >> > > same > > >> > > > > > time). > > >> > > > > > > > >> > > > > > 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). > > >> > > > > > > > >> > > > > > I would understand the above behavior with > > concurentConsumers=1 > > >> > but I > > >> > > > > don't > > >> > > > > > with concurentConsumers=5. Could someone tell me what I've > > >> > > > misunderstood > > >> > > > > ? > > >> > > > > > > > >> > > > > > Thank you in advance! > > >> > > > > > Arnaud > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > > >> > -- > > >> > > > >> > > >> > > >
