Java 类com.amazonaws.services.dynamodbv2.AmazonDynamoDB 实例源码

private static AmazonDynamoDB getClient() {
    if (null != dynamodbClient) {
        return dynamodbClient;
    }

    String region = System.getProperty("DYNAMODB_REGION");
    if (null == region) {
        System.err.println("Region not set, default \"" + Regions.US_EAST_1.name() + "\" is used");
        region = Regions.US_EAST_1.name();
    }
    System.out.println("DynamoDB region: " + region);

    dynamodbClient = AmazonDynamoDBClientBuilder.standard()
            .withRegion(region)
            .build();

    return dynamodbClient;
}

public static final AmazonDynamoDB getClient() {
    if (null != dynamodbClient) {
        return dynamodbClient;
    }

    String region = System.getenv("DYNAMODB_REGION");
    if (null == region) {
        System.err.println("Region is null, using default \"" + Regions.US_WEST_1 + "\"");
        region = Regions.US_WEST_1.name();
    }
    System.out.println("DynamoDB region: " + region);

    dynamodbClient = AmazonDynamoDBClientBuilder.standard()
            .withRegion(region)
            .build();

    System.out.println("Got DynamoDB client...");

    return dynamodbClient;
}

private static AmazonDynamoDB getClient() {
    if (null != dynamodbClient)
        return dynamodbClient;

    String region = System.getProperty("DYNAMODB_REGION");
    if (null == region) {
        System.err.println("Region not set, default \"" + Regions.US_WEST_1.name() + "\" is used");
        region = Regions.US_WEST_1.name();
    }
    System.out.println("DynamoDB region: " + region);

    dynamodbClient = AmazonDynamoDBClientBuilder.standard()
            .withRegion(region)
            .build();

    return dynamodbClient;
}

@Inject
public DefaultGroupStorage(
    AmazonDynamoDB amazonDynamoDB,
    TableConfiguration tableConfiguration,
    @Named("dynamodbGroupWriteHystrix") HystrixConfiguration dynamodbGroupWriteHystrix,
    @Named("dynamodbGraphWriteHystrix") HystrixConfiguration dynamodbGraphWriteHystrix,
    @Named("dynamodbNamespaceGraphQueryHystrix")
        HystrixConfiguration dynamodbNamespaceGraphQueryHystrix,
    MetricRegistry metrics
) {
  this.amazonDynamoDB = amazonDynamoDB;
  this.dynamoDB = new DynamoDB(this.amazonDynamoDB);
  this.groupTableName = tableConfiguration.outlandGroupsTable;
  this.groupGraphTableName = tableConfiguration.outlandAppGraphTable;
  this.dynamodbGroupWriteHystrix = dynamodbGroupWriteHystrix;
  this.dynamodbGraphWriteHystrix = dynamodbGraphWriteHystrix;
  this.dynamodbNamespaceGraphQueryHystrix = dynamodbNamespaceGraphQueryHystrix;
  this.metrics = metrics;
}

private static void createTable(AmazonDynamoDB dynamodb, String tableName, String tableKeyFieldName) {
        List<AttributeDefinition> attributeDefinitions= new ArrayList<>();
        attributeDefinitions.add(new AttributeDefinition().withAttributeName(tableKeyFieldName).withAttributeType("S"));

        List<KeySchemaElement> keySchema = new ArrayList<>();
        keySchema.add(new KeySchemaElement().withAttributeName(tableKeyFieldName).withKeyType(KeyType.HASH));

        CreateTableRequest request = new CreateTableRequest()
                .withTableName(tableName)
                .withKeySchema(keySchema)
                .withAttributeDefinitions(attributeDefinitions)
                .withProvisionedThroughput(new ProvisionedThroughput()
                        .withReadCapacityUnits(1L)
                        .withWriteCapacityUnits(1L));

        dynamodb.createTable(request);
//TODO: just to look at the result, if needed
//        TableDescription table = dynamodb.describeTable(tableName).getTable();
    }

public GenericDynamoDB(AmazonDynamoDB client, AWSCredentialsProvider awsCredentials,
                       ClientConfiguration clientConfiguration,
                       SecretsGroupIdentifier groupIdentifier, Class<Entry> clazz, Converters converters,
                       ReadWriteLock readWriteLock) {
    this.clazz = clazz;
    buildMappings();
    this.converters = converters;
    this.awsCredentials = awsCredentials;
    this.clientConfiguration = clientConfiguration;
    this.client = client;
    this.region = RegionUtils.getRegion(groupIdentifier.region.getName());
    this.readWriteLock = readWriteLock;

    RegionLocalResourceName resourceName = new RegionLocalResourceName(groupIdentifier);
    this.tableName = resourceName.toString();
}

