You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "gguptp (via GitHub)" <gi...@apache.org> on 2024/10/13 21:20:00 UTC
[PR] [FLINK-36523][Connectors/DynamoDB][BugFix] Fixing LATEST behaviour for DDB Connector [flink-connector-aws]
gguptp opened a new pull request, #177:
URL: https://github.com/apache/flink-connector-aws/pull/177
<!--
*Thank you for contributing to Apache Flink AWS Connectors - we are happy that you want to help us improve our Flink connectors. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- The name of the pull request should correspond to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Commits should be in the form of "[FLINK-XXXX][component] Title of the pull request", where [FLINK-XXXX] should be replaced by the actual issue number.
Generally, [component] should be the connector you are working on.
For example: "[FLINK-XXXX][Connectors/Kinesis] XXXX" if you are working on the Kinesis connector or "[FLINK-XXXX][Connectors/AWS] XXXX" if you are working on components shared among all the connectors.
- Each pull request should only have one JIRA issue.
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
-->
## Purpose of the change
Fixing latest behaviour for DDB Connector.
## Verifying this change
Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing
*(Please pick either of the following options)*
Added new unit tests
## Significant changes
*(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for convenience.)*
- [ ] Dependencies have been added or upgraded
- [ ] Public API has been changed (Public API is any class annotated with `@Public(Evolving)`)
- [ ] Serializers have been changed
- [ ] New feature has been introduced
- If yes, how is this documented? (not applicable / docs / JavaDocs / not documented)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36523][Connectors/DynamoDB][BugFix] Fixing LATEST behaviour for DDB Connector [flink-connector-aws]
Posted by "gguptp (via GitHub)" <gi...@apache.org>.
gguptp commented on code in PR #177:
URL: https://github.com/apache/flink-connector-aws/pull/177#discussion_r1811942393
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/DynamoDbStreamsSourceEnumerator.java:
##########
@@ -75,6 +76,7 @@ public class DynamoDbStreamsSourceEnumerator
private final DynamoDbStreamsShardAssigner shardAssigner;
private final ShardAssignerContext shardAssignerContext;
private final SplitTracker splitTracker;
+ private Instant startTimestamp;
Review Comment:
ack changing it to final
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36523][Connectors/DynamoDB][BugFix] Fixing LATEST behaviour for DDB Connector [flink-connector-aws]
Posted by "gguptp (via GitHub)" <gi...@apache.org>.
gguptp commented on code in PR #177:
URL: https://github.com/apache/flink-connector-aws/pull/177#discussion_r1811945191
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -89,13 +95,69 @@ public SplitTracker(
* @param shardsToAdd collection of splits to add to tracking
*/
public void addSplits(Collection<Shard> shardsToAdd) {
- Set<String> discoveredShardIds =
- shardsToAdd.stream().map(Shard::shardId).collect(Collectors.toSet());
+ if (TRIM_HORIZON.equals(initialPosition)) {
+ addSplitsForTrimHorizon(shardsToAdd);
+ return;
+ }
+ addSplitsForLatest(shardsToAdd);
+ }
+
+ private void addSplitsForLatest(Collection<Shard> shardsToAdd) {
+ List<Shard> openShards =
+ shardsToAdd.stream()
+ .filter(shard -> shard.sequenceNumberRange().endingSequenceNumber() == null)
+ .collect(Collectors.toList());
+ Map<String, Shard> shardIdToShardMap =
+ shardsToAdd.stream().collect(Collectors.toMap(Shard::shardId, shard -> shard));
+ for (Shard shard : openShards) {
+ String shardId = shard.shardId();
+ if (knownSplits.containsKey(shardId)) {
+ continue;
+ }
+ String firstShardIdBeforeTimestamp =
+ findFirstShardIdBeforeTimestamp(shardId, shardIdToShardMap);
+ putAllAncestorShardsTillFirstShardId(
+ shardId, firstShardIdBeforeTimestamp, shardIdToShardMap);
+ }
+ }
+
+ private void putAllAncestorShardsTillFirstShardId(
+ String shardId,
+ String firstShardIdBeforeTimestamp,
+ Map<String, Shard> shardIdToShardMap) {
+ String currentShardId = shardId;
+ while (currentShardId != null && shardIdToShardMap.containsKey(currentShardId)) {
+ Shard currentShard = shardIdToShardMap.get(currentShardId);
+ if (Objects.equals(currentShardId, firstShardIdBeforeTimestamp)) {
+ DynamoDbStreamsShardSplit newSplit = mapToSplit(currentShard, LATEST);
+ knownSplits.putIfAbsent(currentShardId, newSplit);
Review Comment:
i think if we put it as a break condition of the while loop this shard will not be assigned the LATEST shard iterator type hence i prefer to keep it as-is
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36523][Connectors/DynamoDB][BugFix] Fixing LATEST behaviour for DDB Connector [flink-connector-aws]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #177:
URL: https://github.com/apache/flink-connector-aws/pull/177#discussion_r1810789785
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/DynamoDbStreamsSourceEnumerator.java:
##########
@@ -75,6 +76,7 @@ public class DynamoDbStreamsSourceEnumerator
private final DynamoDbStreamsShardAssigner shardAssigner;
private final ShardAssignerContext shardAssignerContext;
private final SplitTracker splitTracker;
+ private Instant startTimestamp;
Review Comment:
this should be `final`
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -89,13 +95,69 @@ public SplitTracker(
* @param shardsToAdd collection of splits to add to tracking
*/
public void addSplits(Collection<Shard> shardsToAdd) {
- Set<String> discoveredShardIds =
- shardsToAdd.stream().map(Shard::shardId).collect(Collectors.toSet());
+ if (TRIM_HORIZON.equals(initialPosition)) {
+ addSplitsForTrimHorizon(shardsToAdd);
+ return;
+ }
+ addSplitsForLatest(shardsToAdd);
+ }
+
+ private void addSplitsForLatest(Collection<Shard> shardsToAdd) {
+ List<Shard> openShards =
+ shardsToAdd.stream()
+ .filter(shard -> shard.sequenceNumberRange().endingSequenceNumber() == null)
+ .collect(Collectors.toList());
+ Map<String, Shard> shardIdToShardMap =
+ shardsToAdd.stream().collect(Collectors.toMap(Shard::shardId, shard -> shard));
+ for (Shard shard : openShards) {
+ String shardId = shard.shardId();
+ if (knownSplits.containsKey(shardId)) {
+ continue;
+ }
+ String firstShardIdBeforeTimestamp =
+ findFirstShardIdBeforeTimestamp(shardId, shardIdToShardMap);
+ putAllAncestorShardsTillFirstShardId(
+ shardId, firstShardIdBeforeTimestamp, shardIdToShardMap);
+ }
+ }
+
+ private void putAllAncestorShardsTillFirstShardId(
+ String shardId,
+ String firstShardIdBeforeTimestamp,
+ Map<String, Shard> shardIdToShardMap) {
+ String currentShardId = shardId;
+ while (currentShardId != null && shardIdToShardMap.containsKey(currentShardId)) {
+ Shard currentShard = shardIdToShardMap.get(currentShardId);
+ if (Objects.equals(currentShardId, firstShardIdBeforeTimestamp)) {
+ DynamoDbStreamsShardSplit newSplit = mapToSplit(currentShard, LATEST);
+ knownSplits.putIfAbsent(currentShardId, newSplit);
+ break;
+ } else {
+ DynamoDbStreamsShardSplit newSplit = mapToSplit(currentShard, TRIM_HORIZON);
+ knownSplits.putIfAbsent(currentShardId, newSplit);
+ }
+ currentShardId = currentShard.parentShardId();
+ }
+ }
+
+ private String findFirstShardIdBeforeTimestamp(
+ String shardIdToStartWith, Map<String, Shard> shardIdToShardMap) {
+ String shardId = shardIdToStartWith;
+ while (shardId != null && shardIdToShardMap.containsKey(shardId)) {
+ Shard shard = shardIdToShardMap.get(shardId);
Review Comment:
nit: we could optimise this algorithm by doing a check if `split` is in `knownSplits` as a `break` condition of the while loop
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/util/ShardUtils.java:
##########
@@ -78,4 +78,14 @@ public static boolean isShardOlderThanInconsistencyDetectionRetentionPeriod(Stri
.plus(
DDB_STREAMS_MAX_RETENTION_PERIOD_FOR_RESOLVING_INCONSISTENCIES));
}
+
+ /**
+ * Returns true if shard ue if the shard was created before the given timestamp.
Review Comment:
typo!
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -89,13 +95,69 @@ public SplitTracker(
* @param shardsToAdd collection of splits to add to tracking
*/
public void addSplits(Collection<Shard> shardsToAdd) {
- Set<String> discoveredShardIds =
- shardsToAdd.stream().map(Shard::shardId).collect(Collectors.toSet());
+ if (TRIM_HORIZON.equals(initialPosition)) {
+ addSplitsForTrimHorizon(shardsToAdd);
+ return;
+ }
+ addSplitsForLatest(shardsToAdd);
+ }
+
+ private void addSplitsForLatest(Collection<Shard> shardsToAdd) {
+ List<Shard> openShards =
+ shardsToAdd.stream()
+ .filter(shard -> shard.sequenceNumberRange().endingSequenceNumber() == null)
+ .collect(Collectors.toList());
+ Map<String, Shard> shardIdToShardMap =
+ shardsToAdd.stream().collect(Collectors.toMap(Shard::shardId, shard -> shard));
+ for (Shard shard : openShards) {
+ String shardId = shard.shardId();
+ if (knownSplits.containsKey(shardId)) {
+ continue;
+ }
+ String firstShardIdBeforeTimestamp =
+ findFirstShardIdBeforeTimestamp(shardId, shardIdToShardMap);
+ putAllAncestorShardsTillFirstShardId(
+ shardId, firstShardIdBeforeTimestamp, shardIdToShardMap);
+ }
+ }
+
+ private void putAllAncestorShardsTillFirstShardId(
+ String shardId,
+ String firstShardIdBeforeTimestamp,
+ Map<String, Shard> shardIdToShardMap) {
+ String currentShardId = shardId;
+ while (currentShardId != null && shardIdToShardMap.containsKey(currentShardId)) {
+ Shard currentShard = shardIdToShardMap.get(currentShardId);
+ if (Objects.equals(currentShardId, firstShardIdBeforeTimestamp)) {
+ DynamoDbStreamsShardSplit newSplit = mapToSplit(currentShard, LATEST);
Review Comment:
Technically we still drop a bunch of records from the given timestamp, but we need to just make this clear on docs.
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -89,13 +95,69 @@ public SplitTracker(
* @param shardsToAdd collection of splits to add to tracking
*/
public void addSplits(Collection<Shard> shardsToAdd) {
- Set<String> discoveredShardIds =
- shardsToAdd.stream().map(Shard::shardId).collect(Collectors.toSet());
+ if (TRIM_HORIZON.equals(initialPosition)) {
+ addSplitsForTrimHorizon(shardsToAdd);
+ return;
+ }
+ addSplitsForLatest(shardsToAdd);
+ }
+
+ private void addSplitsForLatest(Collection<Shard> shardsToAdd) {
+ List<Shard> openShards =
+ shardsToAdd.stream()
+ .filter(shard -> shard.sequenceNumberRange().endingSequenceNumber() == null)
+ .collect(Collectors.toList());
+ Map<String, Shard> shardIdToShardMap =
+ shardsToAdd.stream().collect(Collectors.toMap(Shard::shardId, shard -> shard));
+ for (Shard shard : openShards) {
+ String shardId = shard.shardId();
+ if (knownSplits.containsKey(shardId)) {
+ continue;
+ }
+ String firstShardIdBeforeTimestamp =
+ findFirstShardIdBeforeTimestamp(shardId, shardIdToShardMap);
+ putAllAncestorShardsTillFirstShardId(
+ shardId, firstShardIdBeforeTimestamp, shardIdToShardMap);
+ }
+ }
+
+ private void putAllAncestorShardsTillFirstShardId(
+ String shardId,
+ String firstShardIdBeforeTimestamp,
+ Map<String, Shard> shardIdToShardMap) {
+ String currentShardId = shardId;
+ while (currentShardId != null && shardIdToShardMap.containsKey(currentShardId)) {
+ Shard currentShard = shardIdToShardMap.get(currentShardId);
+ if (Objects.equals(currentShardId, firstShardIdBeforeTimestamp)) {
+ DynamoDbStreamsShardSplit newSplit = mapToSplit(currentShard, LATEST);
+ knownSplits.putIfAbsent(currentShardId, newSplit);
Review Comment:
nit: we could optimise this algorithm by doing this check as a `break` condition of the while loop
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36523][Connectors/DynamoDB][BugFix] Fixing LATEST behaviour for DDB Connector [flink-connector-aws]
Posted by "gguptp (via GitHub)" <gi...@apache.org>.
gguptp commented on code in PR #177:
URL: https://github.com/apache/flink-connector-aws/pull/177#discussion_r1803887401
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -89,13 +91,77 @@ public SplitTracker(
* @param shardsToAdd collection of splits to add to tracking
*/
public void addSplits(Collection<Shard> shardsToAdd) {
- Set<String> discoveredShardIds =
- shardsToAdd.stream().map(Shard::shardId).collect(Collectors.toSet());
+ if (TRIM_HORIZON.equals(initialPosition)) {
+ addSplitsForTrimHorizon(shardsToAdd);
+ return;
+ }
+ addSplitsForLatest(shardsToAdd);
+ }
+
+ /**
+ * If there is an ancestor of an open shard in knownSplits, we put all of its ancestors after
+ * the tracked ancestors as TRIM_HORIZON otherwise, we ignore all the ancestors and track only
+ * the open shard with LATEST. There might be a case when restoring from expired snapshots that
+ * there is an ancestor in knownSplits, but we wont know that it is an ancestor, so in that case
+ * also, all the ancestors will not be tracked and rather only the current open shard will be
+ * tracked with LATEST.
Review Comment:
changed the algo to resolve this
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36523][Connectors/DynamoDB][BugFix] Fixing LATEST behaviour for DDB Connector [flink-connector-aws]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #177:
URL: https://github.com/apache/flink-connector-aws/pull/177#discussion_r1799315333
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -89,13 +91,77 @@ public SplitTracker(
* @param shardsToAdd collection of splits to add to tracking
*/
public void addSplits(Collection<Shard> shardsToAdd) {
- Set<String> discoveredShardIds =
- shardsToAdd.stream().map(Shard::shardId).collect(Collectors.toSet());
+ if (TRIM_HORIZON.equals(initialPosition)) {
+ addSplitsForTrimHorizon(shardsToAdd);
+ return;
+ }
+ addSplitsForLatest(shardsToAdd);
+ }
+
+ /**
+ * If there is an ancestor of an open shard in knownSplits, we put all of its ancestors after
+ * the tracked ancestors as TRIM_HORIZON otherwise, we ignore all the ancestors and track only
+ * the open shard with LATEST. There might be a case when restoring from expired snapshots that
+ * there is an ancestor in knownSplits, but we wont know that it is an ancestor, so in that case
+ * also, all the ancestors will not be tracked and rather only the current open shard will be
+ * tracked with LATEST.
Review Comment:
We should align that the expected behavior, when using `LATEST`, should be:
1. When starting without snapshot:
- All shards from start of application read from `LATEST`. If we want, we can focus on just the open shards.
- All child shards are read from `TRIM_HORIZON`
2. When reading with snapshot:
- All shards from time of start of application are read from `TRIM_HORIZON`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36523][Connectors/DynamoDB][BugFix] Fixing LATEST behaviour for DDB Connector [flink-connector-aws]
Posted by "gguptp (via GitHub)" <gi...@apache.org>.
gguptp commented on code in PR #177:
URL: https://github.com/apache/flink-connector-aws/pull/177#discussion_r1811943331
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -89,13 +95,69 @@ public SplitTracker(
* @param shardsToAdd collection of splits to add to tracking
*/
public void addSplits(Collection<Shard> shardsToAdd) {
- Set<String> discoveredShardIds =
- shardsToAdd.stream().map(Shard::shardId).collect(Collectors.toSet());
+ if (TRIM_HORIZON.equals(initialPosition)) {
+ addSplitsForTrimHorizon(shardsToAdd);
+ return;
+ }
+ addSplitsForLatest(shardsToAdd);
+ }
+
+ private void addSplitsForLatest(Collection<Shard> shardsToAdd) {
+ List<Shard> openShards =
+ shardsToAdd.stream()
+ .filter(shard -> shard.sequenceNumberRange().endingSequenceNumber() == null)
+ .collect(Collectors.toList());
+ Map<String, Shard> shardIdToShardMap =
+ shardsToAdd.stream().collect(Collectors.toMap(Shard::shardId, shard -> shard));
+ for (Shard shard : openShards) {
+ String shardId = shard.shardId();
+ if (knownSplits.containsKey(shardId)) {
+ continue;
+ }
+ String firstShardIdBeforeTimestamp =
+ findFirstShardIdBeforeTimestamp(shardId, shardIdToShardMap);
+ putAllAncestorShardsTillFirstShardId(
+ shardId, firstShardIdBeforeTimestamp, shardIdToShardMap);
+ }
+ }
+
+ private void putAllAncestorShardsTillFirstShardId(
+ String shardId,
+ String firstShardIdBeforeTimestamp,
+ Map<String, Shard> shardIdToShardMap) {
+ String currentShardId = shardId;
+ while (currentShardId != null && shardIdToShardMap.containsKey(currentShardId)) {
+ Shard currentShard = shardIdToShardMap.get(currentShardId);
+ if (Objects.equals(currentShardId, firstShardIdBeforeTimestamp)) {
+ DynamoDbStreamsShardSplit newSplit = mapToSplit(currentShard, LATEST);
Review Comment:
yes correct we need to update this in docs
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/util/ShardUtils.java:
##########
@@ -78,4 +78,14 @@ public static boolean isShardOlderThanInconsistencyDetectionRetentionPeriod(Stri
.plus(
DDB_STREAMS_MAX_RETENTION_PERIOD_FOR_RESOLVING_INCONSISTENCIES));
}
+
+ /**
+ * Returns true if shard ue if the shard was created before the given timestamp.
Review Comment:
will fix
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36523][Connectors/DynamoDB][BugFix] Fixing LATEST behaviour for DDB Connector [flink-connector-aws]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 merged PR #177:
URL: https://github.com/apache/flink-connector-aws/pull/177
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36523][Connectors/DynamoDB][BugFix] Fixing LATEST behaviour for DDB Connector [flink-connector-aws]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #177:
URL: https://github.com/apache/flink-connector-aws/pull/177#discussion_r1799315333
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -89,13 +91,77 @@ public SplitTracker(
* @param shardsToAdd collection of splits to add to tracking
*/
public void addSplits(Collection<Shard> shardsToAdd) {
- Set<String> discoveredShardIds =
- shardsToAdd.stream().map(Shard::shardId).collect(Collectors.toSet());
+ if (TRIM_HORIZON.equals(initialPosition)) {
+ addSplitsForTrimHorizon(shardsToAdd);
+ return;
+ }
+ addSplitsForLatest(shardsToAdd);
+ }
+
+ /**
+ * If there is an ancestor of an open shard in knownSplits, we put all of its ancestors after
+ * the tracked ancestors as TRIM_HORIZON otherwise, we ignore all the ancestors and track only
+ * the open shard with LATEST. There might be a case when restoring from expired snapshots that
+ * there is an ancestor in knownSplits, but we wont know that it is an ancestor, so in that case
+ * also, all the ancestors will not be tracked and rather only the current open shard will be
+ * tracked with LATEST.
Review Comment:
We should align that the expected behavior, when using `LATEST`, should be:
1. When starting without snapshot,
- All shards read from `LATEST`. If we want, we can focus on just the open shards.
2. When reading with snapshot,
- All shards from time of start of application are read from `TRIM_HORIZON`,
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -89,13 +91,77 @@ public SplitTracker(
* @param shardsToAdd collection of splits to add to tracking
*/
public void addSplits(Collection<Shard> shardsToAdd) {
- Set<String> discoveredShardIds =
- shardsToAdd.stream().map(Shard::shardId).collect(Collectors.toSet());
+ if (TRIM_HORIZON.equals(initialPosition)) {
+ addSplitsForTrimHorizon(shardsToAdd);
+ return;
+ }
+ addSplitsForLatest(shardsToAdd);
+ }
+
+ /**
+ * If there is an ancestor of an open shard in knownSplits, we put all of its ancestors after
+ * the tracked ancestors as TRIM_HORIZON otherwise, we ignore all the ancestors and track only
+ * the open shard with LATEST. There might be a case when restoring from expired snapshots that
+ * there is an ancestor in knownSplits, but we wont know that it is an ancestor, so in that case
+ * also, all the ancestors will not be tracked and rather only the current open shard will be
+ * tracked with LATEST.
Review Comment:
We should align that the expected behavior, when using `LATEST`, should be:
1. When starting without snapshot:
- All shards read from `LATEST`. If we want, we can focus on just the open shards.
2. When reading with snapshot:
- All shards from time of start of application are read from `TRIM_HORIZON`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36523][Connectors/DynamoDB][BugFix] Fixing LATEST behaviour for DDB Connector [flink-connector-aws]
Posted by "gguptp (via GitHub)" <gi...@apache.org>.
gguptp commented on code in PR #177:
URL: https://github.com/apache/flink-connector-aws/pull/177#discussion_r1811947253
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -89,13 +95,69 @@ public SplitTracker(
* @param shardsToAdd collection of splits to add to tracking
*/
public void addSplits(Collection<Shard> shardsToAdd) {
- Set<String> discoveredShardIds =
- shardsToAdd.stream().map(Shard::shardId).collect(Collectors.toSet());
+ if (TRIM_HORIZON.equals(initialPosition)) {
+ addSplitsForTrimHorizon(shardsToAdd);
+ return;
+ }
+ addSplitsForLatest(shardsToAdd);
+ }
+
+ private void addSplitsForLatest(Collection<Shard> shardsToAdd) {
+ List<Shard> openShards =
+ shardsToAdd.stream()
+ .filter(shard -> shard.sequenceNumberRange().endingSequenceNumber() == null)
+ .collect(Collectors.toList());
+ Map<String, Shard> shardIdToShardMap =
+ shardsToAdd.stream().collect(Collectors.toMap(Shard::shardId, shard -> shard));
+ for (Shard shard : openShards) {
+ String shardId = shard.shardId();
+ if (knownSplits.containsKey(shardId)) {
+ continue;
+ }
+ String firstShardIdBeforeTimestamp =
+ findFirstShardIdBeforeTimestamp(shardId, shardIdToShardMap);
+ putAllAncestorShardsTillFirstShardId(
+ shardId, firstShardIdBeforeTimestamp, shardIdToShardMap);
+ }
+ }
+
+ private void putAllAncestorShardsTillFirstShardId(
+ String shardId,
+ String firstShardIdBeforeTimestamp,
+ Map<String, Shard> shardIdToShardMap) {
+ String currentShardId = shardId;
+ while (currentShardId != null && shardIdToShardMap.containsKey(currentShardId)) {
+ Shard currentShard = shardIdToShardMap.get(currentShardId);
+ if (Objects.equals(currentShardId, firstShardIdBeforeTimestamp)) {
+ DynamoDbStreamsShardSplit newSplit = mapToSplit(currentShard, LATEST);
+ knownSplits.putIfAbsent(currentShardId, newSplit);
+ break;
+ } else {
+ DynamoDbStreamsShardSplit newSplit = mapToSplit(currentShard, TRIM_HORIZON);
+ knownSplits.putIfAbsent(currentShardId, newSplit);
+ }
+ currentShardId = currentShard.parentShardId();
+ }
+ }
+
+ private String findFirstShardIdBeforeTimestamp(
+ String shardIdToStartWith, Map<String, Shard> shardIdToShardMap) {
+ String shardId = shardIdToStartWith;
+ while (shardId != null && shardIdToShardMap.containsKey(shardId)) {
+ Shard shard = shardIdToShardMap.get(shardId);
Review Comment:
makes sense let me do this
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36523][Connectors/DynamoDB][BugFix] Fixing LATEST behaviour for DDB Connector [flink-connector-aws]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #177:
URL: https://github.com/apache/flink-connector-aws/pull/177#discussion_r1799008421
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -89,13 +91,77 @@ public SplitTracker(
* @param shardsToAdd collection of splits to add to tracking
*/
public void addSplits(Collection<Shard> shardsToAdd) {
- Set<String> discoveredShardIds =
- shardsToAdd.stream().map(Shard::shardId).collect(Collectors.toSet());
+ if (TRIM_HORIZON.equals(initialPosition)) {
+ addSplitsForTrimHorizon(shardsToAdd);
+ return;
+ }
+ addSplitsForLatest(shardsToAdd);
+ }
+
+ /**
+ * If there is an ancestor of an open shard in knownSplits, we put all of its ancestors after
+ * the tracked ancestors as TRIM_HORIZON otherwise, we ignore all the ancestors and track only
+ * the open shard with LATEST. There might be a case when restoring from expired snapshots that
+ * there is an ancestor in knownSplits, but we wont know that it is an ancestor, so in that case
+ * also, all the ancestors will not be tracked and rather only the current open shard will be
+ * tracked with LATEST.
Review Comment:
I don't think this is correct. When restoring from snapshot, we should read all records since the restored snapshot (`TRIM_HORIZON`), and not `LATEST`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org