gguptp commented on code in PR #190:
URL:
https://github.com/apache/flink-connector-aws/pull/190#discussion_r1973555792
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/reader/PollingDynamoDbStreamsShardSplitReader.java:
##########
@@ -57,48 +59,101 @@ public class PollingDynamoDbStreamsShardSplitReader
new DynamoDbStreamRecordsWithSplitIds(Collections.emptyIterator(),
null, false);
private final StreamProxy dynamodbStreams;
+ private final Duration getRecordsIdlePollingTimeBetweenNonEmptyPolls;
+ private final Duration getRecordsIdlePollingTimeBetweenEmptyPolls;
- private final Deque<DynamoDbStreamsShardSplitState> assignedSplits = new
ArrayDeque<>();
+ private final PriorityQueue<SplitContext> assignedSplits;
private final Map<String, DynamoDbStreamsShardMetrics> shardMetricGroupMap;
private final Set<String> pausedSplitIds = new HashSet<>();
public PollingDynamoDbStreamsShardSplitReader(
StreamProxy dynamodbStreamsProxy,
+ Duration getRecordsIdlePollingTimeBetweenNonEmptyPolls,
+ Duration getRecordsIdlePollingTimeBetweenEmptyPolls,
Map<String, DynamoDbStreamsShardMetrics> shardMetricGroupMap) {
this.dynamodbStreams = dynamodbStreamsProxy;
+ this.getRecordsIdlePollingTimeBetweenNonEmptyPolls =
+ getRecordsIdlePollingTimeBetweenNonEmptyPolls;
+ this.getRecordsIdlePollingTimeBetweenEmptyPolls =
+ getRecordsIdlePollingTimeBetweenEmptyPolls;
this.shardMetricGroupMap = shardMetricGroupMap;
+ this.assignedSplits =
+ new PriorityQueue<>(
+ (a, b) -> {
+ // First, handle paused splits
+ boolean aIsPaused =
pausedSplitIds.contains(a.splitState.getSplitId());
+ boolean bIsPaused =
pausedSplitIds.contains(b.splitState.getSplitId());
+ if (aIsPaused && !bIsPaused) {
+ return 1;
+ }
+ if (!aIsPaused && bIsPaused) {
+ return -1;
+ }
+ if (aIsPaused && bIsPaused) {
+ return 0;
+ }
+
+ // Get next eligible time for both splits
+ long aNextEligibleTime = getNextEligibleTime(a);
+ long bNextEligibleTime = getNextEligibleTime(b);
+
+ return Long.compare(aNextEligibleTime,
bNextEligibleTime);
+ });
+ }
Review Comment:
yeah, makes sense we can do that
--
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-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