@Before
public void setUp() {
    dynamoDBClient = Mockito.mock(AmazonDynamoDB.class);

    GenerateDataKeyResult generateDatakeyResult = new GenerateDataKeyResult();
    generateDatakeyResult.setCiphertextBlob(Mockito.mock(ByteBuffer.class));
    generateDatakeyResult.setPlaintext(Mockito.mock(ByteBuffer.class));

    DecryptResult decryptResult = new DecryptResult();
    decryptResult.setKeyId("alias/foo");
    decryptResult.setPlaintext(Mockito.mock(ByteBuffer.class));

    awskmsClient = Mockito.mock(AWSKMS.class);
    Mockito.when(awskmsClient.generateDataKey(Mockito.any(GenerateDataKeyRequest.class))).thenReturn(generateDatakeyResult);
    Mockito.when(awskmsClient.decrypt(Mockito.any(DecryptRequest.class))).thenReturn(decryptResult);
}

boolean createItemIfNotExists(String key, long currentTimeMillis, Context context) {

        LambdaLogger logger = context.getLogger();
        AmazonDynamoDB client = createDynamoDBClient(cc);
        String functionName = context.getFunctionName();

        try {
            // Create a record if it does not exist
            PutItemRequest req = new PutItemRequest().withTableName(TABLE_NAME)
                    .addItemEntry(COL_FUNCTION_NAME, new AttributeValue(functionName))
                    .addItemEntry(COL_KEY, new AttributeValue(key))
                    .addItemEntry(COL_CREATED_TIME, new AttributeValue().withN(Long.toString(currentTimeMillis)))
                    .addExpectedEntry(COL_FUNCTION_NAME, new ExpectedAttributeValue().withExists(false))
                    .addExpectedEntry(COL_KEY, new ExpectedAttributeValue().withExists(false));
            client.putItem(req);
            return true;
        } catch (ConditionalCheckFailedException e) {
            logger.log("Record exsited. functionName[" + functionName + "] key[" + key + "]");
            return false;
        } finally {
            client.shutdown();
        }
    }

boolean createItemIfNotExists(String key, long currentTimeMillis, Context context) {

        LambdaLogger logger = context.getLogger();
        AmazonDynamoDB client = createDynamoDBClient(cc);
        String functionName = context.getFunctionName();

        try {
            // Create a record if it does not exist
            PutItemRequest req = new PutItemRequest().withTableName(TABLE_NAME)
                    .addItemEntry(COL_FUNCTION_NAME, new AttributeValue(functionName))
                    .addItemEntry(COL_KEY, new AttributeValue(key))
                    .addItemEntry(COL_CREATED_TIME, new AttributeValue().withN(Long.toString(currentTimeMillis)))
                    .addExpectedEntry(COL_FUNCTION_NAME, new ExpectedAttributeValue().withExists(false))
                    .addExpectedEntry(COL_KEY, new ExpectedAttributeValue().withExists(false));
            client.putItem(req);
            return true;
        } catch (ConditionalCheckFailedException e) {
            logger.log("Record exsited. functionName[" + functionName + "] key[" + key + "]");
            return false;
        } finally {
            client.shutdown();
        }
    }

boolean createItemIfNotExists(String key, long currentTimeMillis, Context context) {

        LambdaLogger logger = context.getLogger();
        AmazonDynamoDB client = createDynamoDBClient(cc);
        String functionName = context.getFunctionName();

        try {
            // Create a record if it does not exist
            PutItemRequest req = new PutItemRequest().withTableName(TABLE_NAME)
                    .addItemEntry(COL_FUNCTION_NAME, new AttributeValue(functionName))
                    .addItemEntry(COL_KEY, new AttributeValue(key))
                    .addItemEntry(COL_CREATED_TIME, new AttributeValue().withN(Long.toString(currentTimeMillis)))
                    .addExpectedEntry(COL_FUNCTION_NAME, new ExpectedAttributeValue().withExists(false))
                    .addExpectedEntry(COL_KEY, new ExpectedAttributeValue().withExists(false));
            client.putItem(req);
            return true;
        } catch (ConditionalCheckFailedException e) {
            logger.log("Record exsited. functionName[" + functionName + "] key[" + key + "]");
            return false;
        } finally {
            client.shutdown();
        }
    }

