ferenc-csaky commented on code in PR #258:
URL:
https://github.com/apache/flink-connector-aws/pull/258#discussion_r4323266739
##########
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";);
Review Comment:
Why is the `fake_aws_endpoint` → `fake-aws-endpoint` change needed? These
tests only build the writer and never send anything to that endpoint, and
nothing else in the PR seems to depend on it. If it isn't needed, please revert
it to keep the diff to the feature. (Same in `KinesisStreamsSinkWriterTest` and
`SqsSinkWriterTest`.)
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/DynamoDbStreamsSourceEnumerator.java:
##########
@@ -93,7 +94,10 @@ public 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
+ ?
DynamodbStreamsSourceConfigUtil.parseInitialTimestamp(sourceConfig)
Review Comment:
The timestamp is parsed twice. Once in the source constructor, on the client
building the job graph, for validation. Then again here on the JobManager,
which gives the T that's actually used.
With a custom `flink.stream.initpos.timestamp.format` that has no zone (e.g.
`yyyy-MM-dd HH:mm:ss`), `SimpleDateFormat` uses the JVM default time zone, so
the two parses can give different instants. Example: a client on UTC+2 submits
`2026-10-20 10:00:00` at 10:30 local time. Validation reads it as 08:00Z, which
is in the past, so it passes. A JobManager on UTC then reads it as 10:00Z, 1.5
hours in the future. That's exactly the restart loop the validation is meant to
prevent, and in milder cases the job just starts from a different point than
was checked.
Could we resolve the `Instant` once in the `DynamoDbStreamsSource`
constructor, keep it as a field, and pass it to the enumerator? Then the
validated value and the value used are the same.
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -179,13 +196,68 @@ private void addSplitToMapping(DynamoDbStreamsShardSplit
split) {
.add(split.splitId());
}
+ /**
+ * Adds shards discovered under the AT_TIMESTAMP initial position. Each
newly discovered shard
+ * (one that is not already tracked and is not an ancestor of an
already-tracked shard) is added
+ * at AT_TIMESTAMP: the service resolves the correct per-record position
via
+ * GetShardIterator(AT_TIMESTAMP), so a shard that straddles the timestamp
never emits
+ * pre-timestamp records and the position is independent of the discovery
path or restore
+ * timing. Ancestors of already-tracked shards are skipped because they
have already been
Review Comment:
The Javadoc says ancestors of tracked shards are skipped "because they have
already been processed". That isn't always true. In
`testAtTimestampMissedShardAddedByPeriodicDiscovery`, shard 0 was never tracked
or read. It's skipped only to keep parent-before-child order for 5 and 6, so
any records it has at or after T are dropped. That's a reasonable trade-off,
but could the Javadoc say so, and could we log a WARN when an untracked
ancestor gets skipped, since it means possible data loss?
Smaller things in the same method:
- The inline comment at lines 215–218 repeats the Javadoc.
- The `(a, b) -> a` merge function and the `shardId() != null` filter are
defensive checks the LATEST path doesn't have. Keep them consistent one way or
the other.
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigUtilTest.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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 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.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 testParseInitialTimestampThrowsWhenMissing() {
Review Comment:
Nit: the PR description says unparseable values are covered, but I can't
find a test for something like `"not-a-timestamp"` hitting the "Unable to
parse" path. There's also no test for a custom
`STREAM_INITIAL_TIMESTAMP_FORMAT`. Could you add both?
##########
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>
Review Comment:
Bumping `aws.sdkv2.version` to 2.56.0 leaves the bundled NOTICE files out of
date. These three still list every `software.amazon.awssdk:*` artifact at
`2.40.3`:
- `flink-sql-connector-dynamodb/src/main/resources/META-INF/NOTICE`
-
`flink-sql-connector-aws-kinesis-streams/src/main/resources/META-INF/NOTICE`
-
`flink-sql-connector-aws-kinesis-firehose/src/main/resources/META-INF/NOTICE`
They need updating, including any new transitive SDK artifacts pulled in
between 2.40.3 and 2.56.0. FLINK-35819 did this for the previous bump.
Also, the PR description says "(see note above)" for the dependency, but
there's no note. What's the minimum SDK version that has `AT_TIMESTAMP` for
DynamoDB Streams?
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigUtil.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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 java.text.ParseException;
+import java.text.SimpleDateFormat;
+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;
+
+/** 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 using the configured {@link
+ * DynamodbStreamsSourceConfigConstants#STREAM_INITIAL_TIMESTAMP_FORMAT};
if that fails it is
+ * interpreted as a (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);
+ try {
+ return new
SimpleDateFormat(format).parse(timestampString).toInstant();
+ } catch (ParseException | IllegalArgumentException e) {
+ // Fall back to interpreting the value as epoch seconds (may be
fractional).
+ try {
+ return Instant.ofEpochMilli((long)
(Double.parseDouble(timestampString) * 1000));
Review Comment:
Nit: the epoch-seconds fallback accepts some odd inputs without complaint:
- `"NaN"` becomes epoch 0 and silently behaves like TRIM_HORIZON.
- Negative values are accepted. The Kinesis source's validation message asks
for a "non-negative double value".
Maybe reject non-finite and negative values here.
Also, `SimpleDateFormat` is lenient by default and only parses a prefix of
the string. So `2024-02-30T…` rolls over to March 1, and trailing garbage is
ignored. That matches the Kinesis behaviour, but calling `setLenient(false)`
would cost nothing.
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -298,6 +370,10 @@ public Map<String, DynamoDbStreamsShardSplit>
getKnownSplits() {
return knownSplits;
}
+ public Set<String> getFinishedSplits() {
Review Comment:
`getFinishedSplits()` has no callers in either main or test code, and it
returns the mutable internal set. Please remove it, or if it's needed, mark it
`@VisibleForTesting` and return an unmodifiable view.
##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTrackerTest.java:
##########
@@ -539,4 +539,190 @@ private DynamoDbStreamsShardSplit getSplitFromShard(Shard
shard) {
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 testAtTimestampChildInheritsNewlyAddedParentPosition() {
+ 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): its parent(1) was added in THIS cycle at AT_TIMESTAMP, so
it inherits
Review Comment:
Nit: some test comments look left over from an earlier design where
positions were inherited:
- "so it inherits AT_TIMESTAMP too" (SplitTrackerTest:617), "inherits it"
(:688), and the test name
`testAtTimestampChildInheritsNewlyAddedParentPosition`. Every shard now gets
AT_TIMESTAMP unconditionally.
- "the child-handoff TRIM_HORIZON does not override…" (EnumeratorTest:277).
Under AT_TIMESTAMP the handoff now uses AT_TIMESTAMP.
- EnumeratorTest:224–226 has a broken line wrap ("dedup / keeps sync /
idempotent").
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
