/**
 * Creates the table in DynamoDB. The hash+range key is:
 * 
 *           Hash (String)        |    Range (String)
 *             File Path                 File Name
 * Example:  s3n://netflix/data           test.xml
 *   
 * The table also includes one attribute for epoch (time created),but
 * that is only defined during the put operation (see add()).
 *
 */
private void createTable() {
    log.info("Creating table in DynamoDB: " + tableName);

    CreateTableRequest createRequest = new CreateTableRequest();
    createRequest.withTableName(tableName);

    //Key
    KeySchemaElement pathKey = new KeySchemaElement().withAttributeName(HASH_KEY).withAttributeType(ScalarattributeType.S);
    KeySchemaElement fileKey = new KeySchemaElement().withAttributeName(RANGE_KEY).withAttributeType(ScalarattributeType.S);
    KeySchema schema = new KeySchema();
    schema.setHashKeyElement(pathKey);
    schema.setRangeKeyElement(fileKey);
    createRequest.setKeySchema(schema);

    //Throughput
    ProvisionedThroughput tp = new ProvisionedThroughput();
    tp.setReadCapacityUnits(readUnits);
    tp.setWriteCapacityUnits(writeUnits);

    createRequest.setProvisionedThroughput(tp);

    db.createTable(createRequest);
}

/**
 * Executes a create table request using the DynamoDB client and waits
 * the default time until it's been created.
 * @param tableName
 */
private void executeCreateTableRequest(String tableName){
  CreateTableRequest createTableRequest = getCreateTableRequest(tableName,mapping.getKeySchema(tableName),mapping.getProvisionedThroughput(tableName));
  // use the client to perform the request
  dynamoDBClient.createTable(createTableRequest).getTableDescription();
  // wait for table to become active
  waitForTabletoBecomeAvailable(tableName);
  LOG.info(tableName + "Schema Now available");
}

/**
 * Builds the necessary requests to create tables 
 * @param tableName
 * @param keySchema
 * @param proThrou
 * @return
 */
private CreateTableRequest getCreateTableRequest(String tableName,KeySchema keySchema,ProvisionedThroughput proThrou){
  CreateTableRequest createTableRequest = new CreateTableRequest();
  createTableRequest.setTableName(tableName);
  createTableRequest.setKeySchema(keySchema);
  createTableRequest.setProvisionedThroughput(proThrou);
  return createTableRequest;
}

/**
 * Executes a create table request using the DynamoDB client and waits
 * the default time until it's been created.
 * @param tableName
 */
private void executeCreateTableRequest(String tableName){
  CreateTableRequest createTableRequest = getCreateTableRequest(tableName,mapping.getProvisionedThroughput(tableName));
  // use the client to perform the request
  dynamoDBClient.createTable(createTableRequest).getTableDescription();
  // wait for table to become active
  waitForTabletoBecomeAvailable(tableName);
  LOG.info(tableName + "Schema Now available");
}

/**
 * Builds the necessary requests to create tables 
 * @param tableName
 * @param keySchema
 * @param proThrou
 * @return
 */
private CreateTableRequest getCreateTableRequest(String tableName,ProvisionedThroughput proThrou){
  CreateTableRequest createTableRequest = new CreateTableRequest();
  createTableRequest.setTableName(tableName);
  createTableRequest.setKeySchema(keySchema);
  createTableRequest.setProvisionedThroughput(proThrou);
  return createTableRequest;
}

public CreateTableRequest constructCreateTableRequest() {
    ArrayList<AttributeDeFinition> attributeDeFinitions = new ArrayList<>();
    attributeDeFinitions.add(new AttributeDeFinition().withAttributeName(partitionKeyName.toString()).withAttributeType("S"));
    attributeDeFinitions.add(new AttributeDeFinition().withAttributeName(sortKeyName.toString()).withAttributeType("N"));

    ArrayList<KeySchemaElement> keySchema = new ArrayList<>();
    keySchema.add(new KeySchemaElement().withAttributeName(partitionKeyName.toString()).withKeyType(KeyType.HASH));
    keySchema.add(new KeySchemaElement().withAttributeName(sortKeyName.toString()).withKeyType(KeyType.RANGE));

    ProvisionedThroughput provisionedThroughput = new ProvisionedThroughput()
            .withReadCapacityUnits(1L)
            .withWriteCapacityUnits(1L);
    CreateTableRequest request = new CreateTableRequest()
            .withTableName(tableName)
            .withKeySchema(keySchema)
            .withAttributeDeFinitions(attributeDeFinitions)
            .withProvisionedThroughput(provisionedThroughput);
    return request;
}