boolean createItemIfNotExists(String key, long currentTimeMillis, Context context) {

        LambdaLogger logger = context.getLogger();
        AmazonDynamoDB client = createDynamoDBClient(cc);
        String functionName = context.getFunctionName();

        try {
            // Create a record if it does not exist
            PutItemRequest req = new PutItemRequest().withTableName(TABLE_NAME)
                    .addItemEntry(COL_FUNCTION_NAME, new AttributeValue(functionName))
                    .addItemEntry(COL_KEY, new AttributeValue(key))
                    .addItemEntry(COL_CREATED_TIME, new AttributeValue().withN(Long.toString(currentTimeMillis)))
                    .addExpectedEntry(COL_FUNCTION_NAME, new ExpectedAttributeValue().withExists(false))
                    .addExpectedEntry(COL_KEY, new ExpectedAttributeValue().withExists(false));
            client.putItem(req);
            return true;
        } catch (ConditionalCheckFailedException e) {
            logger.log("Record exsited. functionName[" + functionName + "] key[" + key + "]");
            return false;
        } finally {
            client.shutdown();
        }
    }

public static final AmazonDynamoDB getClient() {
    if (null != dynamodbClient) {
        return dynamodbClient;
    }

    String region = System.getenv("DYNAMODB_REGION");
    if (null == region) {
        System.err.println("Region is null, using default \"" + Regions.US_WEST_1 + "\"");
        region = Regions.US_WEST_1.name();
    }
    System.out.println("DynamoDB region: " + region);

    dynamodbClient = AmazonDynamoDBClientBuilder.standard()
            .withRegion(region)
            .build();

    System.out.println("Got DynamoDB client...");

    return dynamodbClient;
}

@Test
public void testProvide() {
    Env env = new Env().register("DYNAMODB_ENDPOINT", "fakeendpoint");
    ClientConfiguration config = new ClientConfiguration();
    config.setRetryPolicy(PredefinedRetryPolicies.NO_RETRY_POLICY);

    AWSCredentialsProvider credentialsProvider = new StaticCredentialsProvider(
            new BasicAWSCredentials("fake_access_key_id", "fake_secret_access_key"));

    AmazonDynamoDBFactory factory = new AmazonDynamoDBFactory(env, credentialsProvider, config);
    AmazonDynamoDB dynamoDB = factory.provide();

    String message = "";

    try {
        dynamoDB.listTables();  
    } catch (AmazonClientException e) {
        message = e.getMessage();
    }

    assertTrue(message.startsWith("Unable to execute HTTP request: fakeendpoint"));

    factory.dispose(dynamoDB);
}

/**
 * Collect data for DynamoDB.
 *
 * @param stats
 *            current statistics object.
 * @param account
 *            currently used credentials object.
 * @param region
 *            currently used aws region.
 */
public static void scanDynamoDB(AwsStats stats, AwsAccount account, Regions region) {
    LOG.debug("Scan for DynamoDB in region " + region.getName() + " in account " + account.getAccountId());

    /*
     * Amazon DynamoDB
     */
    try {
        AmazonDynamoDB dynamoDB = new AmazonDynamoDBClient(account.getCredentials());
        dynamoDB.setRegion(Region.getRegion(region));

        List<String> list = dynamoDB.listTables().getTableNames();

        int totalItems = list.size();
        for (String tableName : list) {
            AwsResource res = new AwsResource(tableName, account.getAccountId(), AwsResourceType.DynamoDB, region);
            stats.add(res);
        }

        LOG.info(totalItems + " DynamoDB tables in region " + region.getName() + " in account " + account.getAccountId());
    } catch (AmazonServiceException ase) {
        LOG.error("Exception of DynamoDB: " + ase.getMessage());
    }
}

public KinesisShardCheckpointer(AmazonDynamoDB dynamoDBClient,
        String dynamoDBTable,
        KinesisSplit kinesisSplit,
        String logicalProcessName,
        int curIterationNumber,
        long checkpointIntervalMS,
        long dynamoReadCapacity,
        long dynamoWriteCapacity)
{
    this(new KinesisClientLeaseManager(dynamoDBTable, dynamoDBClient),
            kinesisSplit,
            logicalProcessName,
            curIterationNumber,
            checkpointIntervalMS,
            dynamoReadCapacity,
            dynamoWriteCapacity);
}

