riyarawat-amazon commented on code in PR #258:
URL:
https://github.com/apache/flink-connector-aws/pull/258#discussion_r4326986161
##########
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:
Done: renamed the test to
testAtTimestampParentAndChildAddedInSameCycleBothStartAtTimestamp, and reworded
the stale "inherits" and child-handoff TRIM_HORIZON comments, as every shard
now starts at AT_TIMESTAMP. Also fixed the broken line wrap in EnumeratorTest.
--
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]