@Test
public void testCreateTableWithWait() throws Exception {
    // Create fake responses from AWS. First response is still creating the table,second response the table
    // has become active.
    TableDescription creatingDescription = constructTableDescription(TableStatus.CREATING);
    TableDescription createdDescription = constructTableDescription(TableStatus.ACTIVE);
    CreateTableResult mockCreateResult = new CreateTableResult().withTableDescription(creatingDescription);
    DescribeTableResult mockDescribeResultCreating = new DescribeTableResult().withTable(creatingDescription);
    DescribeTableResult mockDescribeResultCreated = new DescribeTableResult().withTable(createdDescription);

    // Create the table.
    CreateTableRequest expectedRequest = dynamoDB.constructCreateTableRequest();
    when(mockDynamoDBClient.createTable(expectedRequest)).thenReturn(mockCreateResult);
    when(mockDynamoDBClient.describeTable(tableName)).thenReturn(mockDescribeResultCreating,mockDescribeResultCreated);
    assertEquals(dynamoDB.create(),TEST_ARN);

    verify(mockDynamoDBClient,times(1)).createTable(expectedRequest);
    verify(mockDynamoDBClient,times(2)).describeTable(tableName);
}

private void createRecipientTable() {
    CreateTableRequest request
            = new CreateTableRequest()
                    .withTableName(TABLE_NAME)
                    .withAttributeDeFinitions(
                            new AttributeDeFinition("_id",ScalarattributeType.S)
                    )
                    .withKeySchema(
                            new KeySchemaElement("_id",KeyType.HASH)
                    )
                    .withProvisionedThroughput(new ProvisionedThroughput(10L,10L));

    ddb.createTable(request);
    try {
        Tableutils.waitUntilActive(ddb,TABLE_NAME);
    } catch (InterruptedException  e) {
        throw new RuntimeException(e);
    }
}

protected Table createHashAndSortTable(String pk,String sort) throws InterruptedException {
  ArrayList<AttributeDeFinition> attributeDeFinitions = new ArrayList<>();
  ScalarattributeType type = ScalarattributeType.S;
  attributeDeFinitions.add(new AttributeDeFinition()
    .withAttributeName(pk).withAttributeType(type));
  attributeDeFinitions
    .add(new AttributeDeFinition().withAttributeName(sort).withAttributeType(type));
  ArrayList<KeySchemaElement> keySchema = new ArrayList<>();
  keySchema.add(new KeySchemaElement().withAttributeName(pk).withKeyType(KeyType.HASH));
  keySchema.add(new KeySchemaElement().withAttributeName(sort).withKeyType(KeyType.RANGE));

  CreateTableRequest request = new CreateTableRequest()
    .withKeySchema(keySchema)
    .withAttributeDeFinitions(attributeDeFinitions);
  return createTable(request);
}

protected Table createTable(CreateTableRequest request) throws InterruptedException {
  DynamoDB dynamoDB = new DynamoDB(tables.getAsyncclient());
  request.withProvisionedThroughput(new ProvisionedThroughput()
    .withReadCapacityUnits(5L)
    .withWriteCapacityUnits(6L));

  if (request.getTableName() == null) {
    String tableName = tables.getTestTableName();
    tableName = tableName.replace('-','_');
    request.setTableName(tableName);
  }

  Table table = dynamoDB.createTable(request);
  table.waitForActive();
  return table;
}

public void deploy() {

        final AttributeDeFinition idAttr = new AttributeDeFinition().withAttributeName("id")
                .withAttributeType(ScalarattributeType.S);
        final ProvisionedThroughput throughput = new ProvisionedThroughput().withReadCapacityUnits(5L)
                .withWriteCapacityUnits(5L);

        final KeySchemaElement idKey = new KeySchemaElement().withAttributeName("id").withKeyType(KeyType.HASH);

        final CreateTableRequest createTableRequest = new CreateTableRequest().withTableName("TranslateSlack")
                .withAttributeDeFinitions(idAttr)
                .withKeySchema(idKey)
                .withProvisionedThroughput(throughput);
        ;
        ;

        ddb.createTable(createTableRequest);

    }

@Before
public void setUp() {
    // Unique table for each run
    tableName = "table" + String.valueOf(tableCount++);
    dynamoDB.getDynamoDbClient().createTable(
        new CreateTableRequest()
            .withTableName(tableName)
            .withKeySchema(new KeySchemaElement("id","HASH"))
            .withAttributeDeFinitions(
                new AttributeDeFinition("id","S")
            )
            .withProvisionedThroughput(
                new ProvisionedThroughput(1L,1L)
            )
    );

    locker = new DynamoDbLocker(
        new DynamoDB(dynamoDB.getDynamoDbClient()),tableName,Clock.systemUTC()
    );
}