@Test
public void shouldNotGenerateKeys_withNoSequenceConfigurations() throws Exception {
    // Given
    final String sequenceName = randomString(10);
    final Collection<SequenceConfiguration> sequenceConfigurations = new ArrayList<>();
    when(mockDatabaseSchemaHolder.sequenceConfigurations()).thenReturn(sequenceConfigurations);
    final AmazonDynamoDB mockAmazonDynamoDbClient = mock(AmazonDynamoDB.class);
    final SequenceKeyGenerator sequenceKeyGenerator = new SequenceKeyGenerator(sequenceName);
    final DynamoDbTemplate dynamoDbTemplate = new DynamoDbTemplate(mockDatabaseSchemaHolder);
    dynamoDbTemplate.initialize(mockAmazonDynamoDbClient);

    // When
    IllegalStateException actualException = null;
    try {
        dynamoDbTemplate.generateKeys(sequenceKeyGenerator);
    } catch (final IllegalStateException e) {
        actualException = e;
    }

    // Then
    assertNotNull(actualException);
}

/**
 * Private interface for creating tables which handles any instances of
 * Throttling of the API
 * 
 * @param dynamoClient
 * @param dynamoTable
 * @return
 * @throws Exception
 */
public static CreateTableResult safeCreateTable(
        final AmazonDynamoDB dynamoClient,
        final CreateTableRequest createTableRequest) throws Exception {
    CreateTableResult res = null;
    final int tryMax = 10;
    int tries = 0;
    while (true) {
        try {
            res = dynamoClient.createTable(createTableRequest);
            return res;
        } catch (LimitExceededException le) {
            if (tries < tryMax) {
                // back off for 1 second
                Thread.sleep(1000);
                tries++;
            } else {
                throw le;
            }
        } catch (ResourceInUseException rie) {
            // someone else is trying to create the table while we are, so
            // return ok
            return null;
        }
    }
}

/**
 * Private interface for describing tables which handles any instances of
 * Throttling of the API
 * 
 * @param dynamoClient
 * @param dynamoTable
 * @return
 * @throws Exception
 */
public static DescribeTableResult safeDescribeTable(
        final AmazonDynamoDB dynamoClient, final String dynamoTable)
        throws Exception {
    DescribeTableResult res = null;
    final int tryMax = 10;
    int tries = 0;
    while (true) {
        try {
            res = dynamoClient.describeTable(dynamoTable);

            return res;
        } catch (ResourceNotFoundException e) {
            if (tries < tryMax) {
                // sleep for a short time as this is potentially an eventual
                // consistency issue with the table having been created ms
                // ago
                Thread.sleep(10);
                tries++;
            } else {
                throw e;
            }
        }
    }
}

/**
 * Interface which will block until a dynamo table reaches a specified
 * state. Also returns immediately if the object doesn't exist
 * 
 * @param dynamoClient
 *            Dynamo DB Client to use for connection to Dynamo DB.
 * @param dynamoTable
 *            The table name to check.
 * @param status
 *            The status to wait for
 * @throws Exception
 */
private static void waitForTableState(final AmazonDynamoDB dynamoClient,
        final String dynamoTable, TableStatus status) throws Exception {
    DescribeTableResult tableRequest = null;
    while (true) {
        try {
            tableRequest = dynamoClient.describeTable(dynamoTable);
            if (tableRequest.getTable().getTableStatus()
                    .equals(status.name()))
                break;

            Thread.sleep(1000);
        } catch (InterruptedException e) {
            return;
        }
    }
}

/**
 * Generate a list of attribute names found in the Aggregator's dynamo
 * table. Assumes that all Items in the Aggregator table are of the same
 * structure.
 * 
 * @param dynamoClient
 *            Dynamo DB Client to use for connection to Dynamo DB.
 * @param dynamoTable
 *            The Dynamo Table for the Aggregator
 * @return A list of attribute names from the Dynamo table
 * @throws Exception
 */
public static List<String> getDictionaryEntry(
        final AmazonDynamoDB dynamoClient, final String dynamoTable)
        throws Exception {
    // get a list of all columns in the table, with keys first
    List<String> columns = new ArrayList<>();
    List<KeySchemaElement> keys = dynamoClient.describeTable(dynamoTable)
            .getTable().getKeySchema();
    for (KeySchemaElement key : keys) {
        columns.add(key.getAttributeName());
    }
    ScanResult scan = dynamoClient.scan(new ScanRequest()
            .withTableName(dynamoTable).withSelect(Select.ALL_ATTRIBUTES)
            .withLimit(1));
    List<Map<String, AttributeValue>> scannedItems = scan.getItems();
    for (Map<String, AttributeValue> map : scannedItems) {
        for (String s : map.keySet()) {
            if (!columns.contains(s))
                columns.add(s);
        }
    }

    return columns;
}

