luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #53 from ashwing/ltr_ddb_scan_opt

Browse source 
Limiting max page for ddb scan in isEmptyLeaseTable check to be 1
This commit is contained in:
ashwing 2020-06-11 18:42:48 -07:00 committed by GitHub
commit 1aeb8ed9a1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 19 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/dynamodb/DynamoDBLeaseRefresher.java
View file

@ -292,7 +292,7 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
@Override @Override
public boolean isLeaseTableEmpty() public boolean isLeaseTableEmpty()
throws DependencyException, InvalidStateException, ProvisionedThroughputException { throws DependencyException, InvalidStateException, ProvisionedThroughputException {
return list(1, null).isEmpty(); return list(1, 1, null).isEmpty();
} }
/** /**
@ -305,7 +305,23 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
* @throws DependencyException if DynamoDB scan fail in an unexpected way * @throws DependencyException if DynamoDB scan fail in an unexpected way
* @throws ProvisionedThroughputException if DynamoDB scan fail due to exceeded capacity * @throws ProvisionedThroughputException if DynamoDB scan fail due to exceeded capacity
*/ */
List<Lease> list(Integer limit, StreamIdentifier streamIdentifier) throws DependencyException, InvalidStateException, List<Lease> list(Integer limit, StreamIdentifier streamIdentifier)
throws DependencyException, InvalidStateException, ProvisionedThroughputException {
return list(limit, Integer.MAX_VALUE, streamIdentifier);
}
/**
* List with the given page size. Package access for integration testing.
*
* @param limit number of items to consider at a time - used by integration tests to force paging.
* @param maxPages mad paginated scan calls
* @param streamIdentifier streamIdentifier for multi-stream mode. Can be null.
* @return list of leases
* @throws InvalidStateException if table does not exist
* @throws DependencyException if DynamoDB scan fail in an unexpected way
* @throws ProvisionedThroughputException if DynamoDB scan fail due to exceeded capacity
*/
private List<Lease> list(Integer limit, Integer maxPages, StreamIdentifier streamIdentifier) throws DependencyException, InvalidStateException,
ProvisionedThroughputException { ProvisionedThroughputException {
log.debug("Listing leases from table {}", table); log.debug("Listing leases from table {}", table);
@ -341,7 +357,7 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
} }
Map<String, AttributeValue> lastEvaluatedKey = scanResult.lastEvaluatedKey(); Map<String, AttributeValue> lastEvaluatedKey = scanResult.lastEvaluatedKey();
if (CollectionUtils.isNullOrEmpty(lastEvaluatedKey)) { if (CollectionUtils.isNullOrEmpty(lastEvaluatedKey) || --maxPages <= 0) {
// Signify that we're done. // Signify that we're done.
scanResult = null; scanResult = null;
log.debug("lastEvaluatedKey was null - scan finished."); log.debug("lastEvaluatedKey was null - scan finished.");