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 eb8ecb1b4543 CAMEL-25741: camel-aws2-ddb - aws2-ddbstream reads every
shard to its end and starts from streamIteratorType only once (#28068)
eb8ecb1b4543 is described below
commit eb8ecb1b45434e0a88acd453083516543101c9c1
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 19 17:00:46 2026 +0530
CAMEL-25741: camel-aws2-ddb - aws2-ddbstream reads every shard to its end
and starts from streamIteratorType only once (#28068)
* CAMEL-25741: camel-aws2-ddb - aws2-ddbstream reads every shard to its end
and starts from streamIteratorType only once
* CAMEL-25741: camel-aws2-ddb - simpler wording of the upgrade-guide entry
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../aws2/ddbstream/ShardIteratorHandler.java | 45 ++-
.../Ddb2StreamConsumerShardRotationTest.java | 326 +++++++++++++++++++++
.../aws2/ddbstream/ShardIteratorHandlerTest.java | 5 +
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 7 +
4 files changed, 359 insertions(+), 24 deletions(-)
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 cb32f9398fc0..265fcfb3d175 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
@@ -21,7 +21,6 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Set;
import
org.apache.camel.component.aws2.ddbstream.Ddb2StreamConfiguration.StreamIteratorType;
@@ -44,6 +43,7 @@ class ShardIteratorHandler {
private final Set<String> pendingClosedShards = new HashSet<>();
private String streamArn;
+ private boolean initialized;
private Map<String, String> currentShardIterators = new HashMap<>();
ShardIteratorHandler(Ddb2StreamEndpoint endpoint) {
@@ -54,41 +54,38 @@ class ShardIteratorHandler {
if (streamArn == null) {
streamArn = getStreamArn();
}
- // Either return cached ones or get new ones via GetShardIterator
requests.
- if (currentShardIterators.isEmpty()) {
+ // The configured iterator type is the start position of the consumer:
it is used once, and not again when
+ // every shard that was being read has closed (that would read the
stream again from its start, or skip the
+ // records of the new shards with FROM_LATEST)
+ if (!initialized) {
shardTree.populate(describeStreamPaginated());
StreamIteratorType streamIteratorType =
getEndpoint().getConfiguration().getStreamIteratorType();
currentShardIterators =
getCurrentShardIterators(streamIteratorType);
- } else {
- // Refresh the shard tree when any tracked shard has closed, so
that
- // children created by DynamoDB's shard rotation become visible.
- if (!pendingClosedShards.isEmpty()) {
- shardTree.populate(describeStreamPaginated());
- }
-
- Map<String, String> childShardIterators = new HashMap<>();
- for (Entry<String, String> currentShardIterator :
currentShardIterators.entrySet()) {
- List<Shard> children =
shardTree.getChildren(currentShardIterator.getKey());
- if (children.isEmpty()) { // This is still an active leaf
shard, reuse it.
- childShardIterators.put(currentShardIterator.getKey(),
currentShardIterator.getValue());
- } else {
- for (Shard child : children) { // Inactive shard, move
down to its children.
- String shardIterator =
getShardIterator(child.shardId(), ShardIteratorType.TRIM_HORIZON);
- childShardIterators.put(child.shardId(),
shardIterator);
- }
- }
- }
+ // no shard to read yet: try again at the next poll
+ initialized = !currentShardIterators.isEmpty();
+ } else if (!pendingClosedShards.isEmpty()) {
+ // A shard is left only when it was read to its end
(updateShardIterator got no next iterator): a closed
+ // shard, which already has children, can still hold records that
were not read. Refresh the shard tree,
+ // so that children created by DynamoDB's shard rotation become
visible, and read them from their start.
+ shardTree.populate(describeStreamPaginated());
+ Map<String, String> childShardIterators = new
HashMap<>(currentShardIterators);
+ Set<String> followed = new HashSet<>();
for (String closedShardId : pendingClosedShards) {
- for (Shard child : shardTree.getChildren(closedShardId)) {
+ List<Shard> children = shardTree.getChildren(closedShardId);
+ for (Shard child : children) {
if (!childShardIterators.containsKey(child.shardId())) {
String shardIterator =
getShardIterator(child.shardId(), ShardIteratorType.TRIM_HORIZON);
childShardIterators.put(child.shardId(),
shardIterator);
}
}
+ if (!children.isEmpty()) {
+ followed.add(closedShardId);
+ }
}
- pendingClosedShards.clear();
+ // a shard whose children are not listed yet is looked up again at
the next poll
+ pendingClosedShards.removeAll(followed);
currentShardIterators = childShardIterators;
}
diff --git
a/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumerShardRotationTest.java
b/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumerShardRotationTest.java
new file mode 100644
index 000000000000..a420b9583c49
--- /dev/null
+++
b/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumerShardRotationTest.java
@@ -0,0 +1,326 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+import org.apache.camel.BindToRegistry;
+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.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.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;
+
+/**
+ * DynamoDB closes the shards of a stream and creates child shards (shard
rotation, about every four hours, and splits).
+ * A closed shard keeps its records until they are read: the consumer must
read every shard to its end before it moves
+ * to the children, and the configured iterator type is only the start
position of the consumer.
+ */
+class Ddb2StreamConsumerShardRotationTest extends CamelTestSupport {
+
+ @BindToRegistry("ddbStreams")
+ final FakeStreamsClient client = new FakeStreamsClient();
+
+ final List<String> routed = new CopyOnWriteArrayList<>();
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("aws2-ddbstream://table?amazonDynamoDbStreamsClient=#ddbStreams&streamIteratorType=FROM_START"
+ + "&maxResultsPerRequest=2&initialDelay=0&delay=10")
+ .routeId("start").autoStartup(false)
+ .to("direct:records");
+
+
from("aws2-ddbstream://table?amazonDynamoDbStreamsClient=#ddbStreams&streamIteratorType=FROM_LATEST"
+ + "&maxResultsPerRequest=2&initialDelay=0&delay=10")
+ .routeId("latest").autoStartup(false)
+ .to("direct:records");
+
+ from("direct:records")
+ .process(exchange -> routed.add(
+
exchange.getMessage().getHeader(Ddb2StreamConstants.SEQUENCE_NUMBER,
String.class)));
+ }
+ };
+ }
+
+ @Test
+ void closedShardWithChildrenIsReadToItsEndFromStart() throws Exception {
+ // a stream of a few hours: shard-0 (5 records) rotated into shard-1,
which is open
+ client.shard("shard-0", null).put("1", "2", "3", "4", "5").close();
+ client.shard("shard-1", "shard-0").put("6");
+
+ context.getRouteController().startRoute("start");
+
+ awaitRoutedAndNothingMore(List.of("1", "2", "3", "4", "5", "6"));
+ }
+
+ @Test
+ void streamIsNotReadAgainWhenItsOnlyShardIsReadToItsEnd() throws Exception
{
+ // one partition: the only root shard is closed, its child is open
+ client.shard("shard-0", null).put("1", "2").close();
+ client.shard("shard-1", "shard-0").put("3");
+
+ context.getRouteController().startRoute("start");
+
+ awaitRoutedAndNothingMore(List.of("1", "2", "3"));
+ }
+
+ @Test
+ void recordsOfTheChildOfTheOnlyShardAreRoutedFromLatest() throws Exception
{
+ FakeShard parent = client.shard("shard-0", null).put("1");
+
+ context.getRouteController().startRoute("latest");
+ // the consumer has its LATEST position in shard-0
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
client.recordRequests() >= 1);
+
+ // shard rotation: shard-0 gets one more record and closes; the next
writes go to the new shard-1, and the
+ // first of them is written before the consumer has read shard-0 to
its end
+ FakeShard child = client.shard("shard-1", "shard-0");
+ client.onEndOfShard = closed -> {
+ if (closed == parent) {
+ child.put("4");
+ }
+ };
+ parent.put("2").close();
+
+ awaitRoutedAndNothingMore(List.of("2", "4"));
+ child.put("5");
+ awaitRoutedAndNothingMore(List.of("2", "4", "5"));
+ }
+
+ @Test
+ void childThatIsListedAfterItsParentWasReadToItsEndIsRead() throws
Exception {
+ FakeShard parent = client.shard("shard-0", null).put("1");
+
+ context.getRouteController().startRoute("latest");
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
client.recordRequests() >= 1);
+
+ // the shard closes, and DescribeStream does not list its child yet
when the consumer reads its end
+ parent.put("2").close();
+ await().atMost(20, TimeUnit.SECONDS).until(() -> client.endsRead() >=
1);
+ int describes = client.describeCalls();
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
client.describeCalls() >= describes + 2);
+ client.shard("shard-1", "shard-0").put("3", "4");
+
+ awaitRoutedAndNothingMore(List.of("2", "3", "4"));
+ }
+
+ @Test
+ void shardThatClosedWhileAnotherShardIsReadToItsEndIsReadToItsEnd() throws
Exception {
+ // two partitions
+ FakeShard a = client.shard("shard-a", null);
+ FakeShard b = client.shard("shard-b", null);
+
+ context.getRouteController().startRoute("latest");
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
client.recordRequests() >= 2);
+
+ // both shards rotate (the children are listed before their parents
close, so that the consumer finds them
+ // whenever it reads the end of a parent); shard-b has more records
than one GetRecords response returns
+ client.shard("shard-a1", "shard-a").put("8");
+ client.shard("shard-b1", "shard-b").put("9");
+ a.put("1").close();
+ b.put("2", "3", "4", "5", "6", "7").close();
+
+ awaitRoutedAndNothingMore(List.of("1", "2", "3", "4", "5", "6", "7",
"8", "9"));
+ }
+
+ private void awaitRoutedAndNothingMore(List<String> expected) {
+ await().atMost(20, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(expected.size(),
routed.size(), "routed so far: " + shown()));
+ // a few more polls of every shard: nothing is routed twice
+ int requests = client.recordRequests();
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
client.recordRequests() >= requests + 10);
+ List<String> sorted = new ArrayList<>(routed);
+ sorted.sort((x, y) -> Integer.compare(Integer.parseInt(x),
Integer.parseInt(y)));
+ assertEquals(expected, sorted, "each record routed once, in any order
across shards: " + shown());
+ }
+
+ private List<String> shown() {
+ List<String> copy = new ArrayList<>(routed);
+ return copy.size() > 20 ? copy.subList(0, 20) : copy;
+ }
+
+ final class FakeShard {
+ final String id;
+ final String parentId;
+ final List<Record> records = new ArrayList<>();
+ volatile boolean closed;
+
+ FakeShard(String id, String parentId) {
+ this.id = id;
+ this.parentId = parentId;
+ }
+
+ FakeShard put(String... sequenceNumbers) {
+ synchronized (client) {
+ for (String sequenceNumber : sequenceNumbers) {
+ records.add(Record.builder().eventID("event-" +
sequenceNumber).eventName(OperationType.INSERT)
+
.dynamodb(StreamRecord.builder().sequenceNumber(sequenceNumber).build()).build());
+ }
+ }
+ return this;
+ }
+
+ FakeShard close() {
+ closed = true;
+ return this;
+ }
+ }
+
+ /**
+ * A stream whose shards have records; an iterator is a position in a
shard. GetRecords on a closed shard that
+ * reaches its end returns no next iterator.
+ */
+ final class FakeStreamsClient implements DynamoDbStreamsClient {
+ private final Map<String, FakeShard> shards = new LinkedHashMap<>();
+ private final Map<String, Object[]> iterators = new HashMap<>();
+ volatile Consumer<FakeShard> onEndOfShard;
+ private int recordRequests;
+ private int describeCalls;
+ private int endsRead;
+
+ synchronized FakeShard shard(String id, String parentId) {
+ FakeShard shard = new FakeShard(id, parentId);
+ shards.put(id, shard);
+ return shard;
+ }
+
+ synchronized int recordRequests() {
+ return recordRequests;
+ }
+
+ synchronized int describeCalls() {
+ return describeCalls;
+ }
+
+ synchronized int endsRead() {
+ return endsRead;
+ }
+
+ @Override
+ public ListStreamsResponse listStreams(ListStreamsRequest request) {
+ return
ListStreamsResponse.builder().streams(Stream.builder().streamArn("arn:stream").build()).build();
+ }
+
+ @Override
+ public synchronized DescribeStreamResponse
describeStream(DescribeStreamRequest request) {
+ describeCalls++;
+ List<Shard> answer = new ArrayList<>();
+ for (FakeShard shard : shards.values()) {
+
answer.add(Shard.builder().shardId(shard.id).parentShardId(shard.parentId)
+
.sequenceNumberRange(SequenceNumberRange.builder().startingSequenceNumber("1")
+ // a closed shard has an ending sequence number
+ .endingSequenceNumber(shard.closed ? "1000" :
null).build())
+ .build());
+ }
+ return
DescribeStreamResponse.builder().streamDescription(StreamDescription.builder().shards(answer).build())
+ .build();
+ }
+
+ @Override
+ public synchronized GetShardIteratorResponse
getShardIterator(GetShardIteratorRequest request) {
+ FakeShard shard = shards.get(request.shardId());
+ int position;
+ switch (request.shardIteratorType()) {
+ case TRIM_HORIZON:
+ position = 0;
+ break;
+ case LATEST:
+ position = shard.records.size();
+ break;
+ default:
+ throw new IllegalArgumentException("Not expected in this
test: " + request.shardIteratorType());
+ }
+ return
GetShardIteratorResponse.builder().shardIterator(newIterator(shard,
position)).build();
+ }
+
+ @Override
+ public GetRecordsResponse getRecords(GetRecordsRequest request) {
+ FakeShard endOfShard = null;
+ GetRecordsResponse response;
+ synchronized (this) {
+ recordRequests++;
+ Object[] iterator = iterators.get(request.shardIterator());
+ FakeShard shard = (FakeShard) iterator[0];
+ int from = (Integer) iterator[1];
+ int to = Math.min(shard.records.size(), from +
request.limit());
+ String next = null;
+ if (shard.closed && to == shard.records.size()) {
+ endOfShard = shard;
+ endsRead++;
+ } else {
+ next = newIterator(shard, to);
+ }
+ response = GetRecordsResponse.builder().records(new
ArrayList<>(shard.records.subList(from, to)))
+ .nextShardIterator(next).build();
+ }
+ Consumer<FakeShard> hook = onEndOfShard;
+ if (endOfShard != null && hook != null) {
+ hook.accept(endOfShard);
+ }
+ return response;
+ }
+
+ private String newIterator(FakeShard shard, int position) {
+ String iterator = "iterator-" + (iterators.size() + 1);
+ iterators.put(iterator, new Object[] { shard, position });
+ return iterator;
+ }
+
+ @Override
+ public DynamoDbStreamsServiceClientConfiguration
serviceClientConfiguration() {
+ return null;
+ }
+
+ @Override
+ public String serviceName() {
+ return DynamoDbStreamsClient.SERVICE_NAME;
+ }
+
+ @Override
+ public void close() {
+ // nothing to close
+ }
+ }
+}
diff --git
a/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/ShardIteratorHandlerTest.java
b/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/ShardIteratorHandlerTest.java
index ef370fbc76eb..adeec23b96e1 100644
---
a/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/ShardIteratorHandlerTest.java
+++
b/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/ShardIteratorHandlerTest.java
@@ -94,10 +94,15 @@ class ShardIteratorHandlerTest extends CamelTestSupport {
endpoint.doStart();
assertEquals(Collections.singletonMap(SHARD_0.shardId(),
SHARD_ITERATOR_0), underTest.getShardIterators());
+ // a closed shard is left for its children once it was read to its
end, not before
+ assertEquals(Collections.singletonMap(SHARD_0.shardId(),
SHARD_ITERATOR_0), underTest.getShardIterators());
+ underTest.updateShardIterator(SHARD_0.shardId(), null);
Map<String, String> expectedShardIterators1 = new HashMap<>();
expectedShardIterators1.put(SHARD_1.shardId(), SHARD_ITERATOR_1);
expectedShardIterators1.put(SHARD_2.shardId(), SHARD_ITERATOR_2);
assertEquals(expectedShardIterators1, underTest.getShardIterators());
+ underTest.updateShardIterator(SHARD_1.shardId(), null);
+ underTest.updateShardIterator(SHARD_2.shardId(), null);
Map<String, String> expectedShardIterators2 = new HashMap<>();
expectedShardIterators2.put(SHARD_3.shardId(), SHARD_ITERATOR_3);
expectedShardIterators2.put(SHARD_4.shardId(), SHARD_ITERATOR_4);
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index efc53903f469..08ed3d96753b 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -828,6 +828,13 @@ now resolves to no value, as it already did without a
field, so its default valu
(`{{?hashicorp:secret:missing#field}}`) applies. Before, without a default
value, it failed with a
`NullPointerException`.
+=== camel-aws2-ddb - aws2-ddbstream reads every shard to its end
+
+The `aws2-ddbstream` consumer now reads a closed shard to its end before it
reads its child shards, and uses
+`streamIteratorType` only for its start position. Before, records left in a
closed shard could be skipped, and after a
+shard rotation the consumer started again from `streamIteratorType`: with
`FROM_START` it routed records of the stream
+again, and with `FROM_LATEST` it skipped records already written to the new
shards.
+
=== camel-bindy - a number field without precision keeps all its decimals when
marshalled
A `BigDecimal`, `Double` or `Float` field whose `@DataField` (or
`@KeyValuePairField`) has no `precision` and no
