@Test public void testCreateLeaseTableBillingMode() throws Exception { leaseRefresher = new DynamoDBLeaseRefresher(TABLE_NAME, dynamoDbClient, leaseSerializer, CONSISTENT_READS, tableCreatorCallback, LeaseManagementConfig.DEFAULT_REQUEST_TIMEOUT, BillingMode.PAY_PER_REQUEST); TimeoutException te = setRuleForDependencyTimeout(); when(dynamoDbClient.describeTable(any(DescribeTableRequest.class))).thenReturn(mockDescribeTableFuture); when(mockDescribeTableFuture.get(anyLong(), any())) .thenThrow(ResourceNotFoundException.builder().message("Table doesn't exist").build()); when(dynamoDbClient.createTable(any(CreateTableRequest.class))).thenReturn(mockCreateTableFuture); when(mockCreateTableFuture.get(anyLong(), any())).thenThrow(te); verifyCancel(mockCreateTableFuture, () -> leaseRefresher.createLeaseTableIfNotExists(10L, 10L)); }
@Override public boolean createLeaseTableIfNotExists(@NonNull final Long readCapacity, @NonNull final Long writeCapacity) throws ProvisionedThroughputException, DependencyException { try { if (tableStatus() != null) { return newTableCreated; } } catch (DependencyException de) { log.error("Failed to get table status for {}", table, de); } ProvisionedThroughput throughput = ProvisionedThroughput.builder().readCapacityUnits(readCapacity) .writeCapacityUnits(writeCapacity).build(); final CreateTableRequest request; if(BillingMode.PAY_PER_REQUEST.equals(billingMode)){ request = CreateTableRequest.builder().tableName(table).keySchema(serializer.getKeySchema()) .attributeDefinitions(serializer.getAttributeDefinitions()) .billingMode(billingMode).build(); }else{ request = CreateTableRequest.builder().tableName(table).keySchema(serializer.getKeySchema()) .attributeDefinitions(serializer.getAttributeDefinitions()).provisionedThroughput(throughput) .build(); } final AWSExceptionManager exceptionManager = createExceptionManager(); exceptionManager.add(ResourceInUseException.class, t -> t); exceptionManager.add(LimitExceededException.class, t -> t); try { try { FutureUtils.resolveOrCancelFuture(dynamoDBClient.createTable(request), dynamoDbRequestTimeout); newTableCreated = true; } catch (ExecutionException e) { throw exceptionManager.apply(e.getCause()); } catch (InterruptedException e) { throw new DependencyException(e); } } catch (ResourceInUseException e) { log.info("Table {} already exists.", table); return newTableCreated; } catch (LimitExceededException e) { throw new ProvisionedThroughputException("Capacity exceeded when creating table " + table, e); } catch (DynamoDbException | TimeoutException e) { throw new DependencyException(e); } return newTableCreated; }
DynamoDBLeaseRefresher implements LeaseRefresher { @Override public boolean createLeaseTableIfNotExists(@NonNull final Long readCapacity, @NonNull final Long writeCapacity) throws ProvisionedThroughputException, DependencyException { try { if (tableStatus() != null) { return newTableCreated; } } catch (DependencyException de) { log.error("Failed to get table status for {}", table, de); } ProvisionedThroughput throughput = ProvisionedThroughput.builder().readCapacityUnits(readCapacity) .writeCapacityUnits(writeCapacity).build(); final CreateTableRequest request; if(BillingMode.PAY_PER_REQUEST.equals(billingMode)){ request = CreateTableRequest.builder().tableName(table).keySchema(serializer.getKeySchema()) .attributeDefinitions(serializer.getAttributeDefinitions()) .billingMode(billingMode).build(); }else{ request = CreateTableRequest.builder().tableName(table).keySchema(serializer.getKeySchema()) .attributeDefinitions(serializer.getAttributeDefinitions()).provisionedThroughput(throughput) .build(); } final AWSExceptionManager exceptionManager = createExceptionManager(); exceptionManager.add(ResourceInUseException.class, t -> t); exceptionManager.add(LimitExceededException.class, t -> t); try { try { FutureUtils.resolveOrCancelFuture(dynamoDBClient.createTable(request), dynamoDbRequestTimeout); newTableCreated = true; } catch (ExecutionException e) { throw exceptionManager.apply(e.getCause()); } catch (InterruptedException e) { throw new DependencyException(e); } } catch (ResourceInUseException e) { log.info("Table {} already exists.", table); return newTableCreated; } catch (LimitExceededException e) { throw new ProvisionedThroughputException("Capacity exceeded when creating table " + table, e); } catch (DynamoDbException | TimeoutException e) { throw new DependencyException(e); } return newTableCreated; } }
DynamoDBLeaseRefresher implements LeaseRefresher { @Override public boolean createLeaseTableIfNotExists(@NonNull final Long readCapacity, @NonNull final Long writeCapacity) throws ProvisionedThroughputException, DependencyException { try { if (tableStatus() != null) { return newTableCreated; } } catch (DependencyException de) { log.error("Failed to get table status for {}", table, de); } ProvisionedThroughput throughput = ProvisionedThroughput.builder().readCapacityUnits(readCapacity) .writeCapacityUnits(writeCapacity).build(); final CreateTableRequest request; if(BillingMode.PAY_PER_REQUEST.equals(billingMode)){ request = CreateTableRequest.builder().tableName(table).keySchema(serializer.getKeySchema()) .attributeDefinitions(serializer.getAttributeDefinitions()) .billingMode(billingMode).build(); }else{ request = CreateTableRequest.builder().tableName(table).keySchema(serializer.getKeySchema()) .attributeDefinitions(serializer.getAttributeDefinitions()).provisionedThroughput(throughput) .build(); } final AWSExceptionManager exceptionManager = createExceptionManager(); exceptionManager.add(ResourceInUseException.class, t -> t); exceptionManager.add(LimitExceededException.class, t -> t); try { try { FutureUtils.resolveOrCancelFuture(dynamoDBClient.createTable(request), dynamoDbRequestTimeout); newTableCreated = true; } catch (ExecutionException e) { throw exceptionManager.apply(e.getCause()); } catch (InterruptedException e) { throw new DependencyException(e); } } catch (ResourceInUseException e) { log.info("Table {} already exists.", table); return newTableCreated; } catch (LimitExceededException e) { throw new ProvisionedThroughputException("Capacity exceeded when creating table " + table, e); } catch (DynamoDbException | TimeoutException e) { throw new DependencyException(e); } return newTableCreated; } @Deprecated  DynamoDBLeaseRefresher(final String table, final DynamoDbAsyncClient dynamoDBClient,
            final LeaseSerializer serializer, final boolean consistentReads); @Deprecated  DynamoDBLeaseRefresher(final String table, final DynamoDbAsyncClient dynamoDBClient,
                                  final LeaseSerializer serializer, final boolean consistentReads,
                                  @NonNull final TableCreatorCallback tableCreatorCallback); @Deprecated  DynamoDBLeaseRefresher(final String table, final DynamoDbAsyncClient dynamoDBClient,
                                  final LeaseSerializer serializer, final boolean consistentReads,
                                  @NonNull final TableCreatorCallback tableCreatorCallback, Duration dynamoDbRequestTimeout);  DynamoDBLeaseRefresher(final String table, final DynamoDbAsyncClient dynamoDBClient,
                                  final LeaseSerializer serializer, final boolean consistentReads,
                                  @NonNull final TableCreatorCallback tableCreatorCallback, Duration dynamoDbRequestTimeout,
                                  final BillingMode billingMode); }
