gguptp commented on code in PR #151:
URL:
https://github.com/apache/flink-connector-aws/pull/151#discussion_r1741930430
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/DynamoDbStreamsSourceEnumerator.java:
##########
@@ -87,25 +92,29 @@ public DynamoDbStreamsSourceEnumerator(
this.streamProxy = streamProxy;
this.shardAssigner = shardAssigner;
this.shardAssignerContext = new ShardAssignerContext(splitAssignment,
context);
+ InitialPosition initialPosition =
sourceConfig.get(STREAM_INITIAL_POSITION);
+ this.hasInitialState = state != null;
if (state == null) {
- this.lastSeenShardId = null;
- this.unassignedSplits = new HashSet<>();
+ this.splitTracker = new SplitTracker(streamArn, initialPosition);
} else {
- this.lastSeenShardId = state.getLastSeenShardId();
- this.unassignedSplits = state.getUnassignedSplits();
+ this.splitTracker =
+ new SplitTracker(state.getKnownSplits(), streamArn,
initialPosition);
}
}
@Override
public void start() {
- if (lastSeenShardId == null) {
- context.callAsync(this::initialDiscoverSplits, this::assignSplits);
+ // if there is already a saved state, we can just start with that
state instead of waiting
+ // for shard discovery to complete
+ if (!hasInitialState) {
+ context.callAsync(this::discoverSplits,
this::processDiscoveredSplits);
Review Comment:
there might be cases where parent is finished, the job died and children are
unassigned. Its an edge case though, so thought its better to handle here
--
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]