void createTableAndWaitForActive(final CreateTableRequest request) throws BackendException {
    final String tableName = request.getTableName();
    Preconditions.checkArgument(!Strings.isNullOrEmpty(tableName),"Table name was null or empty");
    final TableDescription desc;
    try {
        desc = this.describeTable(tableName);
        if (null != desc && isTableAcceptingWrites(desc.getTableStatus())) {
            return; //store existed
        }
    } catch (BackendNotFoundException e) {
        log.debug(tableName + " did not exist yet,creating it",e);
    }

    createTable(request);
    waitForTableCreation(tableName,false /*verifyIndexesList*/,null /*expectedLsiList*/,null /*expectedGsiList*/);
}

@Override
public CreateTableRequest getTableSchema() {
    return super.getTableSchema()
        .withAttributeDeFinitions(
            new AttributeDeFinition()
                .withAttributeName(Constants.JANUSGRAPH_HASH_KEY)
                .withAttributeType(ScalarattributeType.S),new AttributeDeFinition()
                .withAttributeName(Constants.JANUSGRAPH_RANGE_KEY)
                .withAttributeType(ScalarattributeType.S))
        .withKeySchema(
            new KeySchemaElement()
                .withAttributeName(Constants.JANUSGRAPH_HASH_KEY)
                .withKeyType(KeyType.HASH),new KeySchemaElement()
                .withAttributeName(Constants.JANUSGRAPH_RANGE_KEY)
                .withKeyType(KeyType.RANGE));
}

/**
 * Used to create the Identity Table. This function only needs to be called
 * once.
 */
protected void createIdentityTable() throws DataAccessException {
    ProvisionedThroughput provisionedThroughput = new ProvisionedThroughput()
            .withReadCapacityUnits(10L)
            .withWriteCapacityUnits(5L);

    ArrayList<AttributeDeFinition> attributeDeFinitions = new ArrayList<AttributeDeFinition>();
    attributeDeFinitions
            .add(new AttributeDeFinition().withAttributeName(ATTRIBUTE_USERNAME).withAttributeType("S"));

    ArrayList<KeySchemaElement> tableKeySchema = new ArrayList<KeySchemaElement>();
    tableKeySchema.add(new KeySchemaElement().withAttributeName(ATTRIBUTE_USERNAME).withKeyType(KeyType.HASH));

    CreateTableRequest createTableRequest = new CreateTableRequest()
            .withTableName(USER_TABLE)
            .withProvisionedThroughput(provisionedThroughput)
            .withAttributeDeFinitions(attributeDeFinitions)
            .withKeySchema(tableKeySchema);

    try {
        ddb.createTable(createTableRequest);
    } catch (AmazonClientException e) {
        throw new DataAccessException("Failed to create table: " + USER_TABLE,e);
    }
}

/**
 * Used to create the device table. This function only needs to be called
 * once.
 */
protected void createDeviceTable() throws DataAccessException {
    ProvisionedThroughput provisionedThroughput = new ProvisionedThroughput()
            .withReadCapacityUnits(10L)
            .withWriteCapacityUnits(5L);

    ArrayList<AttributeDeFinition> attributeDeFinitions = new ArrayList<AttributeDeFinition>();
    attributeDeFinitions.add(new AttributeDeFinition().withAttributeName(
            ATTRIBUTE_UID).withAttributeType("S"));

    ArrayList<KeySchemaElement> tableKeySchema = new ArrayList<KeySchemaElement>();
    tableKeySchema.add(new KeySchemaElement().withAttributeName(ATTRIBUTE_UID)
            .withKeyType(KeyType.HASH));

    CreateTableRequest createTableRequest = new CreateTableRequest()
            .withTableName(DEVICE_TABLE)
            .withProvisionedThroughput(provisionedThroughput)
            .withAttributeDeFinitions(attributeDeFinitions)
            .withKeySchema(tableKeySchema);

    try {
        ddb.createTable(createTableRequest);
    } catch (AmazonClientException e) {
        throw new DataAccessException("Failed to create table: " + DEVICE_TABLE,e);
    }
}

public void createTable(Class<? extends IDomain> domain){
    CreateTableRequest tableRequest = dynamoDBMapper.generateCreateTableRequest(domain);
    tableRequest = tableRequest.withProvisionedThroughput(new ProvisionedThroughput(5L,5L));

    //check whether or not we need to add a provisioning throughput value for GSI
    for (Method method : domain.getmethods()) {
        if(method.isAnnotationPresent(DynamoDBIndexHashKey.class)){
            String tempGSI = method.getAnnotation(DynamoDBIndexHashKey.class).globalSecondaryIndexName();
            for (GlobalSecondaryIndex globalSecondaryIndex : tableRequest.getGlobalSecondaryIndexes()) {
                if(globalSecondaryIndex.getIndexName().equals(tempGSI)){
                    globalSecondaryIndex.setProvisionedThroughput(new ProvisionedThroughput(5L,5L));
                }
            }
        }
    }

    amazonDynamoDBClient.createTable(tableRequest);
}