DynamoDBLeaseRefresher implements LeaseRefresher { @Override public boolean createLeaseTableIfNotExists(@NonNull final Long readCapacity, @NonNull final Long writeCapacity) throws ProvisionedThroughputException, DependencyException { try { if (tableStatus() != null) { return newTableCreated; } } catch (DependencyException de) { log.error("Failed to get table status for {}", table, de); } ProvisionedThroughput throughput = ProvisionedThroughput.builder().readCapacityUnits(readCapacity) .writeCapacityUnits(writeCapacity).build(); final CreateTableRequest request; if(BillingMode.PAY_PER_REQUEST.equals(billingMode)){ request = CreateTableRequest.builder().tableName(table).keySchema(serializer.getKeySchema()) .attributeDefinitions(serializer.getAttributeDefinitions()) .billingMode(billingMode).build(); }else{ request = CreateTableRequest.builder().tableName(table).keySchema(serializer.getKeySchema()) .attributeDefinitions(serializer.getAttributeDefinitions()).provisionedThroughput(throughput) .build(); } final AWSExceptionManager exceptionManager = createExceptionManager(); exceptionManager.add(ResourceInUseException.class, t -> t); exceptionManager.add(LimitExceededException.class, t -> t); try { try { FutureUtils.resolveOrCancelFuture(dynamoDBClient.createTable(request), dynamoDbRequestTimeout); newTableCreated = true; } catch (ExecutionException e) { throw exceptionManager.apply(e.getCause()); } catch (InterruptedException e) { throw new DependencyException(e); } } catch (ResourceInUseException e) { log.info("Table {} already exists.", table); return newTableCreated; } catch (LimitExceededException e) { throw new ProvisionedThroughputException("Capacity exceeded when creating table " + table, e); } catch (DynamoDbException | TimeoutException e) { throw new DependencyException(e); } return newTableCreated; } @Deprecated  DynamoDBLeaseRefresher(final String table, final DynamoDbAsyncClient dynamoDBClient,
            final LeaseSerializer serializer, final boolean consistentReads); @Deprecated  DynamoDBLeaseRefresher(final String table, final DynamoDbAsyncClient dynamoDBClient,
                                  final LeaseSerializer serializer, final boolean consistentReads,
                                  @NonNull final TableCreatorCallback tableCreatorCallback); @Deprecated  DynamoDBLeaseRefresher(final String table, final DynamoDbAsyncClient dynamoDBClient,
                                  final LeaseSerializer serializer, final boolean consistentReads,
                                  @NonNull final TableCreatorCallback tableCreatorCallback, Duration dynamoDbRequestTimeout);  DynamoDBLeaseRefresher(final String table, final DynamoDbAsyncClient dynamoDBClient,
                                  final LeaseSerializer serializer, final boolean consistentReads,
                                  @NonNull final TableCreatorCallback tableCreatorCallback, Duration dynamoDbRequestTimeout,
                                  final BillingMode billingMode); @Override boolean createLeaseTableIfNotExists(@NonNull final Long readCapacity, @NonNull final Long writeCapacity); @Override boolean leaseTableExists(); @Override boolean waitUntilLeaseTableExists(long secondsBetweenPolls, long timeoutSeconds); @Override List<Lease> listLeasesForStream(StreamIdentifier streamIdentifier); @Override List<Lease> listLeases(); @Override boolean isLeaseTableEmpty(); @Override boolean createLeaseIfNotExists(@NonNull final Lease lease); @Override Lease getLease(@NonNull final String leaseKey); @Override boolean renewLease(@NonNull final Lease lease); @Override boolean takeLease(@NonNull final Lease lease, @NonNull final String owner); @Override boolean evictLease(@NonNull final Lease lease); void deleteAll(); @Override void deleteLease(@NonNull final Lease lease); @Override boolean updateLease(@NonNull final Lease lease); @Override void updateLeaseWithMetaInfo(Lease lease, UpdateField updateField); @Override ExtendedSequenceNumber getCheckpoint(String leaseKey); }