public static List<Map<String, AttributeValue>> queryUntilDone(
        AmazonDynamoDB dynamoClient, QueryRequest qr, int backoffMillis)
        throws Exception {
    List<Map<String, AttributeValue>> output = new ArrayList<>();

    Map<String, AttributeValue> lastKeyEvaluated = null;
    do {
        int queryAttempts = 0;
        QueryResult result = null;

        do {
            try {
                result = dynamoClient.query(qr).withLastEvaluatedKey(
                        lastKeyEvaluated);

                output.addAll(result.getItems());
            } catch (ProvisionedThroughputExceededException e) {
                LOG.warn(String
                        .format("Provisioned Throughput Exceeded - Retry Attempt %s",
                                queryAttempts));

                Thread.sleep(2 ^ queryAttempts * backoffMillis);

                queryAttempts++;
            }
        } while (queryAttempts < 10 && result == null);

        if (result == null) {
            throw new Exception(String.format(
                    "Unable to execute Query after %s attempts",
                    queryAttempts));
        }

        lastKeyEvaluated = result.getLastEvaluatedKey();
    } while (lastKeyEvaluated != null);

    return output;
}

/**
 * Returns a client instance for AWS DynamoDB.
 * @return a client that talks to DynamoDB
 */
public static AmazonDynamoDB getClient() {
    if (ddbClient != null) {
        return ddbClient;
    }

    if (Config.IN_PRODUCTION) {
        ddbClient = AmazonDynamoDBClientBuilder.standard().withCredentials(new AWSStaticCredentialsProvider(
            new BasicAWSCredentials(Config.AWS_ACCESSKEY, Config.AWS_SECRETKEY))).
                withRegion(Config.AWS_REGION).build();
    } else {
        ddbClient = AmazonDynamoDBClientBuilder.standard().
                withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("local", "null"))).
                withEndpointConfiguration(new EndpointConfiguration(LOCAL_ENDPOINT, "")).build();
    }

    if (!existsTable(Config.getRootAppIdentifier())) {
        createTable(Config.getRootAppIdentifier());
    }

    ddb = new DynamoDB(ddbClient);

    Para.addDestroyListener(new DestroyListener() {
        public void onDestroy() {
            shutdownClient();
        }
    });

    return ddbClient;
}

/**
 * Executes a create table request using the DynamoDB client and waits the
 * default time until it's been created.
 * 
 * @param awsClient
 * @param keySchema
 * @param tableName
 * @param proThrou
 */
public static void executeCreateTableRequest(AmazonDynamoDB awsClient, String tableName,
    ArrayList<KeySchemaElement> keySchema, Map<String, String> attrs, ProvisionedThroughput proThrou) {
  CreateTableRequest createTableRequest = buildCreateTableRequest(tableName,
      keySchema, proThrou, attrs);
  // use the client to perform the request
  try {
    awsClient.createTable(createTableRequest).getTableDescription();
    // wait for table to become active
    waitForTableToBecomeAvailable(awsClient, tableName);
  } catch (ResourceInUseException ex) {
    LOG.warn("Table '{}' already exists.", tableName);
  } finally {
    LOG.info("Table '{}' is available.", tableName);
  }
}

/**
 * Retrieves the stored ETag, if one exists, from DynamoDB.
 *
 * @param dynamoDB
 *            DynamoDB client configured with a region and credentials
 * @param table
 *            The resource table name
 * @param resource
 *            The URL String of the resource
 * @return The ETag String of the last copy processed or null if the resource has never been processed
 */
public static String getStoredETag(final AmazonDynamoDB dynamoDB, final String table, final String resource) {
    String oldETag;
    // Build key to retrieve item
    final Map<String, AttributeValue> resourceKey = new HashMap<String, AttributeValue>();
    resourceKey.put(MarsDynamoDBManager.RESOURCE_TABLE_HASH_KEY, new AttributeValue(resource));
    // Get item
    final GetItemResult result = dynamoDB.getItem(table, resourceKey);
    final Map<String, AttributeValue> item = result.getItem();
    if (item != null && item.containsKey(ETAG_KEY)) {
        // Item was found and contains ETag
        oldETag = item.get(ETAG_KEY).getS();
    } else {
        // Item was not found or did not contain ETag
        oldETag = null;
    }
    return oldETag;
}

