[
https://issues.apache.org/jira/browse/CAMEL-26649?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-26649 started by Federico Mariani.
------------------------------------------------
> camel-aws2-sqs - consumer receives one message per ReceiveMessage and deletes
> messages one by one by default
> ------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-26649
> URL: https://issues.apache.org/jira/browse/CAMEL-26649
> Project: Camel
> Issue Type: Improvement
> Components: camel-aws2-sqs
> Affects Versions: 4.18.5, 4.22.2, 4.23.0
> Reporter: Federico Mariani
> Assignee: Federico Mariani
> Priority: Major
>
> h3. Problem
> By default the {{aws2-sqs}} consumer asks SQS for one message per
> {{ReceiveMessage}} call, and it deletes every routed message with its own
> synchronous {{DeleteMessage}} call. Line numbers are on main (7978db3d42dd):
> *
> {{components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java}}
> line 104: {{maxMessagesPerPoll}} is an {{int}} with no default (0). Its
> javadoc (line 437) and the catalog say "Is default unlimited".
> * {{Sqs2Consumer.java}} line 693: {{Math.max(1,
> endpoint.getMaxMessagesPerPoll())}}, so 0 or a negative value means 1, and
> line 797 sends {{maxNumberOfMessages(1)}}. Values above 10 are already split
> into receives of at most 10 (CAMEL-21031).
> * {{Sqs2Configuration.java}} line 78: {{waitTimeSeconds}} has no default, so
> no {{WaitTimeSeconds}} is sent. The queue attribute
> {{ReceiveMessageWaitTimeSeconds}} then applies, which means short polling
> unless the queue sets it above 0.
> * {{Sqs2Consumer.java}} lines 263-270: {{processCommit}} calls
> {{deleteMessage}} once per exchange. The visibility extender already batches,
> with {{ChangeMessageVisibilityBatch}} (lines 524-544).
> With the defaults ({{delay=500}}, {{greedy=false}}) a consumer thread gets at
> most about 2 messages per second, at 2 billable requests per message. Yet SQS
> returns up to 10 messages per receive (AWS SDK 2.57.7 service model: "Valid
> values: 1 to 10. Default: 1."), and {{DeleteMessageBatch}} takes up to 10
> entries. The effective default has always been 1: before CAMEL-21031 the
> consumer did not send {{MaxNumberOfMessages}} at all, and the SQS default is
> 1. The words "default unlimited" come from the generic batch consumer and are
> wrong here.
> h3. How to reproduce
> These numbers are measured. The route is
> {{from("aws2-sqs://MyQueue?amazonSQSClient=#client&initialDelay=0" +
> options)}}, run on the module's {{AmazonSQSClientMock}} with 30 messages
> queued; it records every request and stops once N messages have been routed:
> ||options||messages routed||ReceiveMessage
> calls||MaxNumberOfMessages||exchanges per poll||DeleteMessage calls||
> |(defaults)|5|5|1|1|5, in 2.1 s|
> |{{maxMessagesPerPoll=0}} or {{-1}}|2|2|1|1|2|
> |{{greedy=true}}|30|31|1|1|30|
> |{{concurrentConsumers=3}}|6|6|1|1|6|
> |{{maxMessagesPerPoll=10}} (control)|30|3|10|10|30|
> |{{maxMessagesPerPoll=25}} (control)|25|3|10, 10, 5|25|25|
> No request carried {{WaitTimeSeconds}}, and {{DeleteMessageBatch}} was never
> called. {{greedy}} only removes the delay between polls, and
> {{concurrentConsumers}} only adds threads; neither changes the batch size.
> h3. Proposed change
> # Default {{maxMessagesPerPoll}} to 10. Fix the javadoc and catalog text: 0
> or a negative value means one message per receive, and above 10 the consumer
> sends several receives of at most 10.
> # Add an opt-in option that collects the receipt handles of completed
> exchanges and sends {{DeleteMessageBatch}} on the existing sync
> {{SqsClient}}. It flushes at 10 handles, at the end of the poll and on
> stop/suspend, and reports failed entries one by one as today. The SDK's
> {{SqsAsyncBatchManager}} does not fit: it needs an {{SqsAsyncClient}}
> ({{Builder.client(SqsAsyncClient)}}), and {{batchManager()}} is generated
> only on the async client.
> # Optionally, document long polling ({{waitTimeSeconds}}), or default to it.
> With short polling SQS samples only some of its servers and can return fewer
> than 10 messages while more are queued. So (1) alone gives up to 10x fewer
> receives, not a guaranteed 10x; long polling is what makes full batches
> reliable.
> Risks:
> * Visibility timeout vs. a batch of 10. The 10 messages of a poll are routed
> one after the other, and the visibility timeout of each one starts at the
> receive. With a slow synchronous route and without
> {{extendMessageVisibility}}, the last messages of a batch can become visible
> again and be delivered twice. With {{extendMessageVisibility=true}} the whole
> batch is extended from the receive since CAMEL-25700 (4.24.0).
> * Deferred deletes widen the window in which a crash leaves already processed
> messages to be delivered again. The extender must keep extending a message
> until its delete is flushed.
> Estimates (not measured against AWS): with full batches, (1) and (2) bring
> billable requests from 2 down to about 0.2 per message.
> h3. Impact / upgrade
> * (1) changes a default, so it needs a {{camel-aws2-sqs}} entry in the 4.24
> upgrade guide. The entry should say that each poll now takes up to 10
> messages, repeat the visibility-timeout point above, and give
> {{maxMessagesPerPoll=1}} to keep the old behaviour.
> * (2) is opt-in and has no upgrade impact.
> * camel-kamelets also needs a change: the {{aws-sqs-source}} Kamelet
> (camel-kamelets 4.23.0) declares {{maxMessagesPerPoll}} with {{default: 1}}
> and {{maximum: 10}} and always passes it. Kamelet users therefore keep one
> message per receive whatever the component default is, and the maximum of 10
> is out of date since CAMEL-21031 started splitting larger values.
> * The same code is on camel-4.18.5, camel-4.22.2 and camel-4.23.0.
> h3. Related
> * CAMEL-21031 (1432192c6ca3) added polling of more than 10 messages and
> {{Math.max(1, ...)}}; its description notes that the consumer receives only 1
> message by default.
> * CAMEL-7590 (2014, resolved as Information Provided) asked for batch
> write/delete through the SDK v1 buffered async client.
> * CAMEL-19656 batched the visibility extender; CAMEL-25700 extends the whole
> batch with {{extendMessageVisibility}}.
> * CAMEL-9592 documented {{maxMessagesPerPoll}} together with
> {{concurrentConsumers}}.
> _Claude Code on behalf of Federico Mariani_
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