public void createTable() {
  List<KeySchemaElement> keySchema = new ArrayList<>();

  keySchema.add(
      new KeySchemaElement()
          .withAttributeName(sequenceNumber.getAttributeName())
          .withKeyType(KeyType.HASH)
  );

  ProvisionedThroughput provisionedThroughput = new ProvisionedThroughput();
  provisionedThroughput.setReadCapacityUnits(10L);
  provisionedThroughput.setWriteCapacityUnits(10L);

  CreateTableRequest request = new CreateTableRequest()
      .withTableName("example_table")
      .withKeySchema(keySchema)
      .withAttributeDeFinitions(singleton(sequenceNumber))
      .withProvisionedThroughput(provisionedThroughput);

  client.createTable(request);
}

/**
 * Create a table with the given hashKey as row id
 * 
 * @param tableName
 * @param primaryKey
 */
public static void createTable(String tableName,String primaryKey) {
    ArrayList<KeySchemaElement> ks = new ArrayList<KeySchemaElement>();
    ArrayList<AttributeDeFinition> attributeDeFinitions = new ArrayList<AttributeDeFinition>();

    ks.add(new KeySchemaElement().withAttributeName(primaryKey)
            .withKeyType(KeyType.HASH));
    attributeDeFinitions.add(new AttributeDeFinition().withAttributeName(
            primaryKey).withAttributeType("S"));

    CreateTableRequest request = new CreateTableRequest()
            .withTableName(tableName).withKeySchema(ks)
            .withProvisionedThroughput(DEFAULT_PROVISIONED_THROUGHPUT);

    request.setAttributeDeFinitions(attributeDeFinitions);
    try {
        DynamoDbHandler.CLIENT.createTable(request);
    } catch (ResourceInUseException e) {
        //System.err.println("Table '" + tableName + "' already exists");
    }
}

/**
 * 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;
        }
    }
}

/**
 * Creates a table in AWS DynamoDB.
 * @param appid name of the {@link com.erudika.para.core.App}
 * @param readCapacity read capacity
 * @param writeCapacity write capacity
 * @return true if created
 */
public static boolean createTable(String appid,long readCapacity,long writeCapacity) {
    if (StringUtils.isBlank(appid)) {
        return false;
    } else if (StringUtils.containsWhitespace(appid)) {
        logger.warn("DynamoDB table name contains whitespace. The name '{}' is invalid.",appid);
        return false;
    } else if (existsTable(appid)) {
        logger.warn("DynamoDB table '{}' already exists.",appid);
        return false;
    }
    try {
        String table = getTableNameForappid(appid);
        getClient().createTable(new CreateTableRequest().withTableName(table).
                withKeySchema(new KeySchemaElement(Config._KEY,KeyType.HASH)).
                withAttributeDeFinitions(new AttributeDeFinition(Config._KEY,ScalarattributeType.S)).
                withProvisionedThroughput(new ProvisionedThroughput(readCapacity,writeCapacity)));
        logger.info("Created DynamoDB table '{}'.",table);
    } catch (Exception e) {
        logger.error(null,e);
        return false;
    }
    return true;
}

/***
 * Create the table and the associated indexes if it does not already exist
 * @param reflections
 * @param clazz
 */
private CreateTableResult createTable(Class<?> clazz) {

    final String tableName = this.getClassAnnotationValue(clazz,DynamoDBTable.class,String.class,"tableName");

    final Method hashKeyMember = this.getmethodForAnnotation(clazz,DynamoDBHashKey.class);
    final DynamoDBHashKey hashKeyAnno = hashKeyMember.getAnnotation(DynamoDBHashKey.class);
    final String hashKeyName = this.getAnnotationValue(hashKeyAnno,"attributeName",String.class);
    String rangeKeyName = null;


    final Method rangeKeyMember = this.getmethodForAnnotation(clazz,DynamoDBRangeKey.class);
    if(rangeKeyMember!=null){
        DynamoDBRangeKey rangeKeyAnno = rangeKeyMember.getAnnotation(DynamoDBRangeKey.class);   
        rangeKeyName = this.getAnnotationValue(rangeKeyAnno,String.class);
    }
    final Set<Method> hashKeyIndexFields = this.getmethodsAnnotatedWith(DynamoDBIndexHashKey.class,clazz);
    final Set<Method> rangeKeyIndexFields = this.getmethodsAnnotatedWith(DynamoDBIndexrangeKey.class,clazz);

    final Map<String,GlobalIndex> globalIndexes = this.createGlobalIndexes(hashKeyIndexFields,rangeKeyIndexFields,clazz);
    final Map<String,RangeKeyIndexField> localIndexes = this.createLocalIndexMap(rangeKeyIndexFields);

    final CreateTableRequest tableRequest = this.createCreateTableRequest(tableName,hashKeyName,rangeKeyName,globalIndexes,localIndexes);
    final CreateTableResult result = this.client.createTable(tableRequest);
    return result;
}