@Test
public void testCreateResourceTable() {
    final AmazonDynamoDB dynamoDB = PowerMock.createMock(AmazonDynamoDB.class);
    PowerMock.mockStatic(DynamoDBManager.class);
    final CreateTableRequest request = new CreateTableRequest();
    request.setAttributeDefinitions(MarsDynamoDBManager.RESOURCE_TABLE_ATTRIBUTE_DEFINITIONS);
    request.setKeySchema(MarsDynamoDBManager.RESOURCE_TABLE_KEY_SCHEMA);
    request.setProvisionedThroughput(PROVISIONED_THROUGHPUT);
    request.setTableName(TABLE_NAME);

    DynamoDBManager.createTable(dynamoDB, request);
    PowerMock.expectLastCall().andReturn(null);
    PowerMock.replayAll();
    MarsDynamoDBManager.createResourceTable(dynamoDB, TABLE_NAME, PROVISIONED_THROUGHPUT);
    PowerMock.verifyAll();
}

@Test
public void testGetStoredETagExists() {
    AmazonDynamoDB dynamoDB = PowerMock.createMock(AmazonDynamoDB.class);
    Map<String, AttributeValue> resourceKey = new HashMap<String, AttributeValue>();
    resourceKey.put(MarsDynamoDBManager.RESOURCE_TABLE_HASH_KEY, new AttributeValue(resource));
    // Get item
    dynamoDB.getItem(table, resourceKey);
    Map<String, AttributeValue> resourceResult = new HashMap<String, AttributeValue>();
    resourceResult.put(MarsDynamoDBManager.RESOURCE_TABLE_HASH_KEY, new AttributeValue(resource));
    resourceResult.put(DynamoDBWorkerUtils.ETAG_KEY, new AttributeValue(eTag));
    GetItemResult result = new GetItemResult().withItem(resourceResult);
    PowerMock.expectLastCall().andReturn(result);
    PowerMock.replayAll();
    String resultETag = DynamoDBWorkerUtils.getStoredETag(dynamoDB, table, resource);
    assertEquals(eTag, resultETag);
    PowerMock.verifyAll();
}

@Test
public void createTableTest() {
    AmazonDynamoDB ddb = DynamoDBEmbedded.create().amazonDynamoDB();
    try {
        String tableName = "Movies";
        String hashKeyName = "film_id";
        CreateTableResult res = createTable(ddb, tableName, hashKeyName);

        TableDescription tableDesc = res.getTableDescription();
        assertEquals(tableName, tableDesc.getTableName());
        assertEquals("[{AttributeName: " + hashKeyName + ",KeyType: HASH}]", tableDesc.getKeySchema().toString());
        assertEquals("[{AttributeName: " + hashKeyName + ",AttributeType: S}]",
            tableDesc.getAttributeDefinitions().toString());
        assertEquals(Long.valueOf(1000L), tableDesc.getProvisionedThroughput().getReadCapacityUnits());
        assertEquals(Long.valueOf(1000L), tableDesc.getProvisionedThroughput().getWriteCapacityUnits());
        assertEquals("ACTIVE", tableDesc.getTableStatus());
        assertEquals("arn:aws:dynamodb:ddblocal:000000000000:table/Movies", tableDesc.getTableArn());

        ListTablesResult tables = ddb.listTables();
        assertEquals(1, tables.getTableNames().size());
    } finally {
        ddb.shutdown();
    }
}

private static CreateTableResult createTable(AmazonDynamoDB ddb, String tableName, String hashKeyName) {
    List<AttributeDefinition> attributeDefinitions = new ArrayList<AttributeDefinition>();
    attributeDefinitions.add(new AttributeDefinition(hashKeyName, ScalarAttributeType.S));

    List<KeySchemaElement> ks = new ArrayList<KeySchemaElement>();
    ks.add(new KeySchemaElement(hashKeyName, KeyType.HASH));

    ProvisionedThroughput provisionedthroughput = new ProvisionedThroughput(1000L, 1000L);

    CreateTableRequest request =
        new CreateTableRequest()
            .withTableName(tableName)
            .withAttributeDefinitions(attributeDefinitions)
            .withKeySchema(ks)
            .withProvisionedThroughput(provisionedthroughput);

    return ddb.createTable(request);
}

