hlteoh37 commented on code in PR #190:
URL:
https://github.com/apache/flink-connector-aws/pull/190#discussion_r1973547168
##########
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:
This seems like it is introducing unnecessary complexity. Can we keep it
simple first, and just keep the `Deque`? Alternatively, if we want to do this,
let's put it behind a feature flag and add debugging logs.
I prefer just keeping it simple first!
--
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
