Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/2925#discussion_r95134466
--- Diff:
flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java
---
@@ -154,42 +166,115 @@ public void run() {
}
}
- // set the nextShardItr so we can
continue iterating in the next while loop
- nextShardItr =
getRecordsResult.getNextShardIterator();
+ // set the startShardItr so we can
continue iterating in the next while loop
+ startShardItr =
getRecordsResult.getNextShardIterator();
} else {
// the last record was non-aggregated,
so we can simply start from the next record
- nextShardItr =
kinesis.getShardIterator(subscribedShard,
ShardIteratorType.AFTER_SEQUENCE_NUMBER.toString(),
lastSequenceNum.getSequenceNumber());
+ startShardItr =
kinesis.getShardIterator(subscribedShard,
ShardIteratorType.AFTER_SEQUENCE_NUMBER.toString(),
lastSequenceNum.getSequenceNumber());
}
}
- while(isRunning()) {
- if (nextShardItr == null) {
-
fetcherRef.updateState(subscribedShardStateIndex,
SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get());
-
- // we can close this consumer thread
once we've reached the end of the subscribed shard
- break;
- } else {
- if (fetchIntervalMillis != 0) {
-
Thread.sleep(fetchIntervalMillis);
- }
+ ArrayBlockingQueue<UserRecord> queue = new
ArrayBlockingQueue<>(maxNumberOfRecordsPerFetch);
+ ShardConsumerFetcher shardConsumerFetcher;
- GetRecordsResult getRecordsResult =
getRecords(nextShardItr, maxNumberOfRecordsPerFetch);
+ if (fetchIntervalMillis > 0L) {
+ shardConsumerFetcher = new
ShardConsumerFetcher(this, startShardItr, queue, false);
+ timer.scheduleAtFixedRate(shardConsumerFetcher,
0L, fetchIntervalMillis);
+ } else {
+ // if fetchIntervalMillis is 0, make the task
run forever and schedule it once only.
+ shardConsumerFetcher = new
ShardConsumerFetcher(this, startShardItr, queue, true);
+ timer.schedule(shardConsumerFetcher, 0L);
+ }
- // each of the Kinesis records may be
aggregated, so we must deaggregate them before proceeding
- List<UserRecord> fetchedRecords =
deaggregateRecords(
- getRecordsResult.getRecords(),
-
subscribedShard.getShard().getHashKeyRange().getStartingHashKey(),
-
subscribedShard.getShard().getHashKeyRange().getEndingHashKey());
+ while(isRunning()) {
+ UserRecord record = queue.poll();
+ if (record != null) {
+
deserializeRecordForCollectionAndUpdateState(record);
+ } else {
+ if (shardConsumerFetcher.nextShardItr
== null) {
+
fetcherRef.updateState(subscribedShardStateIndex,
SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get());
- for (UserRecord record :
fetchedRecords) {
-
deserializeRecordForCollectionAndUpdateState(record);
+ // we can close this consumer
thread once we've reached the end of the subscribed shard
+ break;
}
+ }
- nextShardItr =
getRecordsResult.getNextShardIterator();
+ Throwable throwable = this.error.get();
+ if (throwable != null) {
+ throw throwable;
}
}
} catch (Throwable t) {
fetcherRef.stopWithError(t);
+ } finally {
+ timer.cancel();
--- End diff --
Calling `cancel` on a Timer doesn't handle any in-progress tasks, it only
discards the current _scheduled_ tasks. So for the case where
`fetchIntervalMillis` is 0, the forever running fetcher task will not be
terminated.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