/**
 * Constructs a {@link DynamoDBRepositoryBean}.
 * 
 * @param beanManager
 *            must not be {@literal null}.
 * @param dynamoDBMapperBean
 *            must not be {@literal null}.
 * @param qualifiers
 *            must not be {@literal null}.
 * @param repositoryType
 *            must not be {@literal null}.
 */
DynamoDBRepositoryBean(BeanManager beanManager, Bean<AmazonDynamoDB> amazonDynamoDBBean,
        Bean<DynamoDBMapperConfig> dynamoDBMapperConfigBean,Bean<DynamoDBOperations> dynamoDBOperationsBean, Set<Annotation> qualifiers, Class<T> repositoryType) {

    super(qualifiers, repositoryType, beanManager);
    if (dynamoDBOperationsBean == null)
    {
        Assert.notNull(amazonDynamoDBBean);
    }
    else
    {
        Assert.isNull(amazonDynamoDBBean,"Cannot specify both amazonDynamoDB bean and dynamoDBOperationsBean in repository configuration");
        Assert.isNull(dynamoDBMapperConfigBean,"Cannot specify both dynamoDBMapperConfigBean bean and dynamoDBOperationsBean in repository configuration");

    }
    this.amazonDynamoDBBean = amazonDynamoDBBean;
    this.dynamoDBMapperConfigBean = dynamoDBMapperConfigBean;
    this.dynamoDBOperationsBean = dynamoDBOperationsBean;
}

@Override
public T create(CreationalContext<T> creationalContext, Class<T> repositoryType) {

    // Get an instance from the associated AmazonDynamoDB bean.
    AmazonDynamoDB amazonDynamoDB = getDependencyInstance(amazonDynamoDBBean, AmazonDynamoDB.class);

    // Get an instance from the associated optional AmazonDynamoDB bean.
    DynamoDBMapperConfig dynamoDBMapperConfig = dynamoDBMapperConfigBean == null ? null : getDependencyInstance(
            dynamoDBMapperConfigBean, DynamoDBMapperConfig.class);

    DynamoDBOperations dynamoDBOperations = dynamoDBOperationsBean == null ? null
            :  getDependencyInstance(
                    dynamoDBOperationsBean, DynamoDBOperations.class);

    if (dynamoDBOperations == null)
    {
        dynamoDBOperations = new DynamoDBTemplate(amazonDynamoDB,dynamoDBMapperConfig);
    }

    DynamoDBRepositoryFactory factory = new DynamoDBRepositoryFactory(dynamoDBOperations);
    return factory.getRepository(repositoryType);
}

/**
 * Creates a {@link Bean}.
 * 
 * @param <T>
 *            The type of the repository.
 * @param repositoryType
 *            The class representing the repository.
 * @param beanManager
 *            The BeanManager instance.
 * @return The bean.
 */
private <T> Bean<T> createRepositoryBean(Class<T> repositoryType, Set<Annotation> qualifiers, BeanManager beanManager) {

    // Determine the amazondbclient bean which matches the qualifiers of the
    // repository.
    Bean<AmazonDynamoDB> amazonDynamoDBBean = amazonDynamoDBs.get(qualifiers);

    // Determine the dynamo db mapper configbean which matches the
    // qualifiers of the repository.
    Bean<DynamoDBMapperConfig> dynamoDBMapperConfigBean = dbMapperConfigs.get(qualifiers);

    if (amazonDynamoDBBean == null) {
        throw new UnsatisfiedResolutionException(String.format("Unable to resolve a bean for '%s' with qualifiers %s.",
                AmazonDynamoDBClient.class.getName(), qualifiers));
    }

    Bean<DynamoDBOperations> dynamoDBOperationsBean = dynamoDBOperationss.get(qualifiers);


    // Construct and return the repository bean.
    return new DynamoDBRepositoryBean<T>(beanManager, amazonDynamoDBBean, dynamoDBMapperConfigBean,dynamoDBOperationsBean,qualifiers,
            repositoryType);
}

/**
 * Create table.
 * @throws InterruptedException If something fails
 */
