This is an automated email from the ASF dual-hosted git repository.
ferenc-csaky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-aws.git
The following commit(s) were added to refs/heads/main by this push:
new fc3a8c9d [FLINK-40938] Support `AT_TIMESTAMP` initial position in
DynamoDB Streams source
fc3a8c9d is described below
commit fc3a8c9de287482ed4575a4cd3528228aaf7c6a4
Author: riyarawat-amazon <[email protected]>
AuthorDate: Thu Oct 29 15:26:39 2026 +0530
[FLINK-40938] Support `AT_TIMESTAMP` initial position in DynamoDB Streams
source
---
.../content/docs/connectors/datastream/dynamodb.md | 18 ++
.../sink/KinesisFirehoseSinkWriterTest.java | 2 +-
.../kinesis/sink/KinesisStreamsSinkWriterTest.java | 2 +-
.../dynamodb/source/DynamoDbStreamsSource.java | 13 +
.../DynamodbStreamsSourceConfigConstants.java | 20 +-
.../config/DynamodbStreamsSourceConfigUtil.java | 146 +++++++++++
.../DynamoDbStreamsSourceEnumerator.java | 9 +-
.../source/enumerator/tracker/SplitTracker.java | 105 +++++++-
.../source/proxy/DynamoDbStreamsProxy.java | 11 +
.../split/DynamoDbStreamsShardSplitSerializer.java | 17 +-
.../dynamodb/source/split/StartingPosition.java | 5 +
.../DynamodbStreamsSourceConfigUtilTest.java | 220 +++++++++++++++++
.../DynamoDbStreamsSourceEnumeratorTest.java | 269 +++++++++++++++++++++
.../enumerator/tracker/SplitTrackerTest.java | 216 +++++++++++++++++
.../source/proxy/DynamoDbStreamsProxyTest.java | 39 +++
.../DynamoDbStreamsShardSplitSerializerTest.java | 30 ++-
.../connector/sqs/sink/SqsSinkWriterTest.java | 2 +-
.../pom.xml | 3 +
.../src/main/resources/META-INF/NOTICE | 58 ++---
.../pom.xml | 3 +
.../src/main/resources/META-INF/NOTICE | 62 ++---
.../flink-sql-connector-dynamodb/pom.xml | 3 +
.../src/main/resources/META-INF/NOTICE | 60 ++---
pom.xml | 2 +-
24 files changed, 1207 insertions(+), 108 deletions(-)
diff --git a/docs/content/docs/connectors/datastream/dynamodb.md
b/docs/content/docs/connectors/datastream/dynamodb.md
index 4a738034..7d639e70 100644
--- a/docs/content/docs/connectors/datastream/dynamodb.md
+++ b/docs/content/docs/connectors/datastream/dynamodb.md
@@ -118,6 +118,24 @@ The above is a simple example of using the
`DynamoDbStreamsSource`.
To specify the starting position of the `DynamodbStreamsSource`, users can set
the `DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_POSITION` in
configuration.
- `LATEST`: read all shards of the stream starting from the latest record.
- `TRIM_HORIZON`: read all shards of the stream starting from the earliest
record possible (data is trimmed by DynamoDB after 24 hours).
+- `AT_TIMESTAMP`: read all shards of the stream starting from the first record
at or after a given timestamp.
+ The timestamp is set with
`DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_TIMESTAMP`, either as a
date string in the
+ format given by
`DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_TIMESTAMP_FORMAT` (default
`yyyy-MM-dd'T'HH:mm:ss.SSSXXX`)
+ or as epoch seconds. The timestamp must not be in the future. A timestamp
older than the stream's retention starts each
+ shard from its earliest available record, like `TRIM_HORIZON`.
+
+{{< tabs "dynamodb-streams-source-at-timestamp" >}}
+{{< tab "Java" >}}
+```java
+Configuration sourceConfig = new Configuration();
+sourceConfig.set(DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_POSITION,
DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP);
+sourceConfig.set(DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_TIMESTAMP,
"2024-05-01T12:00:00.000Z");
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+The starting position applies only when the job starts without state. When a
job is restored from a checkpoint or savepoint,
+each shard resumes from its checkpointed position and the configured timestamp
is not used.
### Deserialization Schema
diff --git
a/flink-connector-aws/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriterTest.java
b/flink-connector-aws/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriterTest.java
index 2d9ddefe..6d686aa2 100644
---
a/flink-connector-aws/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriterTest.java
+++
b/flink-connector-aws/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriterTest.java
@@ -50,7 +50,7 @@ public class KinesisFirehoseSinkWriterTest {
@BeforeEach
void setup() throws IOException {
TestSinkInitContext sinkInitContext = new TestSinkInitContext();
- Properties sinkProperties =
AWSServicesTestUtils.createConfig("https://fake_aws_endpoint";);
+ Properties sinkProperties =
AWSServicesTestUtils.createConfig("https://fake-aws-endpoint";);
KinesisFirehoseSink<String> sink =
new KinesisFirehoseSink<>(
ELEMENT_CONVERTER_PLACEHOLDER,
diff --git
a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkWriterTest.java
b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkWriterTest.java
index f3c13d42..2fb4d236 100644
---
a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkWriterTest.java
+++
b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkWriterTest.java
@@ -58,7 +58,7 @@ public class KinesisStreamsSinkWriterTest {
void
testCreateKinesisStreamsSinkWriterInitializesRateLimitingStrategyWithExpectedParameters()
throws IOException {
TestSinkInitContext sinkInitContext = new TestSinkInitContext();
- Properties sinkProperties =
AWSServicesTestUtils.createConfig("https://fake_aws_endpoint";);
+ Properties sinkProperties =
AWSServicesTestUtils.createConfig("https://fake-aws-endpoint";);
KinesisStreamsSink<String> sink =
new KinesisStreamsSink<>(
ELEMENT_CONVERTER_PLACEHOLDER,
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/DynamoDbStreamsSource.java
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/DynamoDbStreamsSource.java
index 1581c9da..96c5eb0d 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/DynamoDbStreamsSource.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/DynamoDbStreamsSource.java
@@ -32,6 +32,7 @@ import org.apache.flink.connector.aws.util.AWSClientUtil;
import org.apache.flink.connector.aws.util.AWSGeneralUtil;
import
org.apache.flink.connector.base.source.reader.fetcher.SingleThreadFetcherManager;
import
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants;
+import
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigUtil;
import
org.apache.flink.connector.dynamodb.source.enumerator.DynamoDbStreamsShardAssigner;
import
org.apache.flink.connector.dynamodb.source.enumerator.DynamoDbStreamsSourceEnumerator;
import
org.apache.flink.connector.dynamodb.source.enumerator.DynamoDbStreamsSourceEnumeratorState;
@@ -60,7 +61,10 @@ import software.amazon.awssdk.services.dynamodb.model.Shard;
import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsClient;
import software.amazon.awssdk.utils.AttributeMap;
+import javax.annotation.Nullable;
+
import java.time.Duration;
+import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -100,6 +104,12 @@ public class DynamoDbStreamsSource<T>
protected final DynamoDbStreamsDeserializationSchema<T>
deserializationSchema;
protected final DynamoDbStreamsShardAssigner dynamoDbStreamsShardAssigner;
+ /**
+ * The AT_TIMESTAMP initial timestamp, resolved and validated once when
the source is built, so
+ * the enumerator uses exactly the value that was validated. Null for
other initial positions.
+ */
+ @Nullable private final Instant initialTimestamp;
+
DynamoDbStreamsSource(
String streamArn,
Configuration sourceConfig,
@@ -110,6 +120,8 @@ public class DynamoDbStreamsSource<T>
Preconditions.checkNotNull(sourceConfig);
Preconditions.checkNotNull(deserializationSchema);
Preconditions.checkNotNull(dynamoDbStreamsShardAssigner);
+ this.initialTimestamp =
+
DynamodbStreamsSourceConfigUtil.validateStreamSourceConfiguration(sourceConfig);
this.streamArn = streamArn;
this.sourceConfig = sourceConfig;
this.deserializationSchema = deserializationSchema;
@@ -203,6 +215,7 @@ public class DynamoDbStreamsSource<T>
enumContext,
streamArn,
sourceConfig,
+ initialTimestamp,
createDynamoDbStreamsProxy(sourceConfig,
adaptiveRetryStrategy),
dynamoDbStreamsShardAssigner,
checkpoint);
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigConstants.java
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigConstants.java
index 94b02b86..2ee97fa5 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigConstants.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigConstants.java
@@ -30,7 +30,8 @@ public class DynamodbStreamsSourceConfigConstants {
/** Marks the initial position to use when reading from the Dynamodb
stream. */
public enum InitialPosition {
LATEST,
- TRIM_HORIZON
+ TRIM_HORIZON,
+ AT_TIMESTAMP
}
public static final ConfigOption<InitialPosition> STREAM_INITIAL_POSITION =
@@ -39,6 +40,23 @@ public class DynamodbStreamsSourceConfigConstants {
.defaultValue(InitialPosition.LATEST)
.withDescription("The initial position to start reading
Dynamodb streams.");
+ public static final ConfigOption<String> STREAM_INITIAL_TIMESTAMP =
+ ConfigOptions.key("flink.stream.initpos.timestamp")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The initial timestamp at which to start reading
from the DynamoDB stream. "
+ + "Used only when the initial position is
AT_TIMESTAMP.");
+
+ public static final ConfigOption<String> STREAM_INITIAL_TIMESTAMP_FORMAT =
+ ConfigOptions.key("flink.stream.initpos.timestamp.format")
+ .stringType()
+ .defaultValue("yyyy-MM-dd'T'HH:mm:ss.SSSXXX")
+ .withDescription(
+ "The date format of the initial timestamp. Used to
parse "
+ + "flink.stream.initpos.timestamp when it
is not an epoch-second "
+ + "value. Used only when the initial
position is AT_TIMESTAMP.");
+
public static final ConfigOption<Duration> SHARD_DISCOVERY_INTERVAL =
ConfigOptions.key("flink.shard.discovery.intervalmillis")
.durationType()
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigUtil.java
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigUtil.java
new file mode 100644
index 00000000..87328751
--- /dev/null
+++
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigUtil.java
@@ -0,0 +1,146 @@
+/*
+ * 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.flink.connector.dynamodb.source.config;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.text.ParsePosition;
+import java.text.SimpleDateFormat;
+import java.time.Instant;
+import java.util.Date;
+
+import static
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_POSITION;
+import static
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_TIMESTAMP;
+import static
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_TIMESTAMP_FORMAT;
+
+/** Utility functions to use with {@link
DynamodbStreamsSourceConfigConstants}. */
+@Internal
+public class DynamodbStreamsSourceConfigUtil {
+
+ private DynamodbStreamsSourceConfigUtil() {
+ // private constructor to prevent initialization of utility class.
+ }
+
+ /**
+ * Parses the timestamp at which to start reading, used for the
AT_TIMESTAMP initial position.
+ * The value is parsed strictly (non-lenient, whole string) using the
configured {@link
+ * DynamodbStreamsSourceConfigConstants#STREAM_INITIAL_TIMESTAMP_FORMAT};
if that fails it is
+ * interpreted as a non-negative, possibly fractional, epoch-second value.
+ *
+ * @param sourceConfig the configuration to parse the timestamp from
+ * @return the parsed initial timestamp
+ */
+ public static Instant parseInitialTimestamp(Configuration sourceConfig) {
+ String timestampString = sourceConfig.get(STREAM_INITIAL_TIMESTAMP);
+ if (timestampString == null) {
+ throw new IllegalArgumentException(
+ STREAM_INITIAL_TIMESTAMP.key()
+ + " must be set when "
+ + STREAM_INITIAL_POSITION.key()
+ + " is AT_TIMESTAMP.");
+ }
+ String format = sourceConfig.get(STREAM_INITIAL_TIMESTAMP_FORMAT);
+ Instant parsed = parseWithFormat(timestampString, format);
+ if (parsed != null) {
+ return parsed;
+ }
+ // Fall back to interpreting the value as epoch seconds (may be
fractional).
+ final double epochSeconds;
+ try {
+ epochSeconds = Double.parseDouble(timestampString);
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Unable to parse "
+ + STREAM_INITIAL_TIMESTAMP.key()
+ + " value '"
+ + timestampString
+ + "' using format '"
+ + format
+ + "' or as epoch seconds.",
+ e);
+ }
+ if (!Double.isFinite(epochSeconds) || epochSeconds < 0) {
+ throw new IllegalArgumentException(
+ STREAM_INITIAL_TIMESTAMP.key()
+ + " value '"
+ + timestampString
+ + "' must be a date in format '"
+ + format
+ + "' or a non-negative epoch-second value.");
+ }
+ return Instant.ofEpochMilli((long) (epochSeconds * 1000));
+ }
+
+ /** Parses the whole string with the format, non-leniently; null if it
does not match. */
+ @Nullable
+ private static Instant parseWithFormat(String timestampString, String
format) {
+ final SimpleDateFormat dateFormat;
+ try {
+ dateFormat = new SimpleDateFormat(format);
+ } catch (IllegalArgumentException e) {
+ return null;
+ }
+ dateFormat.setLenient(false);
+ ParsePosition position = new ParsePosition(0);
+ Date date = dateFormat.parse(timestampString, position);
+ if (date == null || position.getIndex() != timestampString.length()) {
+ return null;
+ }
+ return date.toInstant();
+ }
+
+ /**
+ * Validates the source configuration, failing fast at startup so that a
misconfigured job does
+ * not fail later during shard discovery or reading. When the initial
position is AT_TIMESTAMP,
+ * the timestamp must be present, parseable, and not in the future: a
future timestamp is
+ * rejected by GetShardIterator, which would otherwise make the job
restart repeatedly until the
+ * clock passes it.
+ *
+ * @param sourceConfig the configuration to validate
+ * @return the validated initial timestamp when the initial position is
AT_TIMESTAMP, otherwise
+ * null. The source passes this value to the enumerator, so the
timestamp that was validated
+ * is the one used.
+ */
+ @Nullable
+ public static Instant validateStreamSourceConfiguration(Configuration
sourceConfig) {
+ Preconditions.checkNotNull(sourceConfig, "Config cannot be null");
+ if
(DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP.equals(
+ sourceConfig.get(STREAM_INITIAL_POSITION))) {
+ // Parsing throws for a missing or invalid timestamp, surfacing
the error at startup.
+ final Instant timestamp = parseInitialTimestamp(sourceConfig);
+ final Instant now = Instant.now();
+ if (timestamp.isAfter(now)) {
+ throw new IllegalArgumentException(
+ STREAM_INITIAL_TIMESTAMP.key()
+ + " "
+ + timestamp
+ + " is in the future (current time "
+ + now
+ + "). A future timestamp is not allowed;
supply a timestamp at or"
+ + " before the current time.");
+ }
+ return timestamp;
+ }
+ return null;
+ }
+}
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/DynamoDbStreamsSourceEnumerator.java
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/DynamoDbStreamsSourceEnumerator.java
index 0fb53907..90f40f14 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/DynamoDbStreamsSourceEnumerator.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/DynamoDbStreamsSourceEnumerator.java
@@ -33,6 +33,7 @@ import
org.apache.flink.connector.dynamodb.source.exception.DynamoDbStreamsSourc
import org.apache.flink.connector.dynamodb.source.proxy.StreamProxy;
import
org.apache.flink.connector.dynamodb.source.split.DynamoDbStreamsShardSplit;
import org.apache.flink.connector.dynamodb.source.util.ListShardsResult;
+import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -82,6 +83,7 @@ public class DynamoDbStreamsSourceEnumerator
SplitEnumeratorContext<DynamoDbStreamsShardSplit> context,
String streamArn,
Configuration sourceConfig,
+ @Nullable Instant initialTimestamp,
StreamProxy streamProxy,
DynamoDbStreamsShardAssigner shardAssigner,
DynamoDbStreamsSourceEnumeratorState state) {
@@ -93,7 +95,12 @@ public class DynamoDbStreamsSourceEnumerator
this.shardAssignerContext = new ShardAssignerContext(splitAssignment,
context);
InitialPosition initialPosition =
sourceConfig.get(STREAM_INITIAL_POSITION);
if (state == null) {
- this.startTimestamp = Instant.now();
+ this.startTimestamp =
+ initialPosition == InitialPosition.AT_TIMESTAMP
+ ? Preconditions.checkNotNull(
+ initialTimestamp,
+ "initialTimestamp must be set when the
initial position is AT_TIMESTAMP")
+ : Instant.now();
this.splitTracker = new SplitTracker(streamArn, initialPosition,
this.startTimestamp);
} else {
this.splitTracker =
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java
index 38fdf16c..8a8363f6 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java
@@ -33,6 +33,7 @@ import org.slf4j.LoggerFactory;
import software.amazon.awssdk.services.dynamodb.model.Shard;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
@@ -43,6 +44,7 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
+import static
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP;
import static
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants.InitialPosition.LATEST;
import static
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants.InitialPosition.TRIM_HORIZON;
import static
org.apache.flink.connector.dynamodb.source.enumerator.SplitAssignmentStatus.ASSIGNED;
@@ -58,6 +60,9 @@ public class SplitTracker {
private final Map<String, Set<String>> parentChildSplitMap = new
ConcurrentHashMap<>();
private final Set<String> assignedSplits = new HashSet<>();
private final Set<String> finishedSplits = new HashSet<>();
+ // Untracked ancestors already warned about, so the warning is logged once
per shard rather
+ // than on every discovery while the shard is still listed.
+ private final Set<String> warnedSkippedAncestors = new HashSet<>();
private final String streamArn;
private final InitialPosition initialPosition;
private final Instant startTimestamp;
@@ -95,15 +100,24 @@ public class SplitTracker {
* @param shardsToAdd collection of splits to add to tracking
*/
public void addSplits(Collection<Shard> shardsToAdd) {
- if (TRIM_HORIZON.equals(initialPosition)) {
- addSplitsForTrimHorizon(shardsToAdd);
+ if (AT_TIMESTAMP.equals(initialPosition)) {
+ addSplitsForAtTimestamp(shardsToAdd);
return;
}
- addSplitsForLatest(shardsToAdd);
+ if (LATEST.equals(initialPosition)) {
+ addSplitsForLatest(shardsToAdd);
+ return;
+ }
+ addSplitsWithPosition(shardsToAdd, initialPosition);
}
public void addChildSplits(Collection<Shard> childShardsToAdd) {
- addSplitsForTrimHorizon(childShardsToAdd);
+ // Under AT_TIMESTAMP, children from the finish-handoff also start at
AT_TIMESTAMP, so
+ // they match shards found by periodic discovery and never emit
pre-timestamp records.
+ // Other initial positions keep the TRIM_HORIZON handoff.
+ InitialPosition childPosition =
+ AT_TIMESTAMP.equals(initialPosition) ? AT_TIMESTAMP :
TRIM_HORIZON;
+ addSplitsWithPosition(childShardsToAdd, childPosition);
}
private void addSplitsForLatest(Collection<Shard> shardsToAdd) {
@@ -159,15 +173,22 @@ public class SplitTracker {
return null;
}
- private void addSplitsForTrimHorizon(Collection<Shard> shardsToAdd) {
+ /** Tracks each shard (if not already known) with the given initial
position. */
+ private void addSplitsWithPosition(Collection<Shard> shardsToAdd,
InitialPosition position) {
for (Shard shard : shardsToAdd) {
- String shardId = shard.shardId();
- if (!knownSplits.containsKey(shardId)) {
- DynamoDbStreamsShardSplit newSplit = mapToSplit(shard,
TRIM_HORIZON);
- knownSplits.put(shardId, newSplit);
- addSplitToMapping(newSplit);
- }
+ trackSplit(shard, position);
+ }
+ }
+
+ /** Adds a single shard to tracking with the given initial position,
unless already tracked. */
+ private void trackSplit(Shard shard, InitialPosition position) {
+ String shardId = shard.shardId();
+ if (knownSplits.containsKey(shardId)) {
+ return;
}
+ DynamoDbStreamsShardSplit newSplit = mapToSplit(shard, position);
+ knownSplits.put(shardId, newSplit);
+ addSplitToMapping(newSplit);
}
private void addSplitToMapping(DynamoDbStreamsShardSplit split) {
@@ -179,6 +200,65 @@ public class SplitTracker {
.add(split.splitId());
}
+ /**
+ * Adds shards discovered under the AT_TIMESTAMP initial position. Each
shard that is not
+ * already tracked is added at AT_TIMESTAMP:
GetShardIterator(AT_TIMESTAMP) resolves the
+ * per-record position, so a shard that straddles the timestamp never
emits pre-timestamp
+ * records, and the position does not depend on the discovery path or
restore timing.
+ *
+ * <p>Ancestors of already-tracked shards are not added, to keep the
parent-before-child order.
+ * Usually they have already been read. An ancestor that was never tracked
(for example, one an
+ * earlier discovery did not return) is skipped too, so any of its records
at or after the
+ * timestamp are not read; a warning is logged once per such shard.
+ */
+ private void addSplitsForAtTimestamp(Collection<Shard> shardsToAdd) {
+ Map<String, Shard> shardIdToShardMap =
+ shardsToAdd.stream().collect(Collectors.toMap(Shard::shardId,
shard -> shard));
+
+ Set<String> ancestorsOfTrackedSplits =
findAncestorsOfTrackedSplits(shardIdToShardMap);
+
+ List<Shard> eligibleShards = new ArrayList<>();
+ for (Shard shard : shardsToAdd) {
+ String shardId = shard.shardId();
+ if (!ancestorsOfTrackedSplits.contains(shardId)) {
+ eligibleShards.add(shard);
+ } else if (!knownSplits.containsKey(shardId) &&
warnedSkippedAncestors.add(shardId)) {
+ LOG.warn(
+ "Skipping shard {} of stream {}: it was never tracked
but is an ancestor of "
+ + "a tracked shard, so it is not read to keep
the parent-before-child "
+ + "order. Records in it at or after {} are not
read.",
+ shardId,
+ streamArn,
+ startTimestamp);
+ }
+ }
+ addSplitsWithPosition(eligibleShards, AT_TIMESTAMP);
+ }
+
+ private Set<String> findAncestorsOfTrackedSplits(Map<String, Shard>
shardIdToShardMap) {
+ Set<String> ancestorsOfTrackedSplits = new HashSet<>();
+ // knownSplits already contains finished splits (they are removed from
both only on
+ // cleanup), so its key set covers every tracked shard.
+ Set<String> trackedShardIds = new HashSet<>(knownSplits.keySet());
+
+ for (String trackedShardId : trackedShardIds) {
+ Shard shard = shardIdToShardMap.get(trackedShardId);
+ if (shard == null) {
+ continue;
+ }
+ String currentParentId = shard.parentShardId();
+ while (currentParentId != null &&
!ancestorsOfTrackedSplits.contains(currentParentId)) {
+ ancestorsOfTrackedSplits.add(currentParentId);
+ Shard parentShard = shardIdToShardMap.get(currentParentId);
+ if (parentShard == null) {
+ break;
+ }
+ currentParentId = parentShard.parentShardId();
+ }
+ }
+ return ancestorsOfTrackedSplits;
+ }
+
private DynamoDbStreamsShardSplit mapToSplit(
Shard shard, DynamodbStreamsSourceConfigConstants.InitialPosition
initialPosition) {
StartingPosition startingPosition;
@@ -186,6 +266,9 @@ public class SplitTracker {
case LATEST:
startingPosition = StartingPosition.latest();
break;
+ case AT_TIMESTAMP:
+ startingPosition =
StartingPosition.atTimestamp(startTimestamp);
+ break;
case TRIM_HORIZON:
default:
startingPosition = StartingPosition.fromStart();
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/proxy/DynamoDbStreamsProxy.java
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/proxy/DynamoDbStreamsProxy.java
index 133da8dd..0755460b 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/proxy/DynamoDbStreamsProxy.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/proxy/DynamoDbStreamsProxy.java
@@ -40,6 +40,7 @@ import
software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsClient;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.time.Instant;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
@@ -225,6 +226,16 @@ public class DynamoDbStreamsProxy implements StreamProxy {
case TRIM_HORIZON:
case LATEST:
break;
+ case AT_TIMESTAMP:
+ if (startingPosition.getStartingMarker() instanceof Instant) {
+ requestBuilder =
+ requestBuilder.timestamp(
+ (Instant)
startingPosition.getStartingMarker());
+ } else {
+ throw new IllegalArgumentException(
+ "Invalid object given for
GetShardIteratorRequest() when ShardIteratorType is AT_TIMESTAMP. Must be an
Instant.");
+ }
+ break;
case AT_SEQUENCE_NUMBER:
case AFTER_SEQUENCE_NUMBER:
if (startingPosition.getStartingMarker() instanceof String) {
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/split/DynamoDbStreamsShardSplitSerializer.java
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/split/DynamoDbStreamsShardSplitSerializer.java
index 10ddc79d..4c9eb08c 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/split/DynamoDbStreamsShardSplitSerializer.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/split/DynamoDbStreamsShardSplitSerializer.java
@@ -31,6 +31,7 @@ import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
@@ -45,8 +46,9 @@ import java.util.Set;
public class DynamoDbStreamsShardSplitSerializer
implements SimpleVersionedSerializer<DynamoDbStreamsShardSplit> {
- private static final Set<Integer> COMPATIBLE_VERSIONS = new
HashSet<>(Arrays.asList(0, 1, 2));
- private static final int CURRENT_VERSION = 2;
+ private static final Set<Integer> COMPATIBLE_VERSIONS =
+ new HashSet<>(Arrays.asList(0, 1, 2, 3));
+ private static final int CURRENT_VERSION = 3;
@Override
public int getVersion() {
@@ -69,6 +71,12 @@ public class DynamoDbStreamsShardSplitSerializer
out.writeBoolean(startingMarker instanceof String);
if (startingMarker instanceof String) {
out.writeUTF((String) startingMarker);
+ } else if (startingMarker instanceof Instant) {
+ out.writeLong(((Instant) startingMarker).toEpochMilli());
+ } else {
+ throw new IOException(
+ "Unsupported starting marker type: "
+ + startingMarker.getClass().getName());
}
}
if (split.getParentShardId() == null) {
@@ -116,8 +124,11 @@ public class DynamoDbStreamsShardSplitSerializer
final boolean hasStartingMarker = in.readBoolean();
if (hasStartingMarker) {
- if (in.readBoolean()) {
+ boolean isString = in.readBoolean();
+ if (isString) {
startingMarker = in.readUTF();
+ } else if (version >= 3) {
+ startingMarker = Instant.ofEpochMilli(in.readLong());
}
}
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/split/StartingPosition.java
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/split/StartingPosition.java
index 7de3cfaa..a9089737 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/split/StartingPosition.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/split/StartingPosition.java
@@ -24,6 +24,7 @@ import
software.amazon.awssdk.services.dynamodb.model.ShardIteratorType;
import javax.annotation.Nullable;
+import java.time.Instant;
import java.util.Objects;
import static
software.amazon.awssdk.services.dynamodb.model.ShardIteratorType.AFTER_SEQUENCE_NUMBER;
@@ -63,6 +64,10 @@ public final class StartingPosition {
return new StartingPosition(LATEST, null);
}
+ public static StartingPosition atTimestamp(Instant timestamp) {
+ return new StartingPosition(ShardIteratorType.AT_TIMESTAMP, timestamp);
+ }
+
@Override
public String toString() {
return "StartingPosition{"
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigUtilTest.java
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigUtilTest.java
new file mode 100644
index 00000000..b58a99c3
--- /dev/null
+++
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigUtilTest.java
@@ -0,0 +1,220 @@
+/*
+ * 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.flink.connector.dynamodb.source.config;
+
+import org.apache.flink.configuration.Configuration;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.time.Instant;
+
+import static
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_POSITION;
+import static
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_TIMESTAMP;
+import static
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_TIMESTAMP_FORMAT;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link DynamodbStreamsSourceConfigUtil}. */
+class DynamodbStreamsSourceConfigUtilTest {
+
+ @Test
+ void testParseInitialTimestampWithDefaultFormat() {
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP, "2024-05-01T12:00:00.000Z");
+
+ Instant parsed =
DynamodbStreamsSourceConfigUtil.parseInitialTimestamp(sourceConfig);
+
+ assertThat(parsed).isEqualTo(Instant.parse("2024-05-01T12:00:00Z"));
+ }
+
+ @Test
+ void testParseInitialTimestampFallsBackToEpochSeconds() {
+ Configuration sourceConfig = new Configuration();
+ // Not in the configured date format -> interpreted as epoch seconds.
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP, "1704067200");
+
+ Instant parsed =
DynamodbStreamsSourceConfigUtil.parseInitialTimestamp(sourceConfig);
+
+ assertThat(parsed).isEqualTo(Instant.ofEpochMilli(1704067200000L));
+ }
+
+ @Test
+ void testParseInitialTimestampWithCustomFormat() {
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP_FORMAT, "dd/MM/yyyy HH:mm:ss
XXX");
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP, "01/05/2024 14:00:00
+02:00");
+
+ Instant parsed =
DynamodbStreamsSourceConfigUtil.parseInitialTimestamp(sourceConfig);
+
+ assertThat(parsed).isEqualTo(Instant.parse("2024-05-01T12:00:00Z"));
+ }
+
+ @Test
+ void testParseInitialTimestampThrowsWhenUnparseable() {
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP, "not-a-timestamp");
+
+ assertThatThrownBy(
+ () ->
DynamodbStreamsSourceConfigUtil.parseInitialTimestamp(sourceConfig))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Unable to parse")
+ .hasMessageContaining("not-a-timestamp");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"NaN", "Infinity", "-1"})
+ void
testParseInitialTimestampRejectsNonFiniteOrNegativeEpochSeconds(String value) {
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP, value);
+
+ assertThatThrownBy(
+ () ->
DynamodbStreamsSourceConfigUtil.parseInitialTimestamp(sourceConfig))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("non-negative epoch-second value");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"2024-02-30T00:00:00.000Z",
"2024-05-01T12:00:00.000Zgarbage"})
+ void
testParseInitialTimestampRejectsInvalidOrPartiallyMatchingDates(String value) {
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP, value);
+
+ assertThatThrownBy(
+ () ->
DynamodbStreamsSourceConfigUtil.parseInitialTimestamp(sourceConfig))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Unable to parse");
+ }
+
+ @Test
+ void testValidateReturnsTheValidatedTimestamp() {
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ STREAM_INITIAL_POSITION,
+
DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP);
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP, "2024-05-01T12:00:00.000Z");
+
+
assertThat(DynamodbStreamsSourceConfigUtil.validateStreamSourceConfiguration(sourceConfig))
+ .isEqualTo(Instant.parse("2024-05-01T12:00:00Z"));
+ }
+
+ @Test
+ void testValidateReturnsNullForNonTimestampPositions() {
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ STREAM_INITIAL_POSITION,
+ DynamodbStreamsSourceConfigConstants.InitialPosition.LATEST);
+
+
assertThat(DynamodbStreamsSourceConfigUtil.validateStreamSourceConfiguration(sourceConfig))
+ .isNull();
+ }
+
+ @Test
+ void testParseInitialTimestampThrowsWhenMissing() {
+ Configuration sourceConfig = new Configuration();
+
+ assertThatThrownBy(
+ () ->
DynamodbStreamsSourceConfigUtil.parseInitialTimestamp(sourceConfig))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(STREAM_INITIAL_TIMESTAMP.key());
+ }
+
+ @Test
+ void testValidateFailsFastForAtTimestampWithoutTimestamp() {
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ STREAM_INITIAL_POSITION,
+
DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP);
+
+ assertThatThrownBy(
+ () ->
+
DynamodbStreamsSourceConfigUtil.validateStreamSourceConfiguration(
+ sourceConfig))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(STREAM_INITIAL_TIMESTAMP.key());
+ }
+
+ @Test
+ void testValidatePassesForAtTimestampWithTimestamp() {
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ STREAM_INITIAL_POSITION,
+
DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP);
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP, "2024-05-01T12:00:00.000Z");
+
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+
DynamodbStreamsSourceConfigUtil.validateStreamSourceConfiguration(
+ sourceConfig));
+ }
+
+ @Test
+ void testValidateRejectsFutureAtTimestamp() {
+ // A future AT_TIMESTAMP must fail fast at source construction, so the
job halts instead of
+ // restarting on GetShardIterator rejections until the clock passes
the timestamp.
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ STREAM_INITIAL_POSITION,
+
DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP);
+ sourceConfig.set(
+ STREAM_INITIAL_TIMESTAMP,
+
String.valueOf(Instant.now().plusSeconds(60).getEpochSecond()));
+
+ assertThatThrownBy(
+ () ->
+
DynamodbStreamsSourceConfigUtil.validateStreamSourceConfiguration(
+ sourceConfig))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("future")
+ .hasMessageContaining(STREAM_INITIAL_TIMESTAMP.key());
+ }
+
+ @Test
+ void testValidatePassesForCurrentAtTimestamp() {
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ STREAM_INITIAL_POSITION,
+
DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP);
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP,
String.valueOf(Instant.now().getEpochSecond()));
+
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+
DynamodbStreamsSourceConfigUtil.validateStreamSourceConfiguration(
+ sourceConfig));
+ }
+
+ @Test
+ void testValidateIsNoOpForNonTimestampPositions() {
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ STREAM_INITIAL_POSITION,
+
DynamodbStreamsSourceConfigConstants.InitialPosition.TRIM_HORIZON);
+
+ // No timestamp set, but validation must not fail for
TRIM_HORIZON/LATEST.
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+
DynamodbStreamsSourceConfigUtil.validateStreamSourceConfiguration(
+ sourceConfig));
+ }
+}
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/enumerator/DynamoDbStreamsSourceEnumeratorTest.java
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/enumerator/DynamoDbStreamsSourceEnumeratorTest.java
index 4c543a3b..b255abda 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/enumerator/DynamoDbStreamsSourceEnumeratorTest.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/enumerator/DynamoDbStreamsSourceEnumeratorTest.java
@@ -44,11 +44,13 @@ import
software.amazon.awssdk.services.dynamodb.model.ShardIteratorType;
import java.time.Duration;
import java.time.Instant;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.stream.Stream;
import static
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_POSITION;
+import static
org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants.STREAM_INITIAL_TIMESTAMP;
import static
org.apache.flink.connector.dynamodb.source.util.DynamoDbStreamsProxyProvider.getTestStreamProxy;
import static
org.apache.flink.connector.dynamodb.source.util.TestUtil.generateShard;
import static
org.apache.flink.connector.dynamodb.source.util.TestUtil.generateShardId;
@@ -84,6 +86,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
null);
@@ -140,6 +143,260 @@ class DynamoDbStreamsSourceEnumeratorTest {
}
}
+ @Test
+ void testStartWithAtTimestampListsAllShardsAndAssignsAtTimestamp() throws
Throwable {
+ try (MockSplitEnumeratorContext<DynamoDbStreamsShardSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS)) {
+ DynamoDbStreamsProxyProvider.TestDynamoDbStreamsProxy streamProxy =
+ getTestStreamProxy();
+ final Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ STREAM_INITIAL_POSITION,
+
DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP);
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP,
"2024-01-01T00:00:00.000Z");
+
+ DynamoDbStreamsSourceEnumerator enumerator =
+ new DynamoDbStreamsSourceEnumerator(
+ context,
+ STREAM_ARN,
+ sourceConfig,
+ Instant.parse("2024-01-01T00:00:00.000Z"),
+ streamProxy,
+ ShardAssignerFactory.uniformShardAssigner(),
+ null);
+ enumerator.start();
+
+ final int subtaskId = 1;
+ context.registerReader(TestUtil.getTestReaderInfo(subtaskId));
+ enumerator.addReader(subtaskId);
+ Shard[] shards =
+ new Shard[] {
+ generateShard(0, "1200", null, null), generateShard(1,
"1300", null, null)
+ };
+ streamProxy.addShards(shards);
+
+ // Discovery lists all shards; every one is assigned at
AT_TIMESTAMP.
+ context.runNextOneTimeCallable();
+
+ SplitsAssignment<DynamoDbStreamsShardSplit> initialSplitAssignment
=
+ context.getSplitsAssignmentSequence().get(0);
+ // Every discovered shard is assigned with the AT_TIMESTAMP
starting position; the
+ // per-shard GetShardIterator(AT_TIMESTAMP) resolves the position
at read time.
+ assertThat(
+
initialSplitAssignment.assignment().get(subtaskId).stream()
+
.map(DynamoDbStreamsShardSplit::getStartingPosition))
+ .allSatisfy(
+ s ->
+ assertThat(s.getShardIteratorType())
+
.isEqualTo(ShardIteratorType.AT_TIMESTAMP));
+ }
+ }
+
+ @Test
+ void testAtTimestampUsesTheTimestampResolvedBySource() throws Throwable {
+ try (MockSplitEnumeratorContext<DynamoDbStreamsShardSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS)) {
+ DynamoDbStreamsProxyProvider.TestDynamoDbStreamsProxy streamProxy =
+ getTestStreamProxy();
+ final Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ STREAM_INITIAL_POSITION,
+
DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP);
+ // The enumerator must not re-parse this; it uses the instant the
source resolved.
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP,
"2024-01-01T00:00:00.000Z");
+ final Instant resolved = Instant.parse("2023-06-01T00:00:00Z");
+
+ DynamoDbStreamsSourceEnumerator enumerator =
+ new DynamoDbStreamsSourceEnumerator(
+ context,
+ STREAM_ARN,
+ sourceConfig,
+ resolved,
+ streamProxy,
+ ShardAssignerFactory.uniformShardAssigner(),
+ null);
+ enumerator.start();
+
+ final int subtaskId = 1;
+ context.registerReader(TestUtil.getTestReaderInfo(subtaskId));
+ enumerator.addReader(subtaskId);
+ streamProxy.addShards(generateShard(0, "1200", null, null));
+ context.runNextOneTimeCallable();
+
+ assertThat(
+ context
+ .getSplitsAssignmentSequence()
+ .get(0)
+ .assignment()
+ .get(subtaskId)
+ .stream()
+
.map(DynamoDbStreamsShardSplit::getStartingPosition))
+ .containsExactly(StartingPosition.atTimestamp(resolved));
+
assertThat(enumerator.snapshotState(1L).getStartTimestamp()).isEqualTo(resolved);
+ }
+ }
+
+ @Test
+ void testAtTimestampSecondDiscoveryAddsNoDuplicates() throws Throwable {
+ try (MockSplitEnumeratorContext<DynamoDbStreamsShardSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS)) {
+ DynamoDbStreamsProxyProvider.TestDynamoDbStreamsProxy streamProxy =
+ getTestStreamProxy();
+ final Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ STREAM_INITIAL_POSITION,
+
DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP);
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP,
"2024-01-01T00:00:00.000Z");
+ DynamoDbStreamsSourceEnumerator enumerator =
+ new DynamoDbStreamsSourceEnumerator(
+ context,
+ STREAM_ARN,
+ sourceConfig,
+ Instant.parse("2024-01-01T00:00:00.000Z"),
+ streamProxy,
+ ShardAssignerFactory.uniformShardAssigner(),
+ null);
+ enumerator.start();
+ final int subtaskId = 0;
+ context.registerReader(TestUtil.getTestReaderInfo(subtaskId));
+ enumerator.addReader(subtaskId);
+ streamProxy.addShards(
+ generateShard(0, "1200", null, null), generateShard(1,
"1300", null, null));
+
+ // Bootstrap assigns both shards.
+ context.runNextOneTimeCallable();
+
assertThat(context.getSplitsAssignmentSequence().get(0).assignment().get(subtaskId))
+ .hasSize(2);
+
+ // A periodic shard discovery over the SAME shard set adds/assigns
nothing new: dedup
+ // keeps it idempotent even though the full listing returns every
shard again.
+ context.runPeriodicCallable(0);
+
assertThat(context.getSplitsAssignmentSequence().get(1).assignment().get(subtaskId))
+ .isNullOrEmpty();
+ }
+ }
+
+ @Test
+ void testAtTimestampChildIsGatedBehindParentAndNotDuplicated() throws
Throwable {
+ try (MockSplitEnumeratorContext<DynamoDbStreamsShardSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS)) {
+ DynamoDbStreamsProxyProvider.TestDynamoDbStreamsProxy streamProxy =
+ getTestStreamProxy();
+ final Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ STREAM_INITIAL_POSITION,
+
DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP);
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP,
"2024-01-01T00:00:00.000Z");
+ DynamoDbStreamsSourceEnumerator enumerator =
+ new DynamoDbStreamsSourceEnumerator(
+ context,
+ STREAM_ARN,
+ sourceConfig,
+ Instant.parse("2024-01-01T00:00:00.000Z"),
+ streamProxy,
+ ShardAssignerFactory.uniformShardAssigner(),
+ null);
+ enumerator.start();
+ final int subtaskId = 0;
+ context.registerReader(TestUtil.getTestReaderInfo(subtaskId));
+ enumerator.addReader(subtaskId);
+
+ // Parent (closed) and its child (open) are both present at
bootstrap.
+ Shard parent = generateShard(1, "100", "200", null);
+ Shard child = generateShard(2, "300", null, parent.shardId());
+ streamProxy.addShards(parent, child);
+
+ // Bootstrap: only the parent is assigned; the child is gated
behind it.
+ context.runNextOneTimeCallable();
+ SplitsAssignment<DynamoDbStreamsShardSplit> bootstrap =
+ context.getSplitsAssignmentSequence().get(0);
+ assertThat(
+ bootstrap.assignment().get(subtaskId).stream()
+
.map(DynamoDbStreamsShardSplit::getShardId))
+ .containsExactly(parent.shardId());
+ assertThat(
+ bootstrap.assignment().get(subtaskId).stream()
+
.map(DynamoDbStreamsShardSplit::getStartingPosition)
+
.map(StartingPosition::getShardIteratorType))
+ .containsExactly(ShardIteratorType.AT_TIMESTAMP);
+
+ // When the parent finishes, the child is assigned with the
AT_TIMESTAMP position it
+ // was tracked with at bootstrap; the handoff does not add it
again (dedup).
+ enumerator.handleSourceEvent(
+ subtaskId,
+ new SplitsFinishedEvent(
+ Collections.singleton(
+ new SplitsFinishedEventContext(
+ parent.shardId(),
Collections.singletonList(child)))));
+ SplitsAssignment<DynamoDbStreamsShardSplit> afterFinish =
+ context.getSplitsAssignmentSequence().get(1);
+ assertThat(
+ afterFinish.assignment().get(subtaskId).stream()
+
.map(DynamoDbStreamsShardSplit::getShardId))
+ .containsExactly(child.shardId());
+ assertThat(
+ afterFinish.assignment().get(subtaskId).stream()
+
.map(DynamoDbStreamsShardSplit::getStartingPosition)
+
.map(StartingPosition::getShardIteratorType))
+ .containsExactly(ShardIteratorType.AT_TIMESTAMP);
+ }
+ }
+
+ @Test
+ void testAtTimestampChildDiscoveredViaHandoffGetsAtTimestamp() throws
Throwable {
+ try (MockSplitEnumeratorContext<DynamoDbStreamsShardSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS)) {
+ DynamoDbStreamsProxyProvider.TestDynamoDbStreamsProxy streamProxy =
+ getTestStreamProxy();
+ final Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ STREAM_INITIAL_POSITION,
+
DynamodbStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP);
+ sourceConfig.set(STREAM_INITIAL_TIMESTAMP,
"2024-01-01T00:00:00.000Z");
+ DynamoDbStreamsSourceEnumerator enumerator =
+ new DynamoDbStreamsSourceEnumerator(
+ context,
+ STREAM_ARN,
+ sourceConfig,
+ Instant.parse("2024-01-01T00:00:00.000Z"),
+ streamProxy,
+ ShardAssignerFactory.uniformShardAssigner(),
+ null);
+ enumerator.start();
+ final int subtaskId = 0;
+ context.registerReader(TestUtil.getTestReaderInfo(subtaskId));
+ enumerator.addReader(subtaskId);
+
+ // Only the parent is present at bootstrap; the child appears
later via the
+ // finish-handoff.
+ Shard parent = generateShard(1, "100", "200", null);
+ streamProxy.addShards(parent);
+ context.runNextOneTimeCallable();
+
+ Shard child = generateShard(2, "300", null, parent.shardId());
+ enumerator.handleSourceEvent(
+ subtaskId,
+ new SplitsFinishedEvent(
+ Collections.singleton(
+ new SplitsFinishedEventContext(
+ parent.shardId(),
Collections.singletonList(child)))));
+
+ // A child discovered only via the handoff still starts at
AT_TIMESTAMP
+ // (not TRIM_HORIZON), so a straddling shard can never emit
pre-timestamp records.
+ SplitsAssignment<DynamoDbStreamsShardSplit> afterFinish =
+ context.getSplitsAssignmentSequence().get(1);
+ assertThat(
+ afterFinish.assignment().get(subtaskId).stream()
+
.map(DynamoDbStreamsShardSplit::getShardId))
+ .containsExactly(child.shardId());
+ assertThat(
+ afterFinish.assignment().get(subtaskId).stream()
+
.map(DynamoDbStreamsShardSplit::getStartingPosition)
+
.map(StartingPosition::getShardIteratorType))
+ .containsExactly(ShardIteratorType.AT_TIMESTAMP);
+ }
+ }
+
@Test
void
testLatestOnlyAssignsTheLeafNodesAndSkipsParentShardsDuringInitialDiscovery()
throws Throwable {
@@ -158,6 +415,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
null);
@@ -229,6 +487,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
null);
@@ -369,6 +628,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
state);
@@ -472,6 +732,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
null);
@@ -502,6 +763,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
null);
@@ -560,6 +822,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
null);
@@ -626,6 +889,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
null);
@@ -700,6 +964,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
null);
@@ -751,6 +1016,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
null);
@@ -805,6 +1071,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
null);
@@ -826,6 +1093,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
null);
@@ -845,6 +1113,7 @@ class DynamoDbStreamsSourceEnumeratorTest {
context,
STREAM_ARN,
sourceConfig,
+ null,
streamProxy,
ShardAssignerFactory.uniformShardAssigner(),
null);
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTrackerTest.java
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTrackerTest.java
index d2dc8289..7a9e00b4 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTrackerTest.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTrackerTest.java
@@ -23,8 +23,11 @@ import
org.apache.flink.connector.dynamodb.source.enumerator.DynamoDBStreamsShar
import
org.apache.flink.connector.dynamodb.source.enumerator.SplitAssignmentStatus;
import
org.apache.flink.connector.dynamodb.source.split.DynamoDbStreamsShardSplit;
import org.apache.flink.connector.dynamodb.source.split.StartingPosition;
+import org.apache.flink.testutils.logging.LoggerAuditingExtension;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.event.Level;
import software.amazon.awssdk.services.dynamodb.model.Shard;
import java.time.Duration;
@@ -47,6 +50,10 @@ import static
org.assertj.core.api.AssertionsForInterfaceTypes.assertThat;
class SplitTrackerTest {
+ @RegisterExtension
+ final LoggerAuditingExtension splitTrackerLogs =
+ new LoggerAuditingExtension(SplitTracker.class, Level.WARN);
+
@Test
public void testSplitsWithoutParentsAreAvailableToAssign() {
Instant startTime = Instant.now();
@@ -539,4 +546,213 @@ class SplitTrackerTest {
return new DynamoDbStreamsShardSplit(
STREAM_ARN, shard.shardId(), StartingPosition.fromStart(),
shard.parentShardId());
}
+
+ @Test
+ public void testAtTimestampBootstrapAddsAllShardsWithTimestampPosition() {
+ Instant startTime = Instant.now();
+ SplitTracker splitTracker =
+ new SplitTracker(STREAM_ARN, InitialPosition.AT_TIMESTAMP,
startTime);
+
+ List<Shard> shards =
+ Arrays.asList(
+ generateShard(5, "5000", null, null), generateShard(6,
"6000", null, null));
+
+ splitTracker.addSplits(shards);
+
+ assertThat(splitTracker.getKnownSplits()).hasSize(2);
+ splitTracker
+ .getKnownSplits()
+ .values()
+ .forEach(
+ split ->
+ assertThat(split.getStartingPosition())
+
.isEqualTo(StartingPosition.atTimestamp(startTime)));
+ }
+
+ @Test
+ public void testAtTimestampRootShardGetsAtTimestamp() {
+ Instant startTime = Instant.now();
+ SplitTracker splitTracker =
+ new SplitTracker(STREAM_ARN, InitialPosition.AT_TIMESTAMP,
startTime);
+
+ // A shard with no tracked parent starts at AT_TIMESTAMP.
+ splitTracker.addSplits(Collections.singletonList(generateShard(0,
"100", null, null)));
+
+
assertThat(splitTracker.getKnownSplits().get(generateShardId(0)).getStartingPosition())
+ .isEqualTo(StartingPosition.atTimestamp(startTime));
+ }
+
+ @Test
+ public void testAtTimestampChildOfPreExistingParentGetsAtTimestamp() {
+ Instant startTime = Instant.now();
+ SplitTracker splitTracker =
+ new SplitTracker(STREAM_ARN, InitialPosition.AT_TIMESTAMP,
startTime);
+
+ // Cycle 1: parent (root) is tracked.
+ splitTracker.addSplits(Collections.singletonList(generateShard(0,
"100", null, null)));
+
+ // Cycle 2: parent has closed and a child appears. The child starts at
+ // AT_TIMESTAMP (not TRIM_HORIZON) regardless of the parent being
tracked before this cycle.
+ splitTracker.addSplits(
+ Arrays.asList(
+ generateShard(0, "100", "200", null),
+ generateShard(1, "200", null, generateShardId(0))));
+
+
assertThat(splitTracker.getKnownSplits().get(generateShardId(1)).getStartingPosition())
+ .isEqualTo(StartingPosition.atTimestamp(startTime));
+ }
+
+ @Test
+ public void
testAtTimestampParentAndChildAddedInSameCycleBothStartAtTimestamp() {
+ Instant startTime = Instant.now();
+ SplitTracker splitTracker =
+ new SplitTracker(STREAM_ARN, InitialPosition.AT_TIMESTAMP,
startTime);
+
+ // Cycle 1: grandparent (root) is tracked.
+ splitTracker.addSplits(Collections.singletonList(generateShard(0,
"100", null, null)));
+
+ // Cycle 2: grandparent(0) closed; parent(1) and child(2) both appear
new in this cycle.
+ splitTracker.addSplits(
+ Arrays.asList(
+ generateShard(0, "100", "200", null),
+ generateShard(1, "200", "300", generateShardId(0)),
+ generateShard(2, "300", null, generateShardId(1))));
+
+ // parent(1): every shard starts at AT_TIMESTAMP.
+
assertThat(splitTracker.getKnownSplits().get(generateShardId(1)).getStartingPosition())
+ .isEqualTo(StartingPosition.atTimestamp(startTime));
+ // child(2): also AT_TIMESTAMP, though its parent(1) was added in the
same cycle.
+
assertThat(splitTracker.getKnownSplits().get(generateShardId(2)).getStartingPosition())
+ .isEqualTo(StartingPosition.atTimestamp(startTime));
+ }
+
+ @Test
+ public void testAtTimestampPeriodicDiscoverySkipsAncestors() {
+ Instant startTime = Instant.now();
+ SplitTracker splitTracker =
+ new SplitTracker(STREAM_ARN, InitialPosition.AT_TIMESTAMP,
startTime);
+
+ // Bootstrap: only shards 5 and 6 (active at T)
+ List<Shard> bootstrapShards =
+ Arrays.asList(
+ generateShard(5, "5000", null, generateShardId(3)),
+ generateShard(6, "6000", null, generateShardId(3)));
+ splitTracker.addSplits(bootstrapShards);
+
+ assertThat(splitTracker.getKnownSplits()).hasSize(2);
+
+ // Periodic discovery: returns all shards including ancestors
+ List<Shard> allShards =
+ Arrays.asList(
+ generateShard(0, "100", "500", null), // root
+ generateShard(1, "500", "1000", generateShardId(0)),
// child of 0
+ generateShard(3, "1000", "3000", generateShardId(1)),
// parent of 5,6
+ generateShard(5, "5000", null, generateShardId(3)),
+ generateShard(6, "6000", null, generateShardId(3)));
+ splitTracker.addSplits(allShards);
+
+ // Ancestors 0, 1, 3 should be skipped
+ assertThat(splitTracker.getKnownSplits()).hasSize(2);
+
assertThat(splitTracker.getKnownSplits().containsKey(generateShardId(0))).isFalse();
+
assertThat(splitTracker.getKnownSplits().containsKey(generateShardId(1))).isFalse();
+
assertThat(splitTracker.getKnownSplits().containsKey(generateShardId(3))).isFalse();
+ }
+
+ @Test
+ public void testAtTimestampWarnsOnceForSkippedUntrackedAncestor() {
+ Instant startTime = Instant.now();
+ SplitTracker splitTracker =
+ new SplitTracker(STREAM_ARN, InitialPosition.AT_TIMESTAMP,
startTime);
+
+ // Only the child is tracked at bootstrap; its parent (shard 0) is
never tracked.
+ splitTracker.addSplits(
+ Collections.singletonList(generateShard(1, "200", null,
generateShardId(0))));
+
+ List<Shard> allShards =
+ Arrays.asList(
+ generateShard(0, "100", "200", null),
+ generateShard(1, "200", null, generateShardId(0)));
+ // Several periodic discoveries keep listing the skipped parent.
+ splitTracker.addSplits(allShards);
+ splitTracker.addSplits(allShards);
+ splitTracker.addSplits(allShards);
+
+
assertThat(splitTracker.getKnownSplits()).doesNotContainKey(generateShardId(0));
+ assertThat(splitTrackerLogs.getMessages())
+ .filteredOn(message -> message.contains(generateShardId(0)))
+ .hasSize(1);
+ }
+
+ @Test
+ public void testAtTimestampMissedShardAddedByPeriodicDiscovery() {
+ Instant startTime = Instant.now();
+ SplitTracker splitTracker =
+ new SplitTracker(STREAM_ARN, InitialPosition.AT_TIMESTAMP,
startTime);
+
+ // Bootstrap: shard 5 and 6 returned, shard 7 missed
+ // Shard 0
+ // / \
+ // Shard 1 Shard 2
+ // | |
+ // Shard 3 Shard 7 (missed)
+ // / \
+ // 5 6
+ List<Shard> bootstrapShards =
+ Arrays.asList(
+ generateShard(5, "5000", null, generateShardId(3)),
+ generateShard(6, "6000", null, generateShardId(3)));
+ splitTracker.addSplits(bootstrapShards);
+
+ // Periodic discovery returns all including missed shard 7
+ List<Shard> allShards =
+ Arrays.asList(
+ generateShard(0, "100", "500", null),
+ generateShard(1, "500", "1000", generateShardId(0)),
+ generateShard(2, "500", "4000", generateShardId(0)),
+ generateShard(3, "1000", "3000", generateShardId(1)),
+ generateShard(5, "5000", null, generateShardId(3)),
+ generateShard(6, "6000", null, generateShardId(3)),
+ generateShard(7, "7000", null, generateShardId(2)));
+ splitTracker.addSplits(allShards);
+
+ // Shard 7 should be added at AT_TIMESTAMP (missed, not an ancestor of
anything tracked).
+
assertThat(splitTracker.getKnownSplits().containsKey(generateShardId(7))).isTrue();
+
assertThat(splitTracker.getKnownSplits().get(generateShardId(7)).getStartingPosition())
+ .isEqualTo(StartingPosition.atTimestamp(startTime));
+
+ // Shard 2 should also be added (parent of missed shard 7). Its parent
(shard 0) is neither
+ // tracked before this cycle nor added this cycle, so shard 2 starts
at AT_TIMESTAMP.
+
assertThat(splitTracker.getKnownSplits().containsKey(generateShardId(2))).isTrue();
+
assertThat(splitTracker.getKnownSplits().get(generateShardId(2)).getStartingPosition())
+ .isEqualTo(StartingPosition.atTimestamp(startTime));
+
+ // Ancestors of tracked shards (0, 1, 3) should be skipped
+
assertThat(splitTracker.getKnownSplits().containsKey(generateShardId(0))).isFalse();
+
assertThat(splitTracker.getKnownSplits().containsKey(generateShardId(1))).isFalse();
+
assertThat(splitTracker.getKnownSplits().containsKey(generateShardId(3))).isFalse();
+ }
+
+ @Test
+ public void testAtTimestampChildSplitsGetTimestampPosition() {
+ Instant startTime = Instant.now();
+ SplitTracker splitTracker =
+ new SplitTracker(STREAM_ARN, InitialPosition.AT_TIMESTAMP,
startTime);
+
+ List<Shard> bootstrapShards =
+ Collections.singletonList(generateShard(5, "5000", null,
null));
+ splitTracker.addSplits(bootstrapShards);
+
+ // Child shards discovered via the finish-handoff. They start at
+ // AT_TIMESTAMP so a straddling child never emits pre-timestamp
records.
+ List<Shard> childShards =
+ Arrays.asList(
+ generateShard(8, "8000", null, generateShardId(5)),
+ generateShard(9, "9000", null, generateShardId(5)));
+ splitTracker.addChildSplits(childShards);
+
+
assertThat(splitTracker.getKnownSplits().get(generateShardId(8)).getStartingPosition())
+ .isEqualTo(StartingPosition.atTimestamp(startTime));
+
assertThat(splitTracker.getKnownSplits().get(generateShardId(9)).getStartingPosition())
+ .isEqualTo(StartingPosition.atTimestamp(startTime));
+ }
}
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/proxy/DynamoDbStreamsProxyTest.java
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/proxy/DynamoDbStreamsProxyTest.java
index 9c662a77..c5306064 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/proxy/DynamoDbStreamsProxyTest.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/proxy/DynamoDbStreamsProxyTest.java
@@ -45,6 +45,7 @@ import
software.amazon.awssdk.services.dynamodb.model.StreamDescription;
import software.amazon.awssdk.services.dynamodb.model.StreamStatus;
import
software.amazon.awssdk.services.dynamodb.model.TrimmedDataAccessException;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -442,6 +443,44 @@ public class DynamoDbStreamsProxyTest {
return shards;
}
+ @Test
+ void testGetRecordsInitialReadFromAtTimestamp() {
+ final String streamArn =
+
"arn:aws:dynamodb:us-east-1:1231231230:table/test/stream/2024-01-01T00:00:00.826";
+ final String shardId = "shardId-000000000002";
+ final Instant atTimestamp = Instant.ofEpochMilli(1704067200000L);
+ final StartingPosition startingPosition =
StartingPosition.atTimestamp(atTimestamp);
+
+ final String expectedShardIterator = "some-shard-iterator";
+ final GetRecordsResponse expectedGetRecordsResponse =
+ GetRecordsResponse.builder()
+ .records(Record.builder().build())
+ .nextShardIterator("next-iterator")
+ .build();
+
+ TestingDynamoDbStreamsClient testingDynamoDbStreamsClient =
+ new TestingDynamoDbStreamsClient();
+
testingDynamoDbStreamsClient.setNextShardIterator(expectedShardIterator);
+ testingDynamoDbStreamsClient.setShardIteratorValidation(
+ validateEqual(
+ GetShardIteratorRequest.builder()
+ .streamArn(streamArn)
+ .shardId(shardId)
+
.shardIteratorType(ShardIteratorType.AT_TIMESTAMP)
+ .timestamp(atTimestamp)
+ .build()));
+
testingDynamoDbStreamsClient.setGetRecordsResponse(expectedGetRecordsResponse);
+ testingDynamoDbStreamsClient.setGetRecordsValidation(
+ validateEqual(
+
GetRecordsRequest.builder().shardIterator(expectedShardIterator).build()));
+
+ DynamoDbStreamsProxy dynamoDbStreamsProxy =
+ new DynamoDbStreamsProxy(testingDynamoDbStreamsClient,
HTTP_CLIENT);
+
+ assertThat(dynamoDbStreamsProxy.getRecords(streamArn, shardId,
startingPosition))
+ .isEqualTo(expectedGetRecordsResponse);
+ }
+
private Consumer<DescribeStreamRequest> getDescribeStreamRequestValidation(
final String streamArn, final String startShardId) {
return req -> {
diff --git
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/split/DynamoDbStreamsShardSplitSerializerTest.java
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/split/DynamoDbStreamsShardSplitSerializerTest.java
index 410f93bc..ce4630e8 100644
---
a/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/split/DynamoDbStreamsShardSplitSerializerTest.java
+++
b/flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/split/DynamoDbStreamsShardSplitSerializerTest.java
@@ -24,6 +24,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
+import java.time.Instant;
import java.util.stream.Stream;
import static
org.apache.flink.connector.dynamodb.source.util.TestUtil.SHARD_ID;
@@ -76,11 +77,38 @@ class DynamoDbStreamsShardSplitSerializerTest {
assertThat(deserializedSplit).usingRecursiveComparison().isEqualTo(initialSplit);
}
+ @Test
+ void testSerializeAtTimestampUsesVersion3AndRoundTrips() throws Exception {
+ final DynamoDbStreamsShardSplit initialSplit =
+
getTestSplit(StartingPosition.atTimestamp(Instant.ofEpochMilli(1704067200000L)));
+ DynamoDbStreamsShardSplitSerializer serializer = new
DynamoDbStreamsShardSplitSerializer();
+
+ assertThat(serializer.getVersion()).isEqualTo(3);
+ byte[] serialized = serializer.serialize(initialSplit);
+ DynamoDbStreamsShardSplit deserialized =
+ serializer.deserialize(serializer.getVersion(), serialized);
+
assertThat(deserialized).usingRecursiveComparison().isEqualTo(initialSplit);
+ }
+
+ @Test
+ void testBackwardCompatibilityVersion2SplitDeserializesUnderVersion3()
throws Exception {
+ // A split without an Instant marker is written identically under
schema v2 and v3, so the
+ // v3 deserializer must read data tagged as version 2 (an older
checkpoint) correctly.
+ final DynamoDbStreamsShardSplit initialSplit =
+
getTestSplit(StartingPosition.continueFromSequenceNumber("some-sequence-number"));
+ DynamoDbStreamsShardSplitSerializer serializer = new
DynamoDbStreamsShardSplitSerializer();
+
+ byte[] serialized = serializer.serialize(initialSplit);
+ DynamoDbStreamsShardSplit deserialized = serializer.deserialize(2,
serialized);
+
assertThat(deserialized).usingRecursiveComparison().isEqualTo(initialSplit);
+ }
+
private static Stream<StartingPosition> provideStartingPositions() {
return Stream.of(
StartingPosition.fromStart(),
StartingPosition.continueFromSequenceNumber("some-sequence-number"),
- StartingPosition.latest());
+ StartingPosition.latest(),
+
StartingPosition.atTimestamp(Instant.ofEpochMilli(1704067200000L)));
}
@Test
diff --git
a/flink-connector-aws/flink-connector-sqs/src/test/java/org/apache/flink/connector/sqs/sink/SqsSinkWriterTest.java
b/flink-connector-aws/flink-connector-sqs/src/test/java/org/apache/flink/connector/sqs/sink/SqsSinkWriterTest.java
index c8d7073e..8ea60f97 100644
---
a/flink-connector-aws/flink-connector-sqs/src/test/java/org/apache/flink/connector/sqs/sink/SqsSinkWriterTest.java
+++
b/flink-connector-aws/flink-connector-sqs/src/test/java/org/apache/flink/connector/sqs/sink/SqsSinkWriterTest.java
@@ -223,7 +223,7 @@ public class SqsSinkWriterTest {
final boolean failOnError, TestSqsAsyncClientProvider
testSqsAsyncClientProvider)
throws IOException {
TestSinkInitContext sinkInitContext = new TestSinkInitContext();
- Properties sinkProperties =
AWSServicesTestUtils.createConfig("https://fake_aws_endpoint";);
+ Properties sinkProperties =
AWSServicesTestUtils.createConfig("https://fake-aws-endpoint";);
SqsSink<String> sink =
new SqsSink<>(
ELEMENT_CONVERTER_PLACEHOLDER,
diff --git
a/flink-connector-aws/flink-sql-connector-aws-kinesis-firehose/pom.xml
b/flink-connector-aws/flink-sql-connector-aws-kinesis-firehose/pom.xml
index b18f0934..694a5131 100644
--- a/flink-connector-aws/flink-sql-connector-aws-kinesis-firehose/pom.xml
+++ b/flink-connector-aws/flink-sql-connector-aws-kinesis-firehose/pom.xml
@@ -68,6 +68,9 @@
<include>com.typesafe.netty:*</include>
<include>commons-logging:commons-logging</include>
</includes>
+ <excludes>
+
<exclude>software.amazon.awssdk:apache5-client</exclude>
+ </excludes>
</artifactSet>
<relocations>
<relocation>
diff --git
a/flink-connector-aws/flink-sql-connector-aws-kinesis-firehose/src/main/resources/META-INF/NOTICE
b/flink-connector-aws/flink-sql-connector-aws-kinesis-firehose/src/main/resources/META-INF/NOTICE
index f9ae9dd2..270ce6d6 100644
---
a/flink-connector-aws/flink-sql-connector-aws-kinesis-firehose/src/main/resources/META-INF/NOTICE
+++
b/flink-connector-aws/flink-sql-connector-aws-kinesis-firehose/src/main/resources/META-INF/NOTICE
@@ -7,34 +7,36 @@ The Apache Software Foundation (http://www.apache.org/).
This project bundles the following dependencies under the Apache Software
License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt)
-- software.amazon.awssdk:checksums-spi:2.40.3
-- software.amazon.awssdk:checksums:2.40.3
-- software.amazon.awssdk:utils:2.40.3
-- software.amazon.awssdk:third-party-jackson-core:2.40.3
-- software.amazon.awssdk:sts:2.40.3
-- software.amazon.awssdk:sdk-core:2.40.3
-- software.amazon.awssdk:regions:2.40.3
-- software.amazon.awssdk:retries-spi:2.40.3
-- software.amazon.awssdk:retries:2.40.3
-- software.amazon.awssdk:protocol-core:2.40.3
-- software.amazon.awssdk:profiles:2.40.3
-- software.amazon.awssdk:netty-nio-client:2.40.3
-- software.amazon.awssdk:metrics-spi:2.40.3
-- software.amazon.awssdk:json-utils:2.40.3
-- software.amazon.awssdk:http-client-spi:2.40.3
-- software.amazon.awssdk:http-auth-aws:2.40.3
-- software.amazon.awssdk:http-auth-spi:2.40.3
-- software.amazon.awssdk:http-auth:2.40.3
-- software.amazon.awssdk:identity-spi:2.40.3
-- software.amazon.awssdk:firehose:2.40.3
-- software.amazon.awssdk:endpoints-spi:2.40.3
-- software.amazon.awssdk:aws-query-protocol:2.40.3
-- software.amazon.awssdk:aws-json-protocol:2.40.3
-- software.amazon.awssdk:aws-core:2.40.3
-- software.amazon.awssdk:auth:2.40.3
-- software.amazon.awssdk:apache-client:2.40.3
-- software.amazon.awssdk:annotations:2.40.3
-- software.amazon.awssdk:arns:2.40.3
+- software.amazon.awssdk:checksums-spi:2.56.0
+- software.amazon.awssdk:checksums:2.56.0
+- software.amazon.awssdk:utils:2.56.0
+- software.amazon.awssdk:third-party-jackson-core:2.56.0
+- software.amazon.awssdk:sts:2.56.0
+- software.amazon.awssdk:sdk-core:2.56.0
+- software.amazon.awssdk:regions:2.56.0
+- software.amazon.awssdk:retries-spi:2.56.0
+- software.amazon.awssdk:retries:2.56.0
+- software.amazon.awssdk:protocol-core:2.56.0
+- software.amazon.awssdk:profiles:2.56.0
+- software.amazon.awssdk:netty-nio-client:2.56.0
+- software.amazon.awssdk:metrics-spi:2.56.0
+- software.amazon.awssdk:json-utils:2.56.0
+- software.amazon.awssdk:http-client-spi:2.56.0
+- software.amazon.awssdk:http-auth-aws:2.56.0
+- software.amazon.awssdk:http-auth-spi:2.56.0
+- software.amazon.awssdk:http-auth:2.56.0
+- software.amazon.awssdk:identity-spi:2.56.0
+- software.amazon.awssdk:firehose:2.56.0
+- software.amazon.awssdk:endpoints-spi:2.56.0
+- software.amazon.awssdk:aws-query-protocol:2.56.0
+- software.amazon.awssdk:aws-json-protocol:2.56.0
+- software.amazon.awssdk:aws-core:2.56.0
+- software.amazon.awssdk:auth:2.56.0
+- software.amazon.awssdk:apache-client:2.56.0
+- software.amazon.awssdk:annotations:2.56.0
+- software.amazon.awssdk:arns:2.56.0
+- software.amazon.awssdk:http-auth-aws-eventstream:2.56.0
+- software.amazon.awssdk:utils-lite:2.56.0
- org.apache.httpcomponents:httpcore:4.4.14
- org.apache.httpcomponents:httpclient:4.5.13
- io.netty:netty-transport:4.1.86.Final
diff --git
a/flink-connector-aws/flink-sql-connector-aws-kinesis-streams/pom.xml
b/flink-connector-aws/flink-sql-connector-aws-kinesis-streams/pom.xml
index 29b66050..e86169c5 100644
--- a/flink-connector-aws/flink-sql-connector-aws-kinesis-streams/pom.xml
+++ b/flink-connector-aws/flink-sql-connector-aws-kinesis-streams/pom.xml
@@ -68,6 +68,9 @@
<include>io.netty:*</include>
<include>commons-logging:commons-logging</include>
</includes>
+ <excludes>
+
<exclude>software.amazon.awssdk:apache5-client</exclude>
+ </excludes>
</artifactSet>
<relocations>
<relocation>
diff --git
a/flink-connector-aws/flink-sql-connector-aws-kinesis-streams/src/main/resources/META-INF/NOTICE
b/flink-connector-aws/flink-sql-connector-aws-kinesis-streams/src/main/resources/META-INF/NOTICE
index 36a9fad0..d2437fcb 100644
---
a/flink-connector-aws/flink-sql-connector-aws-kinesis-streams/src/main/resources/META-INF/NOTICE
+++
b/flink-connector-aws/flink-sql-connector-aws-kinesis-streams/src/main/resources/META-INF/NOTICE
@@ -7,36 +7,38 @@ The Apache Software Foundation (http://www.apache.org/).
This project bundles the following dependencies under the Apache Software
License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt)
-- software.amazon.awssdk:checksums-spi:2.40.3
-- software.amazon.awssdk:checksums:2.40.3
-- software.amazon.awssdk:utils:2.40.3
-- software.amazon.awssdk:third-party-jackson-dataformat-cbor:2.40.3
-- software.amazon.awssdk:third-party-jackson-core:2.40.3
-- software.amazon.awssdk:sts:2.40.3
-- software.amazon.awssdk:sdk-core:2.40.3
-- software.amazon.awssdk:regions:2.40.3
-- software.amazon.awssdk:retries-spi:2.40.3
-- software.amazon.awssdk:retries:2.40.3
-- software.amazon.awssdk:protocol-core:2.40.3
-- software.amazon.awssdk:profiles:2.40.3
-- software.amazon.awssdk:netty-nio-client:2.40.3
-- software.amazon.awssdk:metrics-spi:2.40.3
-- software.amazon.awssdk:kinesis:2.40.3
-- software.amazon.awssdk:json-utils:2.40.3
-- software.amazon.awssdk:http-client-spi:2.40.3
-- software.amazon.awssdk:http-auth-aws:2.40.3
-- software.amazon.awssdk:http-auth-spi:2.40.3
-- software.amazon.awssdk:http-auth:2.40.3
-- software.amazon.awssdk:identity-spi:2.40.3
-- software.amazon.awssdk:endpoints-spi:2.40.3
-- software.amazon.awssdk:aws-query-protocol:2.40.3
-- software.amazon.awssdk:aws-json-protocol:2.40.3
-- software.amazon.awssdk:aws-core:2.40.3
-- software.amazon.awssdk:aws-cbor-protocol:2.40.3
-- software.amazon.awssdk:auth:2.40.3
-- software.amazon.awssdk:arns:2.40.3
-- software.amazon.awssdk:apache-client:2.40.3
-- software.amazon.awssdk:annotations:2.40.3
+- software.amazon.awssdk:checksums-spi:2.56.0
+- software.amazon.awssdk:checksums:2.56.0
+- software.amazon.awssdk:utils:2.56.0
+- software.amazon.awssdk:third-party-jackson-dataformat-cbor:2.56.0
+- software.amazon.awssdk:third-party-jackson-core:2.56.0
+- software.amazon.awssdk:sts:2.56.0
+- software.amazon.awssdk:sdk-core:2.56.0
+- software.amazon.awssdk:regions:2.56.0
+- software.amazon.awssdk:retries-spi:2.56.0
+- software.amazon.awssdk:retries:2.56.0
+- software.amazon.awssdk:protocol-core:2.56.0
+- software.amazon.awssdk:profiles:2.56.0
+- software.amazon.awssdk:netty-nio-client:2.56.0
+- software.amazon.awssdk:metrics-spi:2.56.0
+- software.amazon.awssdk:kinesis:2.56.0
+- software.amazon.awssdk:json-utils:2.56.0
+- software.amazon.awssdk:http-client-spi:2.56.0
+- software.amazon.awssdk:http-auth-aws:2.56.0
+- software.amazon.awssdk:http-auth-spi:2.56.0
+- software.amazon.awssdk:http-auth:2.56.0
+- software.amazon.awssdk:identity-spi:2.56.0
+- software.amazon.awssdk:endpoints-spi:2.56.0
+- software.amazon.awssdk:aws-query-protocol:2.56.0
+- software.amazon.awssdk:aws-json-protocol:2.56.0
+- software.amazon.awssdk:aws-core:2.56.0
+- software.amazon.awssdk:aws-cbor-protocol:2.56.0
+- software.amazon.awssdk:auth:2.56.0
+- software.amazon.awssdk:arns:2.56.0
+- software.amazon.awssdk:apache-client:2.56.0
+- software.amazon.awssdk:annotations:2.56.0
+- software.amazon.awssdk:http-auth-aws-eventstream:2.56.0
+- software.amazon.awssdk:utils-lite:2.56.0
- org.apache.httpcomponents:httpcore:4.4.14
- org.apache.httpcomponents:httpclient:4.5.13
- io.netty:netty-transport:4.1.86.Final
diff --git a/flink-connector-aws/flink-sql-connector-dynamodb/pom.xml
b/flink-connector-aws/flink-sql-connector-dynamodb/pom.xml
index f14b2e2a..93e8cd65 100644
--- a/flink-connector-aws/flink-sql-connector-dynamodb/pom.xml
+++ b/flink-connector-aws/flink-sql-connector-dynamodb/pom.xml
@@ -78,6 +78,9 @@ under the License.
<include>commons-logging:commons-logging</include>
<include>commons-codec:commons-codec</include>
</includes>
+ <excludes>
+
<exclude>software.amazon.awssdk:apache5-client</exclude>
+ </excludes>
</artifactSet>
<relocations>
<relocation>
diff --git
a/flink-connector-aws/flink-sql-connector-dynamodb/src/main/resources/META-INF/NOTICE
b/flink-connector-aws/flink-sql-connector-dynamodb/src/main/resources/META-INF/NOTICE
index 558a8784..aa443902 100644
---
a/flink-connector-aws/flink-sql-connector-dynamodb/src/main/resources/META-INF/NOTICE
+++
b/flink-connector-aws/flink-sql-connector-dynamodb/src/main/resources/META-INF/NOTICE
@@ -7,35 +7,37 @@ The Apache Software Foundation (http://www.apache.org/).
This project bundles the following dependencies under the Apache Software
License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt)
-- software.amazon.awssdk:checksums-spi:2.40.3
-- software.amazon.awssdk:checksums:2.40.3
-- software.amazon.awssdk:utils:2.40.3
-- software.amazon.awssdk:third-party-jackson-core:2.40.3
-- software.amazon.awssdk:sts:2.40.3
-- software.amazon.awssdk:sdk-core:2.40.3
-- software.amazon.awssdk:regions:2.40.3
-- software.amazon.awssdk:retries-spi:2.40.3
-- software.amazon.awssdk:retries:2.40.3
-- software.amazon.awssdk:protocol-core:2.40.3
-- software.amazon.awssdk:profiles:2.40.3
-- software.amazon.awssdk:netty-nio-client:2.40.3
-- software.amazon.awssdk:metrics-spi:2.40.3
-- software.amazon.awssdk:json-utils:2.40.3
-- software.amazon.awssdk:http-client-spi:2.40.3
-- software.amazon.awssdk:http-auth-aws:2.40.3
-- software.amazon.awssdk:http-auth-spi:2.40.3
-- software.amazon.awssdk:http-auth:2.40.3
-- software.amazon.awssdk:identity-spi:2.40.3
-- software.amazon.awssdk:endpoints-spi:2.40.3
-- software.amazon.awssdk:dynamodb:2.40.3
-- software.amazon.awssdk:dynamodb-enhanced:2.40.3
-- software.amazon.awssdk:aws-query-protocol:2.40.3
-- software.amazon.awssdk:aws-json-protocol:2.40.3
-- software.amazon.awssdk:aws-core:2.40.3
-- software.amazon.awssdk:auth:2.40.3
-- software.amazon.awssdk:arns:2.40.3
-- software.amazon.awssdk:apache-client:2.40.3
-- software.amazon.awssdk:annotations:2.40.3
+- software.amazon.awssdk:checksums-spi:2.56.0
+- software.amazon.awssdk:checksums:2.56.0
+- software.amazon.awssdk:utils:2.56.0
+- software.amazon.awssdk:third-party-jackson-core:2.56.0
+- software.amazon.awssdk:sts:2.56.0
+- software.amazon.awssdk:sdk-core:2.56.0
+- software.amazon.awssdk:regions:2.56.0
+- software.amazon.awssdk:retries-spi:2.56.0
+- software.amazon.awssdk:retries:2.56.0
+- software.amazon.awssdk:protocol-core:2.56.0
+- software.amazon.awssdk:profiles:2.56.0
+- software.amazon.awssdk:netty-nio-client:2.56.0
+- software.amazon.awssdk:metrics-spi:2.56.0
+- software.amazon.awssdk:json-utils:2.56.0
+- software.amazon.awssdk:http-client-spi:2.56.0
+- software.amazon.awssdk:http-auth-aws:2.56.0
+- software.amazon.awssdk:http-auth-spi:2.56.0
+- software.amazon.awssdk:http-auth:2.56.0
+- software.amazon.awssdk:identity-spi:2.56.0
+- software.amazon.awssdk:endpoints-spi:2.56.0
+- software.amazon.awssdk:dynamodb:2.56.0
+- software.amazon.awssdk:dynamodb-enhanced:2.56.0
+- software.amazon.awssdk:aws-query-protocol:2.56.0
+- software.amazon.awssdk:aws-json-protocol:2.56.0
+- software.amazon.awssdk:aws-core:2.56.0
+- software.amazon.awssdk:auth:2.56.0
+- software.amazon.awssdk:arns:2.56.0
+- software.amazon.awssdk:apache-client:2.56.0
+- software.amazon.awssdk:annotations:2.56.0
+- software.amazon.awssdk:http-auth-aws-eventstream:2.56.0
+- software.amazon.awssdk:utils-lite:2.56.0
- org.apache.httpcomponents:httpcore:4.4.14
- org.apache.httpcomponents:httpclient:4.5.13
- io.netty:netty-transport:4.1.86.Final
diff --git a/pom.xml b/pom.xml
index 098c169e..9a8d5158 100644
--- a/pom.xml
+++ b/pom.xml
@@ -53,7 +53,7 @@ under the License.
</scm>
<properties>
- <aws.sdkv2.version>2.40.3</aws.sdkv2.version>
+ <aws.sdkv2.version>2.56.0</aws.sdkv2.version>
<netty.version>4.1.86.Final</netty.version>
<flink.version>2.1.2</flink.version>
<jackson-bom.version>2.14.3</jackson-bom.version>