DynamoDBLeaseRefresher implements LeaseRefresher { @Override public boolean createLeaseTableIfNotExists(@NonNull final Long readCapacity, @NonNull final Long writeCapacity) throws ProvisionedThroughputException, DependencyException { try { if (tableStatus() != null) { return newTableCreated; } } catch (DependencyException de) { log.error("Failed to get table status for {}", table, de); } ProvisionedThroughput throughput = ProvisionedThroughput.builder().readCapacityUnits(readCapacity) .writeCapacityUnits(writeCapacity).build(); final CreateTableRequest request; if(BillingMode.PAY_PER_REQUEST.equals(billingMode)){ request = CreateTableRequest.builder().tableName(table).keySchema(serializer.getKeySchema()) .attributeDefinitions(serializer.getAttributeDefinitions()) .billingMode(billingMode).build(); }else{ request = CreateTableRequest.builder().tableName(table).keySchema(serializer.getKeySchema()) .attributeDefinitions(serializer.getAttributeDefinitions()).provisionedThroughput(throughput) .build(); } final AWSExceptionManager exceptionManager = createExceptionManager(); exceptionManager.add(ResourceInUseException.class, t -> t); exceptionManager.add(LimitExceededException.class, t -> t); try { try { FutureUtils.resolveOrCancelFuture(dynamoDBClient.createTable(request), dynamoDbRequestTimeout); newTableCreated = true; } catch (ExecutionException e) { throw exceptionManager.apply(e.getCause()); } catch (InterruptedException e) { throw new DependencyException(e); } } catch (ResourceInUseException e) { log.info("Table {} already exists.", table); return newTableCreated; } catch (LimitExceededException e) { throw new ProvisionedThroughputException("Capacity exceeded when creating table " + table, e); } catch (DynamoDbException | TimeoutException e) { throw new DependencyException(e); } return newTableCreated; } @Deprecated  DynamoDBLeaseRefresher(final String table, final DynamoDbAsyncClient dynamoDBClient,
            final LeaseSerializer serializer, final boolean consistentReads); @Deprecated  DynamoDBLeaseRefresher(final String table, final DynamoDbAsyncClient dynamoDBClient,
                                  final LeaseSerializer serializer, final boolean consistentReads,
                                  @NonNull final TableCreatorCallback tableCreatorCallback); @Deprecated  DynamoDBLeaseRefresher(final String table, final DynamoDbAsyncClient dynamoDBClient,
                                  final LeaseSerializer serializer, final boolean consistentReads,
                                  @NonNull final TableCreatorCallback tableCreatorCallback, Duration dynamoDbRequestTimeout);  DynamoDBLeaseRefresher(final String table, final DynamoDbAsyncClient dynamoDBClient,
                                  final LeaseSerializer serializer, final boolean consistentReads,
                                  @NonNull final TableCreatorCallback tableCreatorCallback, Duration dynamoDbRequestTimeout,
                                  final BillingMode billingMode); @Override boolean createLeaseTableIfNotExists(@NonNull final Long readCapacity, @NonNull final Long writeCapacity); @Override boolean leaseTableExists(); @Override boolean waitUntilLeaseTableExists(long secondsBetweenPolls, long timeoutSeconds); @Override List<Lease> listLeasesForStream(StreamIdentifier streamIdentifier); @Override List<Lease> listLeases(); @Override boolean isLeaseTableEmpty(); @Override boolean createLeaseIfNotExists(@NonNull final Lease lease); @Override Lease getLease(@NonNull final String leaseKey); @Override boolean renewLease(@NonNull final Lease lease); @Override boolean takeLease(@NonNull final Lease lease, @NonNull final String owner); @Override boolean evictLease(@NonNull final Lease lease); void deleteAll(); @Override void deleteLease(@NonNull final Lease lease); @Override boolean updateLease(@NonNull final Lease lease); @Override void updateLeaseWithMetaInfo(Lease lease, UpdateField updateField); @Override ExtendedSequenceNumber getCheckpoint(String leaseKey);  }