This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 7024148c09c3 CAMEL-25570: camel-aws2-ddb - aws2-ddbstream consumer
must not skip the records it did not route when the route is suspended (#27754)
7024148c09c3 is described below
commit 7024148c09c3c5bf6a1d45b193ccbd178e0eef8b
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 13 15:02:24 2026 +0530
CAMEL-25570: camel-aws2-ddb - aws2-ddbstream consumer must not skip the
records it did not route when the route is suspended (#27754)
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../aws2/ddbstream/Ddb2StreamConsumer.java | 53 ++-
.../aws2/ddbstream/ShardIteratorHandler.java | 9 +
...amConsumerSuspendedBatchPooledExchangeTest.java | 73 +++++
.../Ddb2StreamConsumerSuspendedBatchTest.java | 357 +++++++++++++++++++++
4 files changed, 489 insertions(+), 3 deletions(-)
diff --git
a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java
b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java
index 9ab926d98b76..a0ac9df98c1c 100644
---
a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java
+++
b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java
@@ -18,10 +18,12 @@ package org.apache.camel.component.aws2.ddbstream;
import java.util.ArrayDeque;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
+import java.util.Set;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
@@ -43,6 +45,12 @@ public class Ddb2StreamConsumer extends
ScheduledBatchPollingConsumer {
private final ShardIteratorHandler shardIteratorHandler;
private final Map<String, String> lastSeenSequenceNumbers = new
HashMap<>();
+ // Shards of which only the first records of a response were routed, as
the consumer was being suspended or
+ // stopped: the next poll continues after the last routed record instead
of after the response
+ private final Set<String> shardsOfUnfinishedBatch = new HashSet<>();
+ // Shards of which no record was routed yet, and whose first response with
records was not routed at all for the
+ // same reason: the sequence number of the first record of that response
+ private final Map<String, String> firstNotRoutedSequenceNumbers = new
HashMap<>();
public Ddb2StreamConsumer(Ddb2StreamEndpoint endpoint, Processor
processor) {
this(endpoint, processor, new ShardIteratorHandler(endpoint));
@@ -67,6 +75,11 @@ public class Ddb2StreamConsumer extends
ScheduledBatchPollingConsumer {
getEndpoint().getConfiguration().getMaxResultsPerRequest()
/ shardIterators.size());
String shardId = shardIteratorEntry.getKey();
String shardIterator = shardIteratorEntry.getValue();
+ if (shardsOfUnfinishedBatch.contains(shardId)) {
+ // not requested when the batch was cut short: that thread may
have been interrupted by the stop
+ shardIterator =
shardIteratorHandler.requestFreshShardIterator(shardId,
lastSeenSequenceNumbers.get(shardId));
+ shardsOfUnfinishedBatch.remove(shardId);
+ }
GetRecordsResponse result;
try {
GetRecordsRequest req = GetRecordsRequest.builder()
@@ -76,9 +89,21 @@ public class Ddb2StreamConsumer extends
ScheduledBatchPollingConsumer {
result = getEndpoint().getClient().getRecords(req);
} catch (ExpiredIteratorException e) {
String lastSeenSequenceNumber =
lastSeenSequenceNumbers.get(shardId);
- LOG.warn("Expired Shard Iterator, attempting to resume from
{}", lastSeenSequenceNumber, e);
+ String firstNotRoutedSequenceNumber
+ = lastSeenSequenceNumber == null ?
firstNotRoutedSequenceNumbers.get(shardId) : null;
+ String freshShardIterator;
+ if (firstNotRoutedSequenceNumber != null) {
+ // the iterator that was kept for the records that were
not routed has expired, and there is no
+ // routed record to continue after: start at the first of
them, not at the start of the shard
+ LOG.warn("Expired Shard Iterator, attempting to resume at
{}", firstNotRoutedSequenceNumber, e);
+ freshShardIterator
+ =
shardIteratorHandler.requestFreshShardIteratorAt(shardId,
firstNotRoutedSequenceNumber);
+ } else {
+ LOG.warn("Expired Shard Iterator, attempting to resume
from {}", lastSeenSequenceNumber, e);
+ freshShardIterator =
shardIteratorHandler.requestFreshShardIterator(shardId, lastSeenSequenceNumber);
+ }
GetRecordsRequest req = GetRecordsRequest.builder()
-
.shardIterator(shardIteratorHandler.requestFreshShardIterator(shardId,
lastSeenSequenceNumber))
+ .shardIterator(freshShardIterator)
.limit(limitPerRecordsRequest)
.build();
result = getEndpoint().getClient().getRecords(req);
@@ -88,11 +113,33 @@ public class Ddb2StreamConsumer extends
ScheduledBatchPollingConsumer {
for (Record polledRecord : records) {
exchanges.add(createExchange(polledRecord));
}
- processedExchangeCount += processBatch(CastUtils.cast(exchanges));
+ int routed = processBatch(CastUtils.cast(exchanges));
+ processedExchangeCount += routed;
+
+ if (routed < records.size()) {
+ // the route is being suspended or stopped, and processBatch
did not route the rest of the records:
+ // do not move past them, so that they are routed when the
consumer runs again. When nothing was
+ // routed, the shard iterator of this request is kept and used
again.
+ if (routed > 0) {
+ lastSeenSequenceNumbers.put(shardId, records.get(routed -
1).dynamodb().sequenceNumber());
+ firstNotRoutedSequenceNumbers.remove(shardId);
+ shardsOfUnfinishedBatch.add(shardId);
+ } else if (!lastSeenSequenceNumbers.containsKey(shardId)) {
+ // only needed when the kept iterator has expired before
the consumer runs again
+ firstNotRoutedSequenceNumbers.put(shardId,
records.get(0).dynamodb().sequenceNumber());
+ }
+ // processBatch removes the exchanges it routes: those left
are never routed, and nothing else gives
+ // them back to the exchange factory
+ for (Exchange notRouted : exchanges) {
+ releaseExchange(notRouted, true);
+ }
+ break;
+ }
shardIteratorHandler.updateShardIterator(shardId,
result.nextShardIterator());
if (!records.isEmpty()) {
lastSeenSequenceNumbers.put(shardId,
records.get(records.size() - 1).dynamodb().sequenceNumber());
+ firstNotRoutedSequenceNumbers.remove(shardId);
}
}
return processedExchangeCount;
diff --git
a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/ShardIteratorHandler.java
b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/ShardIteratorHandler.java
index e90a1330daa8..cb32f9398fc0 100644
---
a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/ShardIteratorHandler.java
+++
b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/ShardIteratorHandler.java
@@ -114,6 +114,15 @@ class ShardIteratorHandler {
return shardIterator;
}
+ /**
+ * Requests a shard iterator that starts at the record with the given
sequence number (and not after it).
+ */
+ String requestFreshShardIteratorAt(String shardId, String sequenceNumber) {
+ String shardIterator = getShardIterator(shardId,
ShardIteratorType.AT_SEQUENCE_NUMBER, sequenceNumber);
+ currentShardIterators.put(shardId, shardIterator);
+ return shardIterator;
+ }
+
Ddb2StreamEndpoint getEndpoint() {
return endpoint;
}
diff --git
a/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumerSuspendedBatchPooledExchangeTest.java
b/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumerSuspendedBatchPooledExchangeTest.java
new file mode 100644
index 000000000000..8bd5721dce10
--- /dev/null
+++
b/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumerSuspendedBatchPooledExchangeTest.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.ddbstream;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+import org.apache.camel.spi.ExchangeFactory;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * The same with the pooled exchange factory: the exchanges created for the
records that a suspending consumer does not
+ * route are given back to the pool, once, so the records routed after the
resume do not share an exchange.
+ */
+class Ddb2StreamConsumerSuspendedBatchPooledExchangeTest extends
Ddb2StreamConsumerSuspendedBatchTest {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ PooledExchangeFactory factory = new PooledExchangeFactory();
+ factory.setStatisticsEnabled(true);
+ context.getCamelContextExtension().setExchangeFactory(factory);
+ return context;
+ }
+
+ @Test
+ void exchangesOfTheRecordsThatWereNotRoutedAreGivenBackToThePool() throws
Exception {
+ client.put("1", "2", "3", "4");
+ suspendAtRecord = "2";
+
+
context.getCamelContextExtension().getExchangeFactoryManager().setStatisticsEnabled(true);
+ context.getRouteController().startRoute(ROUTE_ID);
+ awaitRouteSuspended();
+ assertEquals(List.of("1", "2"), routed);
+ // one exchange per record of the response: two were routed, and the
two that were not are given back as well
+ awaitExchangesTakenAndReleased(4);
+
+ context.getRouteController().resumeRoute(ROUTE_ID);
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(List.of("1", "2", "3", "4"), routed,
"records routed after the resume"));
+ // and only once: the pool does not hold an exchange twice
+ awaitExchangesTakenAndReleased(6);
+ }
+
+ private void awaitExchangesTakenAndReleased(int expected) {
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
+ ExchangeFactory.Statistics stats
+ =
context.getCamelContextExtension().getExchangeFactoryManager().getStatistics();
+ assertEquals(expected, stats.getCreatedCounter() +
stats.getAcquiredCounter(), "exchanges taken");
+ assertEquals(expected, stats.getReleasedCounter(), "exchanges
given back to the pool");
+ assertEquals(0, stats.getDiscardedCounter(), "exchanges
discarded");
+ });
+ }
+}
diff --git
a/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumerSuspendedBatchTest.java
b/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumerSuspendedBatchTest.java
new file mode 100644
index 000000000000..a6e8b18c9288
--- /dev/null
+++
b/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumerSuspendedBatchTest.java
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.ddbstream;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.dynamodb.model.DescribeStreamRequest;
+import software.amazon.awssdk.services.dynamodb.model.DescribeStreamResponse;
+import software.amazon.awssdk.services.dynamodb.model.ExpiredIteratorException;
+import software.amazon.awssdk.services.dynamodb.model.GetRecordsRequest;
+import software.amazon.awssdk.services.dynamodb.model.GetRecordsResponse;
+import software.amazon.awssdk.services.dynamodb.model.GetShardIteratorRequest;
+import software.amazon.awssdk.services.dynamodb.model.GetShardIteratorResponse;
+import software.amazon.awssdk.services.dynamodb.model.ListStreamsRequest;
+import software.amazon.awssdk.services.dynamodb.model.ListStreamsResponse;
+import software.amazon.awssdk.services.dynamodb.model.OperationType;
+import software.amazon.awssdk.services.dynamodb.model.Record;
+import software.amazon.awssdk.services.dynamodb.model.SequenceNumberRange;
+import software.amazon.awssdk.services.dynamodb.model.Shard;
+import software.amazon.awssdk.services.dynamodb.model.ShardIteratorType;
+import software.amazon.awssdk.services.dynamodb.model.Stream;
+import software.amazon.awssdk.services.dynamodb.model.StreamDescription;
+import software.amazon.awssdk.services.dynamodb.model.StreamRecord;
+import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsClient;
+import
software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsServiceClientConfiguration;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * A route that is suspended while the consumer routes the records of a
GetRecords response, and resumed: the records of
+ * that response that were not routed must be routed after the resume. (A
route that is stopped and started gets a new
+ * consumer, which starts from the configured iterator type.)
+ */
+class Ddb2StreamConsumerSuspendedBatchTest extends CamelTestSupport {
+
+ static final String ROUTE_ID = "ddbstream";
+ static final String LATEST_ROUTE_ID = "ddbstream-latest";
+
+ @BindToRegistry("ddbStreams")
+ final FakeStreamsClient client = new FakeStreamsClient();
+
+ final List<String> routed = new CopyOnWriteArrayList<>();
+ // the sequence number of the record whose processor suspends the route
(once)
+ volatile String suspendAtRecord;
+ private final AtomicBoolean suspendRequested = new AtomicBoolean();
+ private final AtomicReference<Exception> suspendFailure = new
AtomicReference<>();
+ private volatile Thread suspender;
+ // the route that the test starts, suspends and resumes
+ private volatile String routeId = ROUTE_ID;
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("aws2-ddbstream://table?amazonDynamoDbStreamsClient=#ddbStreams&streamIteratorType=FROM_START"
+ + "&initialDelay=0&delay=10")
+ .routeId(ROUTE_ID).autoStartup(false)
+ .to("direct:records");
+
+
from("aws2-ddbstream://table?amazonDynamoDbStreamsClient=#ddbStreams&streamIteratorType=FROM_LATEST"
+ + "&initialDelay=0&delay=10")
+ .routeId(LATEST_ROUTE_ID).autoStartup(false)
+ .to("direct:records");
+
+ from("direct:records")
+ .process(exchange -> {
+ String sequenceNumber
+ =
exchange.getMessage().getHeader(Ddb2StreamConstants.SEQUENCE_NUMBER,
String.class);
+ if (sequenceNumber.equals(suspendAtRecord)) {
+ suspendRouteDuringThePoll();
+ }
+ routed.add(sequenceNumber);
+ });
+ }
+ };
+ }
+
+ @Test
+ void recordsNotRoutedByASuspendingConsumerAreRoutedAfterTheResume() throws
Exception {
+ client.put("1", "2", "3", "4");
+ suspendAtRecord = "2";
+
+ context.getRouteController().startRoute(ROUTE_ID);
+ awaitRouteSuspended();
+ assertEquals(List.of("1", "2"), routed, "the suspend ends the batch
after the record that is being routed");
+
+ context.getRouteController().resumeRoute(ROUTE_ID);
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(List.of("1", "2", "3", "4"), routed,
"records routed after the resume"));
+
+ // the consumer continues after the last record it routed
+ GetShardIteratorRequest last =
client.iteratorRequests.get(client.iteratorRequests.size() - 1);
+ assertEquals(ShardIteratorType.AFTER_SEQUENCE_NUMBER,
last.shardIteratorType());
+ assertEquals("2", last.sequenceNumber());
+ }
+
+ @Test
+ void recordsOfAResponseThatArrivesWhileSuspendingAreRoutedAfterTheResume()
throws Exception {
+ client.put("1", "2");
+ // the route is suspended while the GetRecords request is in flight:
none of its records is routed
+ client.onGetRecords = this::suspendRouteDuringThePoll;
+
+ context.getRouteController().startRoute(ROUTE_ID);
+ awaitRouteSuspended();
+ assertEquals(List.of(), routed, "nothing is routed by the suspending
consumer");
+
+ context.getRouteController().resumeRoute(ROUTE_ID);
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(List.of("1", "2"), routed, "records routed
after the resume"));
+ assertEquals(1, client.iteratorRequests.size(), "the shard iterator of
the request is used again");
+ }
+
+ @Test
+ void laterRecordsAreRoutedOnceAfterABatchWasCutShort() throws Exception {
+ client.put("1", "2", "3");
+ suspendAtRecord = "1";
+
+ context.getRouteController().startRoute(ROUTE_ID);
+ awaitRouteSuspended();
+ client.put("4");
+
+ context.getRouteController().resumeRoute(ROUTE_ID);
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(List.of("1", "2", "3", "4"), routed,
"records routed after the resume"));
+ // a few more polls: nothing is routed twice
+ int requests = client.recordRequests();
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
client.recordRequests() >= requests + 3);
+ assertEquals(List.of("1", "2", "3", "4"), routed);
+ }
+
+ @Test
+ void
firstRecordsOfAShardThatWereNotRoutedAreNotReplacedByTheWholeShardWhenTheIteratorExpires()
throws Exception {
+ routeId = LATEST_ROUTE_ID;
+ // stored before the consumer starts: a FROM_LATEST consumer does not
route them
+ client.put("1", "2");
+
+ context.getRouteController().startRoute(LATEST_ROUTE_ID);
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
client.recordRequests() >= 1);
+ // the first response with records arrives while the route is
suspended: none of its records is routed, and
+ // no record of the shard was routed before
+ client.onRecordsReturned = this::suspendRouteDuringThePoll;
+ client.put("3", "4");
+ awaitRouteSuspended();
+ assertEquals(List.of(), routed, "nothing is routed by the suspending
consumer");
+
+ // the route stays suspended for longer than a shard iterator is valid
+ client.expireIterators();
+ context.getRouteController().resumeRoute(LATEST_ROUTE_ID);
+ await().atMost(20, TimeUnit.SECONDS).until(() -> routed.size() >= 2);
+ // a few more polls: nothing else is routed
+ int requests = client.recordRequests();
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
client.recordRequests() >= requests + 3);
+ assertEquals(List.of("3", "4"), routed, "the records that were not
routed, and not the shard from its start");
+
+ GetShardIteratorRequest last =
client.iteratorRequests.get(client.iteratorRequests.size() - 1);
+ assertEquals(ShardIteratorType.AT_SEQUENCE_NUMBER,
last.shardIteratorType());
+ assertEquals("3", last.sequenceNumber());
+ }
+
+ /**
+ * Called by the thread that polls: suspends the route from another
thread, as a route must not suspend itself, and
+ * returns when the suspend has reached the consumer (it no longer routes
the following records of the batch).
+ */
+ private void suspendRouteDuringThePoll() {
+ if (!suspendRequested.compareAndSet(false, true)) {
+ return;
+ }
+ String id = routeId;
+ Ddb2StreamConsumer consumer = (Ddb2StreamConsumer)
context.getRoute(id).getConsumer();
+ suspender = new Thread(() -> {
+ try {
+ context.getRouteController().suspendRoute(id);
+ } catch (Exception e) {
+ suspendFailure.set(e);
+ }
+ }, "suspend-route");
+ suspender.start();
+
+ CountDownLatch batchEnded = new CountDownLatch(1);
+ Thread watcher = new Thread(() -> {
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
!consumer.isBatchAllowed());
+ batchEnded.countDown();
+ }, "watch-consumer");
+ watcher.start();
+
+ // an interrupt of the thread that polls must not end this wait
+ boolean interrupted = false;
+ boolean ended = false;
+ long remaining = TimeUnit.SECONDS.toNanos(30);
+ long deadline = System.nanoTime() + remaining;
+ while (!ended && remaining > 0) {
+ try {
+ ended = batchEnded.await(remaining, TimeUnit.NANOSECONDS);
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ remaining = deadline - System.nanoTime();
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ void awaitRouteSuspended() throws InterruptedException {
+ await().atMost(20, TimeUnit.SECONDS).until(() -> suspender != null);
+ suspender.join(TimeUnit.SECONDS.toMillis(60));
+ assertFalse(suspender.isAlive(), "the route suspend has returned");
+ assertNull(suspendFailure.get());
+ assertEquals(ServiceStatus.Suspended,
context.getRouteController().getRouteStatus(routeId));
+ }
+
+ /**
+ * A stream with one shard. A shard iterator is a position between two
records.
+ */
+ static final class FakeStreamsClient implements DynamoDbStreamsClient {
+ final List<GetShardIteratorRequest> iteratorRequests = new
CopyOnWriteArrayList<>();
+ volatile Runnable onGetRecords;
+ // as onGetRecords, for a response that has records
+ volatile Runnable onRecordsReturned;
+ private final List<Record> records = new ArrayList<>();
+ private final Map<String, Integer> iterators = new HashMap<>();
+ private final Set<String> expiredIterators = new HashSet<>();
+ private int recordRequests;
+
+ synchronized void put(String... sequenceNumbers) {
+ for (String sequenceNumber : sequenceNumbers) {
+ records.add(Record.builder().eventID("event-" +
sequenceNumber).eventName(OperationType.INSERT)
+
.dynamodb(StreamRecord.builder().sequenceNumber(sequenceNumber).build()).build());
+ }
+ }
+
+ synchronized int recordRequests() {
+ return recordRequests;
+ }
+
+ /** The shard iterators given out so far are no longer valid. */
+ synchronized void expireIterators() {
+ expiredIterators.addAll(iterators.keySet());
+ }
+
+ @Override
+ public ListStreamsResponse listStreams(ListStreamsRequest request) {
+ return
ListStreamsResponse.builder().streams(Stream.builder().streamArn("arn:stream").build()).build();
+ }
+
+ @Override
+ public DescribeStreamResponse describeStream(DescribeStreamRequest
request) {
+ return DescribeStreamResponse.builder().streamDescription(
+
StreamDescription.builder().shards(Shard.builder().shardId("shard-0")
+ // an open shard: no ending sequence number
+
.sequenceNumberRange(SequenceNumberRange.builder().startingSequenceNumber("1").build())
+ .build()).build())
+ .build();
+ }
+
+ @Override
+ public synchronized GetShardIteratorResponse
getShardIterator(GetShardIteratorRequest request) {
+ iteratorRequests.add(request);
+ int position;
+ switch (request.shardIteratorType()) {
+ case TRIM_HORIZON:
+ position = 0;
+ break;
+ case LATEST:
+ position = records.size();
+ break;
+ case AT_SEQUENCE_NUMBER:
+ position = Integer.parseInt(request.sequenceNumber()) - 1;
+ break;
+ case AFTER_SEQUENCE_NUMBER:
+ position = Integer.parseInt(request.sequenceNumber());
+ break;
+ default:
+ throw new IllegalArgumentException("Not expected in this
test: " + request.shardIteratorType());
+ }
+ return
GetShardIteratorResponse.builder().shardIterator(newIterator(position)).build();
+ }
+
+ @Override
+ public GetRecordsResponse getRecords(GetRecordsRequest request) {
+ GetRecordsResponse response;
+ synchronized (this) {
+ recordRequests++;
+ if (expiredIterators.contains(request.shardIterator())) {
+ throw ExpiredIteratorException.builder().message("Iterator
expired").build();
+ }
+ int from = iterators.get(request.shardIterator());
+ int to = Math.min(records.size(), from + request.limit());
+ response = GetRecordsResponse.builder().records(new
ArrayList<>(records.subList(from, to)))
+ .nextShardIterator(newIterator(to)).build();
+ }
+ Runnable hook = onGetRecords;
+ if (hook != null) {
+ hook.run();
+ }
+ hook = onRecordsReturned;
+ if (hook != null && response.hasRecords() &&
!response.records().isEmpty()) {
+ hook.run();
+ }
+ return response;
+ }
+
+ private String newIterator(int position) {
+ String iterator = "iterator-" + (iterators.size() + 1);
+ iterators.put(iterator, position);
+ return iterator;
+ }
+
+ @Override
+ public DynamoDbStreamsServiceClientConfiguration
serviceClientConfiguration() {
+ return null;
+ }
+
+ @Override
+ public String serviceName() {
+ return DynamoDbStreamsClient.SERVICE_NAME;
+ }
+
+ @Override
+ public void close() {
+ // nothing to close
+ }
+ }
+}
