shashank created CAMEL-25570:
--------------------------------
Summary: camel-aws2-ddb - aws2-ddbstream consumer skips the
records of a poll it did not route when the route is suspended during the poll
Key: CAMEL-25570
URL: https://issues.apache.org/jira/browse/CAMEL-25570
Project: Camel
Issue Type: Bug
Components: camel-aws
Reporter: shashank
Assignee: shashank
{{Ddb2StreamConsumer.poll}} (main 5987a263107b) gets up to
{{maxResultsPerRequest}} records (default 100) of a shard, creates one exchange
per record and calls {{processBatch}} (line 91). After that it stores the next
shard iterator of the response and the sequence number of the last record of
the response (lines 93-96), whatever {{processBatch}} has done.
{{processBatch}} stops routing when {{isBatchAllowed()}} returns false (line
106). {{ScheduledBatchPollingConsumer.isBatchAllowed}} (lines 104-118) does
that when the consumer is no longer allowed to run (stopping or stopped), or
when {{deferShutdown}} gave it a {{shutdownRunningTask}} other than
{{CompleteAllTasks}}. Only {{DefaultShutdownStrategy}} calls {{deferShutdown}}
(lines 614-616), when it suspends or shuts down routes, with the
{{shutdownRunningTask}} of the route ({{CompleteCurrentTaskOnly}} by default).
So the batch ends after the exchange that is being routed when the route is
suspended at route level:
* {{RouteController.suspendRoute}} ({{AbstractCamelContext}} line 1651), also
when a route policy calls it ({{RoutePolicySupport.suspendRoute}}, line 156);
* {{CamelContext.suspend()}} ({{AbstractCamelContext}} line 2325).
It does not end when {{consumer.suspend()}} is called directly:
{{isRunAllowed()}} stays true for a suspended service ({{BaseService}} lines
349-351) and no {{shutdownRunningTask}} is set, so the poll that is running
routes all its records. That is what
{{RoutePolicySupport.suspendOrStopConsumer}} does (line 110), which the route
policies shipped with Camel use ({{ThrottlingInflightRoutePolicy}},
{{ThrottlingExceptionRoutePolicy}}, the quartz {{ScheduledRoutePolicy}}): they
are not affected. A route with {{shutdownRunningTask(CompleteAllTasks)}} is not
affected either.
When the route is resumed ({{resumeRoute}}, or {{CamelContext.resume()}} after
{{suspend()}}), the same consumer continues with the stored iterator, which is
after the whole response: the records that were not routed are skipped. Nothing
is logged. With one shard and the defaults that is up to 99 records per
suspend; the following shards of that poll are requested as well and lose their
whole response. If the stored iterator has expired meanwhile, the consumer
continues after {{lastSeenSequenceNumbers}}, which is also the last record of
the response.
When the route or the context is stopped, the batch is cut in the same way (the
same {{deferShutdown}} call), but that consumer is not used again: a route that
is started again gets a new consumer ({{DefaultRoute.gatherRootServices}}, line
711), which starts from the configured {{streamIteratorType}}. Nothing is
skipped there because of the stored iterator.
h3. Reproduction
New {{Ddb2StreamConsumerSuspendedBatchTest}} (3 tests) with a real route
({{aws2-ddbstream://table?streamIteratorType=FROM_START}}) and an in-memory
{{DynamoDbStreamsClient}} (one shard, iterators are positions). The processor
of one record, or the {{GetRecords}} call, suspends the route from another
thread ({{RouteController.suspendRoute}}) and waits until {{isBatchAllowed()}}
is false; then the route is resumed. On main all 3 end with an error, not a
failure ({{Tests run: 3, Failures: 0, Errors: 3}}): the assertion is inside an
Awaitility {{untilAsserted}}, so the result is a {{ConditionTimeoutException}}
that wraps it, for example {{ConditionTimeout Assertion condition defined as a
Lambda expression in
org.apache.camel.component.aws2.ddbstream.Ddb2StreamConsumerSuspendedBatchTest
records routed after the resume ==> expected: <[1, 2, 3, 4]> but was: <[1, 2]>
within 20 seconds.}} The assertion messages:
* {{recordsNotRoutedByASuspendingConsumerAreRoutedAfterTheResume}} (records
1-4, suspend while record 2 is routed): {{records routed after the resume ==>
expected: <[1, 2, 3, 4]> but was: <[1, 2]>}}.
* {{recordsOfAResponseThatArrivesWhileSuspendingAreRoutedAfterTheResume}}
(suspend while the request is in flight): {{records routed after the resume ==>
expected: <[1, 2]> but was: <[]>}}.
* {{laterRecordsAreRoutedOnceAfterABatchWasCutShort}} (suspend at record 1 of
3, record 4 stored while suspended): {{records routed after the resume ==>
expected: <[1, 2, 3, 4]> but was: <[1, 4]>}}.
h3. Proposed fix
{{poll}} compares the number of exchanges {{processBatch}} routed with the
number of records of the response. When the batch was cut short:
* it does not take the next shard iterator of the response;
* when some records were routed, it stores the sequence number of the last
routed record, and the next poll requests an iterator after it
({{ShardIteratorHandler.requestFreshShardIterator}}, {{AFTER_SEQUENCE_NUMBER}},
the call the expired-iterator path already uses). The request is made by the
next poll and not at once, because a stop interrupts the polling thread;
* when nothing was routed, the iterator of the request stays and is used again;
* the following shards are not requested by that poll.
The exchanges created for the records that are not routed are given back to the
exchange factory (on main they are dropped); the next poll creates new ones. A
pooled exchange is given back with {{done()}}, as it was created with
{{createExchange(true)}};
{{Ddb2StreamConsumerSuspendedBatchPooledExchangeTest}} runs the tests with
{{PooledExchangeFactory}} and checks the pool statistics (every exchange taken
is given back once).
4.24 upgrade-guide entry (the skipped records are now routed after the resume).
No new option. camel-aws2-ddb unit tests with the fix: 75, 0 failures
(integration tests need LocalStack and were not run).
Other consumers with the same shape (a cursor that is moved for the whole poll,
and a {{processBatch}} loop on {{isBatchAllowed()}}), read on main and not run:
{{SlackConsumer}} (the timestamp is set in {{createExchanges}}, lines 152 and
161, before {{processBatch}}, line 175) and {{GoogleCalendarStreamConsumer}}
({{selectUndeliveredAndMoveCursor}} is called at line 147, before
{{processBatch}}, line 220). They are in other modules and not part of this
change. {{Kinesis2Consumer.processBatch}} has no {{isBatchAllowed()}} check and
routes every record of the response.
Found with a TLA+ model of a poll (GetRecords, route one record at a time while
the batch is allowed, end of the batch) with a suspend and a resume as
environment steps: on main TLC violates {{NoSkip}} ("between two polls the
consumer has not moved past a record it did not route") in 4 states (GetRecords
returns 3 records, the route is suspended, the batch ends, the iterator is
after record 3); it holds with the fix (also with an expired iterator after the
suspend). Then confirmed with the real consumer and a real route as described.
Affected: main; camel-4.22.x, camel-4.18.x and camel-4.14.x have the same
{{processBatch}} loop and the same update after it (read, not run).
Duplicate check (2026-10-12): JIRA text "ddbstream suspend" (none), "ddbstream
isBatchAllowed" (none), "Ddb2StreamConsumer" (1: CAMEL-24156, other items),
"ddbstream" created since 2025-06-01 (2: CAMEL-24156 and CAMEL-24154,
resharding); GitHub pull requests: #24861 (merged 2026-07-17, CAMEL-24154) is
the pull request that carried the ddbstream work of that review (it changes
{{ShardIteratorHandler}} and its tests), and "Ddb2StreamConsumer" in title or
body finds #6151 (closed, CAMEL-16594); neither is about a suspended route.
#24863 (merged 2026-07-18, CAMEL-24156) changes no ddbstream file (in
camel-aws2-ddb only {{Ddb2JsonDataTypeTransformer}}); no open pull request
touches {{Ddb2StreamConsumer}}.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
