Federico Mariani created CAMEL-26669:
----------------------------------------
Summary: camel-aws2-kinesis - the polling consumer routes the
whole GetRecords response before a stop or suspend
Key: CAMEL-26669
URL: https://issues.apache.org/jira/browse/CAMEL-26669
Project: Camel
Issue Type: Improvement
Components: camel-aws2-kinesis
Affects Versions: 4.18.5, 4.23.0, 4.22.2
Reporter: Federico Mariani
h3. Problem
The polling {{aws2-kinesis}} consumer ({{useKclConsumers=false}}) routes every
record of a {{GetRecords}} response before its poll ends, also when the route
is stopped or suspended meanwhile. Line numbers are on main (7978db3d42dd):
*
{{components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java}}
lines 244-256: {{processBatch}} loops {{while (!exchanges.isEmpty())}} and
never calls {{isBatchAllowed()}}.
*
{{core/camel-support/src/main/java/org/apache/camel/support/ScheduledBatchPollingConsumer.java}}
lines 103-118: {{isBatchAllowed()}} returns false once
{{DefaultShutdownStrategy}} has called {{deferShutdown}} with the
{{shutdownRunningTask}} of the route, which is {{CompleteCurrentTaskOnly}} by
default. The other batch consumers (aws2-ddbstream, aws2-sqs, aws2-s3, file,
...) stop their batch there, as the graceful shutdown documentation describes
for batch consumers.
* Same file, lines 58-78: while a poll runs, {{getPendingExchangesSize()}}
returns at least 1 ({{isPolling()}}), so the graceful shutdown waits for the
whole poll. Suspending the consumer does not stop the poll
({{ScheduledPollConsumer.doSuspend}} is empty).
*
{{core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java}}
line 129: the timeout is 45 seconds by default
({{camel.main.shutdownTimeout}}); after it the routes are shut down by force.
Since CAMEL-26650 the consumer asks each shard for up to 1000 records per poll
(it was 1). With a route that takes 50 ms per record, one poll takes about 50
seconds, more than the default timeout, so a plain stop of the route or of the
CamelContext becomes a forced shutdown. The records of the poll that are not
routed when the timeout fires then fail one by one with
{{RouteStoppingException}}, each reported to the consumer's exception handler.
h3. How to reproduce
Measured on main + PR #29272 with a Mockito {{KinesisClient}}: each shard holds
5000 records, {{GetRecords}} returns {{min(Limit, remaining)}} records. The
route is
{{from("aws2-kinesis://stream?amazonKinesisClient=#kinesisClient&exceptionHandler=#handler"
+ options).delay(20).syncDelayed()}}, the shutdown timeout is set to 5
seconds, and {{stopRoute}} is called once the first record has been routed.
||options||shards||stopRoute took||timeout / forced||records routed||records
failed after the stop||
|(default, Limit 1000)|1|5.0 s|yes|209 of 1000|791 (1 {{InterruptedException}},
790 {{RouteStoppingException}})|
|(default, Limit 1000)|4|5.0 s|yes|823 of 4000|3177 (1
{{InterruptedException}}, 3176 {{RouteStoppingException}})|
|{{maxResultsPerRequest=1}} (control, the old default)|1|0 s|no|1 of 1|0|
The log shows "Timeout occurred during graceful shutdown. Forcing the routes to
be shutdown now." in both default cases. The 45 s / 50 ms case of the review
was computed, not run. A suspend goes through the same wait of
{{DefaultShutdownStrategy}}; that was read from the source, not run.
h3. Proposed change
Do what CAMEL-25570 did for {{Ddb2StreamConsumer}} (commit 7024148c09c3):
# {{processBatch}} stops when {{isBatchAllowed()}} is false.
# In {{fetchAndPrepareRecordsForCamel}} (lines 191-204), when fewer records
were routed than received:
#* do not store {{nextShardIterator}}, and remove the cached iterator, so that
the next poll requests a new one. A null {{nextShardIterator}} (closed shard)
must not be stored either, or the shard is marked closed before its last
records are routed.
#* some were routed: put the sequence number of the last *routed* record in
{{lastSequenceNumbers}}. The next poll then requests an
{{AFTER_SEQUENCE_NUMBER}} iterator through the existing path in
{{getShardIterator}} (lines 282-288, CAMEL-25569).
#* none was routed and the shard has no {{lastSequenceNumbers}} entry yet:
remember the first record, and request {{AT_SEQUENCE_NUMBER}} for it.
CAMEL-25570 added the same position to aws2-ddbstream
({{firstNotRoutedSequenceNumbers}}). This does not rely on using a shard
iterator twice, which the Kinesis API documentation does not describe.
#* release the exchanges that were not routed ({{releaseExchange(exchange,
true)}}), for the pooled exchange factory.
Both parts are needed. Only adding the {{isBatchAllowed()}} check would make a
{{suspendRoute}} / {{resumeRoute}} skip the rest of the response, because the
same consumer continues from the stored iterator (the bug CAMEL-25570 fixed for
aws2-ddbstream). After a stop and start the route gets a new consumer
({{DefaultRoute.gatherRootServices}}), which starts from {{iteratorType}} or
the resume strategy, as today.
Size and risk (estimated from the ddbstream fix, which changed about 60 lines
of main code): about 30-50 lines in {{Kinesis2Consumer}}, plus Mockito tests:
suspend in the middle of a response and resume (no record skipped or routed
twice), a cut before the first record, a closed shard cut short, an expired
iterator after a cut, pooled exchanges released. Points to watch:
* the shards of a poll run in parallel, so each shard cuts at its own record.
The state is already per shard and thread-safe (CAMEL-24080).
* a resume strategy's adapter runs after the {{AFTER_SEQUENCE_NUMBER}} request
is built (line 305) and can change it, as it already can after an expired
iterator.
* {{pendingExchanges}} is one field shared by the parallel shards: leave it
unset. While polling, {{getPendingExchangesSize()}} already returns 1.
h3. Impact / upgrade
* The stop and suspend behaviour changes for routes with the default
{{CompleteCurrentTaskOnly}}: the consumer stops after the record it is routing,
and does not route the rest of the response first. This needs an upgrade-guide
entry. Routes that need the old behaviour set
{{shutdownRunningTask(CompleteAllTasks)}}, and for them nothing changes.
* Only the polling consumer is affected. The KCL consumer
({{KclKinesis2Consumer}}) extends {{DefaultConsumer}}.
* The same code is in camel-4.18.5, camel-4.22.2 and camel-4.23.0, where the
default of 1 record per poll keeps the effect small unless
{{maxResultsPerRequest}} is raised.
h3. Related
* CAMEL-26650 (PR #29272): the new default of 1000 records per shard per poll.
Its review raised this point; the upgrade guide of that PR now mentions the
shutdown timeout.
* CAMEL-25570 (commit 7024148c09c3): the same fix for aws2-ddbstream.
CAMEL-25684 (camel-slack) is similar.
* CAMEL-25569 (commit 043ac9ec84ab): {{lastSequenceNumbers}} and the
{{AFTER_SEQUENCE_NUMBER}} iterator after an expired iterator.
* The follow-up about the {{parallelStream()}} on
{{ForkJoinPool.commonPool()}}, filed at the same time.
_Claude Code on behalf of Federico Mariani_
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
