[
https://issues.apache.org/jira/browse/CAMEL-25570?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-25570:
--------------------------------
Fix Version/s: 4.24.0
> 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
> Priority: Major
> Fix For: 4.24.0
>
>
> {{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)