/**
 * 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);
  }
}

/**
 * Builds the necessary requests to create tables
 * 
 * @param tableName
 * @param keySchema
 * @param proThrou
 * @param attrs 
 * @return
 */
public static CreateTableRequest buildCreateTableRequest(String tableName,ProvisionedThroughput proThrou,String> attrs) {
  CreateTableRequest createTableRequest = new CreateTableRequest();
  createTableRequest.setTableName(tableName);
  createTableRequest.setKeySchema(keySchema);
  ArrayList<AttributeDeFinition> attributeDeFinitions = new ArrayList<AttributeDeFinition>();
  for (KeySchemaElement kEle : keySchema) {
    AttributeDeFinition attrDef = new AttributeDeFinition();
    attrDef.setAttributeName(kEle.getAttributeName());
    attrDef.setAttributeType(attrs.get(kEle.getAttributeName()));
    attributeDeFinitions.add(attrDef);
  }
  createTableRequest.setAttributeDeFinitions(attributeDeFinitions);
  createTableRequest.setProvisionedThroughput(proThrou);
  return createTableRequest;
}

@Override
public void createStoreIfAbsent(String storeName,boolean bBinaryValues) {
    String tableName = storetoTableName(storeName);
    if (!Tables.doesTableExist(m_ddbClient,tableName)) {
        // Create a table with a primary hash key named '_key',which holds a string
        m_logger.info("Creating table: {}",tableName);
        CreateTableRequest createTableRequest = new CreateTableRequest().withTableName(tableName)
            .withKeySchema(new KeySchemaElement()
                .withAttributeName(ROW_KEY_ATTR_NAME)
                .withKeyType(KeyType.HASH))
            .withAttributeDeFinitions(new AttributeDeFinition()
                .withAttributeName(ROW_KEY_ATTR_NAME)
                .withAttributeType(ScalarattributeType.S))
            .withProvisionedThroughput(new ProvisionedThroughput()
                .withReadCapacityUnits(READ_CAPACITY_UNITS)
                .withWriteCapacityUnits(WRITE_CAPACITY_UNITS));
        m_ddbClient.createTable(createTableRequest).getTableDescription();
        try {
            Tables.awaitTabletoBecomeActive(m_ddbClient,tableName);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);  
        }
    }
}

@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();
}

static void createExampleTable() {


    // Provide the initial provisioned throughput values as Java long data types
    ProvisionedThroughput provisionedThroughput = new ProvisionedThroughput()
        .withReadCapacityUnits(5L)
        .withWriteCapacityUnits(6L);
    CreateTableRequest request = new CreateTableRequest()
        .withTableName(tableName)
        .withProvisionedThroughput(provisionedThroughput);

    ArrayList<AttributeDeFinition> attributeDeFinitions= new ArrayList<AttributeDeFinition>();
    attributeDeFinitions.add(new AttributeDeFinition().withAttributeName("Id").withAttributeType("N"));
    request.setAttributeDeFinitions(attributeDeFinitions);

    ArrayList<KeySchemaElement> tableKeySchema = new ArrayList<KeySchemaElement>();
    tableKeySchema.add(new KeySchemaElement().withAttributeName("Id").withKeyType(KeyType.HASH));
    request.setKeySchema(tableKeySchema);

    client.createTable(request);

    waitForTabletoBecomeAvailable(tableName); 

    getTableinformation();

}

private static CreateTableResult createTable(AmazonDynamoDB ddb,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);
}

/**
 * Used to create the device table. This function only needs to be called
 * once.
 */
protected void createDeviceTable() throws DataAccessException {
    ProvisionedThroughput provisionedThroughput = new ProvisionedThroughput()
            .withReadCapacityUnits(10L)
            .withWriteCapacityUnits(5L);

    ArrayList<AttributeDeFinition> attributeDeFinitions = new ArrayList<AttributeDeFinition>();
    attributeDeFinitions.add(new AttributeDeFinition().withAttributeName(
            ATTRIBUTE_UID).withAttributeType("S"));

    ArrayList<KeySchemaElement> tableKeySchema = new ArrayList<KeySchemaElement>();
    tableKeySchema.add(new KeySchemaElement().withAttributeName(ATTRIBUTE_UID)
            .withKeyType(KeyType.HASH));

    CreateTableRequest createTableRequest = new CreateTableRequest()
            .withTableName(DEVICE_TABLE)
            .withProvisionedThroughput(provisionedThroughput)
            .withAttributeDeFinitions(attributeDeFinitions)
            .withKeySchema(tableKeySchema);

    try {
        ddb.createTable(createTableRequest);
    } catch (AmazonClientException e) {
        throw new DataAccessException("Failed to create table: " + DEVICE_TABLE,e);
    }
}

