[
https://issues.apache.org/jira/browse/BEAM-9470?focusedWorklogId=404160&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-404160
]
ASF GitHub Bot logged work on BEAM-9470:
----------------------------------------
Author: ASF GitHub Bot
Created on: 16/Mar/20 18:06
Start Date: 16/Mar/20 18:06
Worklog Time Spent: 10m
Work Description: jfarr commented on 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]
Issue Time Tracking
-------------------
Worklog Id: (was: 404160)
Time Spent: 8h 10m (was: 8h)
> :sdks:java:io:kinesis:test is flaky
> -----------------------------------
>
> Key: BEAM-9470
> URL: https://issues.apache.org/jira/browse/BEAM-9470
> Project: Beam
> Issue Type: Test
> Components: io-java-kinesis
> Reporter: Etienne Chauchot
> Assignee: Jonothan Farr
> Priority: Major
> Time Spent: 8h 10m
> Remaining Estimate: 0h
>
> [https://scans.gradle.com/s/b4jmmu72ku5jc/console-log?task=:sdks:java:io:kinesis:test]
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
