jfarr commented on a change in pull request #11090: [BEAM-9470]
:sdks:java:io:kinesis:test is flaky
URL: https://github.com/apache/beam/pull/11090#discussion_r393216743
##########
File path:
sdks/java/io/kinesis/src/main/java/org/apache/beam/sdk/io/kinesis/ShardReadersPool.java
##########
@@ -126,9 +126,21 @@ void startReadingShards(Iterable<ShardRecordsIterator>
shardRecordsIterators) {
private void readLoop(ShardRecordsIterator shardRecordsIterator,
RateLimitPolicy rateLimiter) {
while (poolOpened.get()) {
try {
- List<KinesisRecord> kinesisRecords;
try {
- kinesisRecords = shardRecordsIterator.readNextBatch();
+ List<KinesisRecord> kinesisRecords =
shardRecordsIterator.readNextBatch();
+ try {
+ for (KinesisRecord kinesisRecord : kinesisRecords) {
+ recordsQueue.put(kinesisRecord);
+
numberOfRecordsInAQueueByShard.get(kinesisRecord.getShardId()).incrementAndGet();
+ }
+ } finally {
+ // One of the paths into this finally block is recordsQueue.put()
throwing
+ // InterruptedException so we should check the thread's
interrupted status before
+ // calling onSuccess().
+ if (!Thread.currentThread().isInterrupted()) {
+ rateLimiter.onSuccess(kinesisRecords);
Review comment:
Yes, in order to rate limit correctly onSuccess() needs to be called after
every successful getRecords() call even if it returned no records. Otherwise we
would be right back to exceeding our API rate limit when the shard has no more
records to process.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