/**
 * Used to create the Identity Table. This function only needs to be called
 * once.
 */
protected void createIdentityTable() throws DataAccessException {
    ProvisionedThroughput provisionedThroughput = new ProvisionedThroughput()
            .withReadCapacityUnits(10L)
            .withWriteCapacityUnits(5L);

    ArrayList<AttributeDeFinition> attributeDeFinitions = new ArrayList<AttributeDeFinition>();
    attributeDeFinitions
            .add(new AttributeDeFinition().withAttributeName(ATTRIBUTE_USERNAME).withAttributeType("S"));

    ArrayList<KeySchemaElement> tableKeySchema = new ArrayList<KeySchemaElement>();
    tableKeySchema.add(new KeySchemaElement().withAttributeName(ATTRIBUTE_USERNAME).withKeyType(KeyType.HASH));

    CreateTableRequest createTableRequest = new CreateTableRequest()
            .withTableName(USER_TABLE)
            .withProvisionedThroughput(provisionedThroughput)
            .withAttributeDeFinitions(attributeDeFinitions)
            .withKeySchema(tableKeySchema);

    try {
        ddb.createTable(createTableRequest);
    } catch (AmazonClientException e) {
        throw new DataAccessException("Failed to create table: " + USER_TABLE,e);
    }
}

private void initializeHashAndRangeTable(String name,String hashName,String rangeName) {
    String tableName = quartzPrefix + name;

    if (!tableExists(tableName)) {
        log.info(String.format("Creating table '%s' with hash and range index.",tableName));
        CreateTableRequest request = new CreateTableRequest()
                .withTableName(tableName)
                .withKeySchema(
                        new KeySchemaElement().withKeyType(KeyType.HASH).withAttributeName(hashName),new KeySchemaElement().withKeyType(KeyType.RANGE).withAttributeName(rangeName))
                .withAttributeDeFinitions(
                        new AttributeDeFinition(hashName,ScalarattributeType.S),new AttributeDeFinition(rangeName,ScalarattributeType.S))
                .withProvisionedThroughput(new ProvisionedThroughput(1L,1L));

        client.createTable(request);

        waitForTable(tableName);
    } else {
        log.info(String.format("Table '%s' already exists.",tableName));
    }
}

private void initializeHashTable(String name,String hashName) {
    String tableName = quartzPrefix + name;

    if (!tableExists(tableName)) {
        log.info(String.format("Creating table '%s' with hash index.",tableName));
        CreateTableRequest request = new CreateTableRequest()
                .withTableName(tableName)
                .withKeySchema(
                        new KeySchemaElement().withKeyType(KeyType.HASH).withAttributeName(hashName))
                .withAttributeDeFinitions(
                        new AttributeDeFinition(hashName,tableName));
    }
}

public static void createSessionTable(AmazonDynamoDBClient dynamo,long readCapacityUnits,long writeCapacityUnits) {
    CreateTableRequest request = new CreateTableRequest().withTableName(tableName);

    request.withKeySchema(new KeySchemaElement().withAttributeName(DynamoSessionItem.SESSION_ID_ATTRIBUTE_NAME)
            .withKeyType(KeyType.HASH));

    request.withAttributeDeFinitions(
            new AttributeDeFinition().withAttributeName(DynamoSessionItem.SESSION_ID_ATTRIBUTE_NAME)
                    .withAttributeType(ScalarattributeType.S));

    request.setProvisionedThroughput(new ProvisionedThroughput().withReadCapacityUnits(readCapacityUnits)
            .withWriteCapacityUnits(writeCapacityUnits));

    dynamo.createTable(request);
}

