Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

add functionality to retry an InvalidArgumentException #1270

Merged
merged 2 commits into from
Mar 8, 2024
Merged

add functionality to retry an InvalidArgumentException #1270

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8f81e3f
add functionality to retry an InvalidArgumentException with a new ite…
vincentvilo-aws Mar 7, 2024
91cd65f
include DEFAULT_MAX_RECORDS value in IllegalArgumentException messages
vincentvilo-aws Mar 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions ...lang/src/main/java/software/amazon/kinesis/coordinator/KinesisClientLibConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -977,6 +977,10 @@ public KinesisClientLibConfiguration withShardSyncIntervalMillis(long shardSyncI
*/
public KinesisClientLibConfiguration withMaxRecords(int maxRecords) {
checkIsValuePositive("MaxRecords", (long) maxRecords);
if (maxRecords > DEFAULT_MAX_RECORDS) {
throw new IllegalArgumentException(
"maxRecords must be less than or equal to " + DEFAULT_MAX_RECORDS + " but current value is " + maxRecords);
}
this.maxRecords = maxRecords;
return this;
}
Expand Down
12 changes: 11 additions & 1 deletion ...kinesis-client/src/main/java/software/amazon/kinesis/retrieval/polling/PollingConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@ public class PollingConfig implements RetrievalSpecificConfig {

public static final Duration DEFAULT_REQUEST_TIMEOUT = Duration.ofSeconds(30);

public static final int DEFAULT_MAX_RECORDS = 10000;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If this is going to be public and static, we can share this value in the multilang as well so that there is one central source of truth.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

as a follow up, it may be worth checking to see how many of these types of values we have "copy pasted" in multiple locations and see if its worth centralizing them.


/**
* Configurable functional interface to override the existing DataFetcher.
*/
Expand Down Expand Up @@ -71,7 +73,7 @@ public PollingConfig(KinesisAsyncClient kinesisClient) {
* Default value: 10000
* </p>
*/
private int maxRecords = 10000;
private int maxRecords = DEFAULT_MAX_RECORDS;

/**
* @param streamName Name of Kinesis stream.
Expand Down Expand Up @@ -129,6 +131,14 @@ public void setIdleTimeBetweenReadsInMillis(long idleTimeBetweenReadsInMillis) {
this.idleTimeBetweenReadsInMillis = idleTimeBetweenReadsInMillis;
}

public void maxRecords(int maxRecords) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: setMaxRecords(...);

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The class has an annotation @Accessors(fluent = true), so the getters and setters do not have the "get" and "set" prefix (ref).

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@vincentvilo-aws this broke backwards compatibility. The method needed to return PollingConfig. See #1274 and #1275

Copy link
Contributor Author

@vincentvilo-aws vincentvilo-aws Mar 15, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for catching this. The chain boolean by default is set to true when using the @Accessors annotation. I'll approve #1275 once the checks pass.

if (maxRecords > DEFAULT_MAX_RECORDS) {
throw new IllegalArgumentException(
"maxRecords must be less than or equal to " + DEFAULT_MAX_RECORDS + " but current value is " + maxRecords());
}
this.maxRecords = maxRecords;
}

/**
* The maximum time to wait for a future request from Kinesis to complete
*/
Expand Down
5 changes: 4 additions & 1 deletion ...ent/src/main/java/software/amazon/kinesis/retrieval/polling/PrefetchRecordsPublisher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import software.amazon.awssdk.services.cloudwatch.model.StandardUnit;
import software.amazon.awssdk.services.kinesis.model.ExpiredIteratorException;
import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
import software.amazon.awssdk.services.kinesis.model.InvalidArgumentException;
import software.amazon.awssdk.services.kinesis.model.ProvisionedThroughputExceededException;
import software.amazon.kinesis.annotations.KinesisClientInternalApi;
import software.amazon.kinesis.common.InitialPositionInStreamExtended;
Expand Down Expand Up @@ -102,7 +103,6 @@ public class PrefetchRecordsPublisher implements RecordsPublisher {
private final PublisherSession publisherSession;
private final ReentrantReadWriteLock resetLock = new ReentrantReadWriteLock();
private boolean wasReset = false;

private Instant lastEventDeliveryTime = Instant.EPOCH;
private final RequestDetails lastSuccessfulRequestDetails = new RequestDetails();

Expand Down Expand Up @@ -511,6 +511,9 @@ private void makeRetrievalAttempt() {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.info("{} : Thread was interrupted, indicating shutdown was called on the cache.", streamAndShardId);
} catch (InvalidArgumentException e) {
vincentvilo-aws marked this conversation as resolved.
Show resolved Hide resolved
log.info("{} : records threw InvalidArgumentException - iterator will be refreshed before retrying", streamAndShardId, e);
publisherSession.dataFetcher().restartIterator();
} catch (ExpiredIteratorException e) {
log.info("{} : records threw ExpiredIteratorException - restarting"
+ " after greatest seqNum passed to customer", streamAndShardId, e);
Expand Down
5 changes: 5 additions & 0 deletions ...sis-client/src/test/java/software/amazon/kinesis/retrieval/polling/PollingConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,4 +44,9 @@ public void testInvalidStateMultiWithStreamName() {
config.validateState(true);
}

@Test(expected = IllegalArgumentException.class)
public void testInvalidRecordLimit() {
config.maxRecords(PollingConfig.DEFAULT_MAX_RECORDS + 1);
}

}
13 changes: 13 additions & 0 deletions ...src/test/java/software/amazon/kinesis/retrieval/polling/PrefetchRecordsPublisherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@
import software.amazon.awssdk.services.kinesis.model.ChildShard;
import software.amazon.awssdk.services.kinesis.model.ExpiredIteratorException;
import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
import software.amazon.awssdk.services.kinesis.model.InvalidArgumentException;
import software.amazon.awssdk.services.kinesis.model.Record;
import software.amazon.kinesis.common.InitialPositionInStreamExtended;
import software.amazon.kinesis.leases.ShardObjectHelper;
Expand Down Expand Up @@ -434,6 +435,18 @@ public void testRetryableRetrievalExceptionContinues() {
assertEquals(records.processRecordsInput().millisBehindLatest(), response.millisBehindLatest());
}

@Test
public void testInvalidArgumentExceptionIsRetried() {
when(getRecordsRetrievalStrategy.getRecords(MAX_RECORDS_PER_CALL))
.thenThrow(InvalidArgumentException.builder().build())
.thenReturn(getRecordsResponse);

getRecordsCache.start(sequenceNumber, initialPosition);
blockUntilConditionSatisfied(() -> getRecordsCache.getPublisherSession().prefetchRecordsQueue().size() == MAX_SIZE, 300);

verify(dataFetcher, times(1)).restartIterator();
}

@Test(timeout = 10000L)
public void testNoDeadlockOnFullQueue() {
//
Expand Down
Loading