public void create() throws InterruptedException {
    final AmazonDynamoDB aws = this.region.aws();
    final String name = this.request.getTableName();
    aws.createTable(this.request);
    Logger.info(this, "DynamoDB table '%s' creation requested...", name);
    final DescribeTableRequest req = new DescribeTableRequest()
        .withTableName(name);
    while (true) {
        final DescribeTableResult result = aws.describeTable(req);
        if ("ACTIVE".equals(result.getTable().getTableStatus())) {
            Logger.info(
                this,
                "DynamoDB table '%s' is %s",
                name, result.getTable().getTableStatus()
            );
            break;
        }
        Logger.info(
            this,
            "waiting for DynamoDB table '%s': %s",
            name, result.getTable().getTableStatus()
        );
        TimeUnit.SECONDS.sleep((long) Tv.TEN);
    }
}

/**
 * Throughput can change throughput of a table.
 * @throws Exception If something went wrong
 */
@Test
public final void adjustsThroughput() throws Exception {
    final Table table = Mockito.mock(Table.class);
    final Region region = Mockito.mock(Region.class);
    final AmazonDynamoDB aws = Mockito.mock(AmazonDynamoDB.class);
    Mockito.when(table.region()).thenReturn(region);
    final String name = "Customers";
    Mockito.when(table.name()).thenReturn(name);
    Mockito.when(region.aws()).thenReturn(aws);
    new Throughput(table).adjust();
    Mockito.verify(aws, Mockito.times(1))
        .updateTable(
            Mockito.eq(name),
            Mockito.<ProvisionedThroughput>any()
        );
}

/**
 * Creates tables.
 * @throws IOException if something goes wrong
 */
public void create() throws IOException {
    final AmazonDynamoDB aws = new AmazonDynamoDBClient(
        new BasicAWSCredentials(this.key, this.secret)
    );
    aws.setEndpoint(String.format("%s:%d", this.endpoint, this.port));
    for (final String table : this.locations) {
        final JsonObject json = this.readJson(table);
        if (json.containsKey("TableName")) {
            final String name = json.getString("TableName");
            if (Tables.exists(aws, name)) {
                Logger.info(
                    this, "Table '%s' already exists, skipping...", name
                );
            } else {
                this.createTable(aws, json);
            }
        } else {
            throw new IOException(
                String.format(
                    "File '%s' does not specify TableName attribute", table
                )
            );
        }
    }
}

public MAVLinkMessagesTable() {
    if (System.getenv(SPL_DYNAMODB_TABLE) != null) {
        tableName = System.getenv(SPL_DYNAMODB_TABLE);
    }

    AmazonDynamoDB dynamoDBClient = AmazonDynamoDBClientBuilder.defaultClient();

    DynamoDB dynamoDB = new DynamoDB(dynamoDBClient);

    table = dynamoDB.getTable(tableName);
}

/**
 * Creates the DynamoDB client {@link Bean}.
 *
 * Uses the default client, but if a region is unspecified, uses {@code us-east-1}.
 *
 * @return The DynamoDB client.
 */
@Bean
public AmazonDynamoDB dynamoDbClient() {
    AmazonDynamoDB client;

    try {
        client = AmazonDynamoDBClientBuilder.defaultClient();
    } catch (SdkClientException exception) {
        API_LOG.info("Default DynamoDB client failed to build, trying again with region us-east-1", exception);
        client = planB();
    }

    return client;
}

@Test
public void testDefaultClient() {
    mockStatic(AmazonDynamoDBClientBuilder.class);
    when(AmazonDynamoDBClientBuilder.defaultClient()).thenReturn(Mockito.mock(AmazonDynamoDB.class));
    assertNotNull(underTest.dynamoDbClient());
    verify(underTest, times(0)).planB();
}

@Inject
public DynamoCreateFeatureTableTask(
    AmazonDynamoDB amazonDynamoDB,
    TableConfiguration tableConfiguration
) {
  super("DynamoCreateFeatureTableTask");
  this.amazonDynamoDB = amazonDynamoDB;
  this.tableName = tableConfiguration.outlandFeaturesTable;
}

@Inject
public DynamoCreateGraphTableTask(
    AmazonDynamoDB dynamoDB,
    TableConfiguration tableConfiguration
) {
  super("DynamoCreateGraphTableTask");
  this.dynamoDB = dynamoDB;
  this.tableConfiguration = tableConfiguration;
}

@Inject
public DynamoCreateGroupsTableTask(
    AmazonDynamoDB amazonDynamoDB,
    TableConfiguration tableConfiguration
) {
  super("DynamoCreateGroupsTableTask");
  this.amazonDynamoDB = amazonDynamoDB;
  this.tableConfiguration = tableConfiguration;
}