private CreateTableResult createTable() {
    final List<AttributeDeFinition> attributeDeFinitions = new ArrayList<>();
    attributeDeFinitions.add(new AttributeDeFinition(RESOURCE_NAME_ATT,ScalarattributeType.S));
    attributeDeFinitions.add(new AttributeDeFinition(RDF_TRIPLE_ATT,ScalarattributeType.S));
    attributeDeFinitions.add(new AttributeDeFinition(RDF_PREDICATE_ATT,ScalarattributeType.S));
    attributeDeFinitions.add(new AttributeDeFinition(RDF_OBJECT_ATT,ScalarattributeType.S));

    final List<KeySchemaElement> keySchema = new ArrayList<>();
    keySchema.add(new KeySchemaElement(RESOURCE_NAME_ATT,KeyType.HASH));
    keySchema.add(new KeySchemaElement(RDF_TRIPLE_ATT,KeyType.RANGE));

    final ProvisionedThroughput provisionedthroughput =
        new ProvisionedThroughput(10L,10L);

    final LocalSecondaryIndex predicateIndex = new LocalSecondaryIndex()
        .withIndexName(PREDICATE_INDEX_NAME)
        .withKeySchema(new KeySchemaElement(RESOURCE_NAME_ATT,KeyType.HASH))
        .withKeySchema(new KeySchemaElement(RDF_PREDICATE_ATT,KeyType.RANGE))
        .withProjection(new Projection().withNonKeyAttributes(RDF_SUBJECT_ATT,RDF_OBJECT_ATT)
                                        .withProjectionType(ProjectionType.INCLUDE));

    final GlobalSecondaryIndex objectIndex = new GlobalSecondaryIndex()
        .withIndexName(OBJECT_INDEX_NAME)
        .withKeySchema(new KeySchemaElement(RDF_OBJECT_ATT,KeyType.RANGE))
        .withProjection(new Projection().withNonKeyAttributes(RDF_SUBJECT_ATT)
                                        .withProjectionType(ProjectionType.INCLUDE))
        .withProvisionedThroughput(new ProvisionedThroughput(10L,10L));

    final CreateTableRequest request =
        new CreateTableRequest()
            .withTableName(TABLE_NAME)
            .withAttributeDeFinitions(attributeDeFinitions)
            .withKeySchema(keySchema)
            .withProvisionedThroughput(provisionedthroughput)
            .withLocalSecondaryIndexes(predicateIndex)
            .withGlobalSecondaryIndexes(objectIndex);

    return dynamodbClient.createTable(request);
}

/**
 * Create ticket tables.
 *
 * @param deleteTables the delete tables
 */
public void createTicketTables(final boolean deleteTables) {
    final Collection<TicketDeFinition> Metadata = this.ticketCatalog.findAll();
    Metadata.forEach(Unchecked.consumer(r -> {
        final CreateTableRequest request = new CreateTableRequest()
                .withAttributeDeFinitions(new AttributeDeFinition(ColumnNames.ID.getName(),ScalarattributeType.S))
                .withKeySchema(new KeySchemaElement(ColumnNames.ID.getName(),KeyType.HASH))
                .withProvisionedThroughput(new ProvisionedThroughput(dynamoDbProperties.getReadCapacity(),dynamoDbProperties.getWriteCapacity()))
                .withTableName(r.getProperties().getStorageName());


        if (deleteTables) {
            final DeleteTableRequest delete = new DeleteTableRequest(r.getProperties().getStorageName());
            LOGGER.debug("Sending delete request [{}] to remove table if necessary",delete);
            Tableutils.deleteTableIfExists(amazonDynamoDBClient,delete);
        }
        LOGGER.debug("Sending delete request [{}] to create table",request);
        Tableutils.createTableIfNotExists(amazonDynamoDBClient,request);

        LOGGER.debug("Waiting until table [{}] becomes active...",request.getTableName());
        Tableutils.waitUntilActive(amazonDynamoDBClient,request.getTableName());

        final DescribeTableRequest describeTableRequest = new DescribeTableRequest().withTableName(request.getTableName());
        LOGGER.debug("Sending request [{}] to obtain table description...",describeTableRequest);

        final TableDescription tableDescription = amazonDynamoDBClient.describeTable(describeTableRequest).getTable();
        LOGGER.debug("Located newly created table with description: [{}]",tableDescription);
    }));
}

/**
 * Create tables.
 *
 * @param deleteTables the delete tables
 */
