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

CR Feedback

Browse source
This commit is contained in:
Cory Bradshaw 2019-11-29 14:21:51 -08:00
parent bd281853a0
commit c2de301444
3 changed files with 12 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java
View file

@ -361,6 +361,7 @@ public class KinesisClientLibConfiguration {
* {@link RecordProcessorCheckpointer#checkpoint(String)}
* @param regionName The region name for the service
* @param shutdownGraceMillis The number of milliseconds before graceful shutdown terminates forcefully
* @param billingMode The DDB Billing mode to set for lease table creation.
*/
// CHECKSTYLE:IGNORE HiddenFieldCheck FOR NEXT 26 LINES
// CHECKSTYLE:IGNORE ParameterNumber FOR NEXT 26 LINES
@ -431,6 +432,7 @@ public class KinesisClientLibConfiguration {
* with a call to Amazon Kinesis before checkpointing for calls to
* {@link RecordProcessorCheckpointer#checkpoint(String)}
* @param regionName The region name for the service
* @param billingMode The DDB Billing mode to set for lease table creation.
*/
// CHECKSTYLE:IGNORE HiddenFieldCheck FOR NEXT 26 LINES
// CHECKSTYLE:IGNORE ParameterNumber FOR NEXT 26 LINES
@ -507,7 +509,7 @@ public class KinesisClientLibConfiguration {
this.shardSyncStrategyType = DEFAULT_SHARD_SYNC_STRATEGY_TYPE;
this.shardPrioritization = DEFAULT_SHARD_PRIORITIZATION;
this.recordsFetcherFactory = new SimpleRecordsFetcherFactory();
this.billingMode=billingMode;
this.billingMode = billingMode;
}
/**
@ -1162,6 +1164,11 @@ public class KinesisClientLibConfiguration {
return this;
}
/**
* The DDB Billing mode to set for lease table creation.
* @param billingMode - Either PAY_PER_REQUEST, or PROVISIONED; Defaults to PROVISIONED
* @return
*/
public KinesisClientLibConfiguration withBillingMode(BillingMode billingMode){
this.billingMode = billingMode == null ? DEFAULT_DDB_BILLING_MODE : billingMode;
return this;

2
src/main/java/com/amazonaws/services/kinesis/leases/impl/LeaseManager.java
View file

@ -70,6 +70,7 @@ public class LeaseManager<T extends Lease> implements ILeaseManager<T> {
* @param table leases table
* @param dynamoDBClient DynamoDB client to use
* @param serializer LeaseSerializer to use to convert to/from DynamoDB objects.
* @param billingMode The DDB Billing mode to set for lease table creation.
*/
public LeaseManager(String table, AmazonDynamoDB dynamoDBClient, ILeaseSerializer<T> serializer, BillingMode billingMode) {
this(table, dynamoDBClient, serializer, false, billingMode);
@ -85,6 +86,7 @@ public class LeaseManager<T extends Lease> implements ILeaseManager<T> {
* @param dynamoDBClient DynamoDB client to use
* @param serializer lease serializer to use
* @param consistentReads true if we want consistent reads for testing purposes.
* @param billingMode The DDB Billing mode to set for lease table creation.
*/
public LeaseManager(String table, AmazonDynamoDB dynamoDBClient, ILeaseSerializer<T> serializer, boolean consistentReads, BillingMode billingMode) {
verifyNotNull(table, "Table name cannot be null");

3
src/test/java/com/amazonaws/services/kinesis/leases/impl/LeaseManagerIntegrationTest.java
View file

@ -18,6 +18,7 @@ import java.util.Collection;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import com.amazonaws.services.dynamodbv2.model.BillingMode;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import junit.framework.Assert;
@ -251,7 +252,7 @@ public class LeaseManagerIntegrationTest extends LeaseIntegrationTest {
@Test
public void testWaitUntilLeaseTableExistsPayPerRequest() throws LeasingException {
KinesisClientLeaseManager manager = new KinesisClientLeaseManager("nagl_ShardProgress_PayPerRequest", ddbClient, true,
KinesisClientLibConfiguration.DEFAULT_DDB_BILLING_MODE) {
BillingMode.PAY_PER_REQUEST) {
@Override
long sleep(long timeToSleepMillis) {