public void createServicesTable(final boolean deleteTables) {
    try {
        final CreateTableRequest request = new CreateTableRequest()
                .withAttributeDeFinitions(new AttributeDeFinition(ColumnNames.ID.getName(),dynamoDbProperties.getWriteCapacity()))
                .withTableName(TABLE_NAME);

        if (deleteTables) {
            final DeleteTableRequest delete = new DeleteTableRequest(request.getTableName());
            LOGGER.debug("Sending delete request [{}] to remove table if necessary",tableDescription);
    } catch (final Exception e) {
        throw Throwables.propagate(e);
    }
}

private static void createSettingsTable(final AmazonDynamoDBClient amazonDynamoDBClient,final boolean deleteTables) {
    try {
        final CreateTableRequest request = new CreateTableRequest()
                .withAttributeDeFinitions(new AttributeDeFinition(ColumnNames.ID.getName(),KeyType.HASH))
                .withProvisionedThroughput(new ProvisionedThroughput(PROVISIONED_THROUGHPUT,PROVISIONED_THROUGHPUT))
                .withTableName(TABLE_NAME);

        if (deleteTables) {
            final DeleteTableRequest delete = new DeleteTableRequest(request.getTableName());
            LOGGER.debug("Sending delete request [{}] to remove table if necessary",tableDescription);
    } catch (final Exception e) {
        throw Throwables.propagate(e);
    }
}

private void createAppsTable(String tableName) {
  final AttributeDeFinition appKey =
      new AttributeDeFinition().withAttributeName(HASH_KEY).withAttributeType(
          ScalarattributeType.S);

  final ArrayList<AttributeDeFinition>
      tableAttributeDeFinitions = Lists.newArrayList(appKey);

  final ArrayList<KeySchemaElement> tableKeySchema = Lists.newArrayList();
  tableKeySchema.add(
      new KeySchemaElement().withAttributeName(HASH_KEY).withKeyType(KeyType.HASH));

  final ProvisionedThroughput tableProvisionedThroughput =
      new ProvisionedThroughput()
          .withReadCapacityUnits(10L)
          .withWriteCapacityUnits(10L);

  final CreateTableRequest createTableRequest =
      new CreateTableRequest()
          .withTableName(tableName)
          .withKeySchema(tableKeySchema)
          .withAttributeDeFinitions(tableAttributeDeFinitions)
          .withProvisionedThroughput(tableProvisionedThroughput);

  final TableDescription tableDescription =
      amazonDynamoDB.createTable(createTableRequest).getTableDescription();

  logger.info("created_table {}",tableDescription);

  final DescribeTableRequest describeTableRequest =
      new DescribeTableRequest().withTableName(tableName);

  final TableDescription description =
      amazonDynamoDB.describeTable(describeTableRequest).getTable();

  logger.info("table_description: " + description);
}

private static Table createTable(final DynamoDBConnection dynamoDBConnection,final String tableName) throws InterruptedException
{
    final CreateTableRequest createTableRequest = createCreateTableRequest(tableName);

    final Table table = dynamoDBConnection.getDynamoDB().createTable(createTableRequest);

    table.waitForActive();
    return table;
}

private static CreateTableRequest createCreateTableRequest(final String tableName)
{
    final List<KeySchemaElement> keySchema = new ArrayList<>();
    final List<AttributeDeFinition> tableAttributes = new ArrayList<>();

    keySchema.add(new KeySchemaElement(FIELD_NAME_PRIMARY_ID,KeyType.HASH));
    tableAttributes.add(new AttributeDeFinition(FIELD_NAME_PRIMARY_ID,ScalarattributeType.S));


    return new CreateTableRequest()
            .withTableName(tableName)
            .withKeySchema(keySchema)
            .withAttributeDeFinitions(tableAttributes)
            .withProvisionedThroughput(new ProvisionedThroughput(1L,1L));
}

private CreateTableResult createPlanorequestsTable() {
    CreateTableRequest request = mapper.generateCreateTableRequest(DynamoDBPlanorequest.class)
            .withProvisionedThroughput(new ProvisionedThroughput()
                    .withReadCapacityUnits(5L)
                    .withWriteCapacityUnits(6L));

    CreateTableResult createTableResult = dynamoDB.createTable(request);

    return createTableResult;
}

private CreateTableResult createPlanorequestsTable() {
    CreateTableRequest request = sDynamoDBMapper.generateCreateTableRequest(DynamoDBPlanorequest.class)
            .withProvisionedThroughput(new ProvisionedThroughput()
                    .withReadCapacityUnits(5L)
                    .withWriteCapacityUnits(6L));

    CreateTableResult createTableResult = sDynamoDB.createTable(request);

    return createTableResult;
}

protected Table createHashTable() throws InterruptedException {
  // single hash PK
  ArrayList<AttributeDeFinition> attributeDeFinitions = new ArrayList<>();
  attributeDeFinitions.add(new AttributeDeFinition()
    .withAttributeName(PK).withAttributeType("S"));
  ArrayList<KeySchemaElement> keySchema = new ArrayList<>();
  keySchema.add(new KeySchemaElement().withAttributeName(PK).withKeyType(KeyType.HASH));

  CreateTableRequest request = new CreateTableRequest()
    .withKeySchema(keySchema)
    .withAttributeDeFinitions(attributeDeFinitions);
  return createTable(request);
}

@Test(expected = NotFoundException.class)
public void testLoadOrNotFound() {
    CreateTableRequest request = new CreateTableRequest();

    when(mapper.generateCreateTableRequest(Model.class)).thenReturn(request);
    when(dynamoDB.createTable(request)).thenReturn(table);

    ModelRepository repository = getService(ModelRepository.class);

    repository.loadOrNotFound("hash");
}