moved classes into different packages. upgraded SDK. tested table create request factory

Author: Alexander Patrikalakis

README.md

@@ -17,21 +17,31 @@ The CLI's usage follows with required parameters marked by asterisks.
       Use this flag to use strongly consistent scan. If the flag is not used 
       it will default to eventually consistent scan
       Default: false
-    --createDestination
-      Create destination table if it does not exist
-      Default: false
     --copyStreamSpecificationWhenCreating
       Use the source table stream specification for the destination table 
       during its creation.
       Default: false
+    --createAllGsi
+      Create all GSI in destination table
+      Default: false
+    --createAllLsi
+      Create all LSI in destination table
+      Default: false
+    --createDestination
+      Create destination table if it does not exist
+      Default: false
     --destinationEndpoint
       Endpoint of the destination table
-  * --destinationRegion
+  * --destinationSigningRegion
       Signing region for the destination endpoint
   * --destinationTable
       Name of the destination table
     --help
       Display usage information
+    --includeGsi
+      Include the following GSI in the destination table
+    --includeLsi
+      Include the following LSI in the destination table
     --maxWriteThreads
       Number of max threads to write to destination table
       Default: 1024
@@ -44,7 +54,7 @@ The CLI's usage follows with required parameters marked by asterisks.
       Default: 0
     --sourceEndpoint
       Endpoint of the source table
-  * --sourceRegion
+  * --sourceSigningRegion
       Signing region for the source endpoint
   * --sourceTable
       Name of the source table
@@ -87,8 +97,8 @@ To transfer to a different region, create two AmazonDynamoDBClients
 with different endpoints to pass into the DynamoDBBootstrapWorker and the DynamoDBConsumer.
 
 ```java
-import com.amazonaws.dynamodb.bootstrap.DynamoDBBootstrapWorker;
-import com.amazonaws.dynamodb.bootstrap.DynamoDBConsumer;
+import DynamoDBBootstrapWorker;
+import com.amazonaws.dynamodb.bootstrap.consumer.DynamoDBConsumer;
 import com.amazonaws.dynamodb.bootstrap.exception.NullReadCapacityException;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
@@ -98,28 +108,22 @@ import java.util.concurrent.Executors;
 
 class TransferDataFromOneTableToAnother {
     public static void main(String[] args) {
-        AmazonDynamoDB client = AmazonDynamoDBClientBuilder.standard()
+        final AmazonDynamoDB client = AmazonDynamoDBClientBuilder.standard()
                 .withRegion(com.amazonaws.regions.Regions.US_WEST_1).build();
-        DynamoDBBootstrapWorker worker = null;
         try {
             // 100.0 read operations per second. 4 threads to scan the table.
-            worker = new DynamoDBBootstrapWorker(client,
+            final DynamoDBBootstrapWorker worker = new DynamoDBBootstrapWorker(client,
                     100.0, "mySourceTable", 4);
+            // 50.0 write operations per second. 8 threads to scan the table.
+            final DynamoDBConsumer consumer = new DynamoDBConsumer(client, "myDestinationTable",
+                    50.0, Executors.newFixedThreadPool(8));
+            worker.pipe(consumer);
         } catch (NullReadCapacityException e) {
             System.err.println("The DynamoDB source table returned a null read capacity.");
             System.exit(1);
-        }
-        // 50.0 write operations per second. 8 threads to scan the table.
-        DynamoDBConsumer consumer = new DynamoDBConsumer(client, "myDestinationTable", 50.0,
-                Executors.newFixedThreadPool(8));
-        try {
-            worker.pipe(consumer);
-        } catch (ExecutionException e) {
+        } catch (ExecutionException | InterruptedException e) {
             System.err.println("Encountered exception when executing transfer: " + e.getMessage());
             System.exit(1);
-        } catch (InterruptedException e){
-            System.err.println("Interrupted when executing transfer: " + e.getMessage());
-            System.exit(1);
         }
     }
 }
@@ -128,13 +132,15 @@ class TransferDataFromOneTableToAnother {
 
 ### 2. Transfer Data From one DynamoDB Table to a Blocking Queue.
 
-The below example will read from a DynamoDB table and export to an array blocking queue. This is useful for when another application would like to consume
-the DynamoDB entries but does not have a setup application for it. They can just retrieve the queue (consumer.getQueue()) and then continually pop() from it
+The below example will read from a DynamoDB table and export to an array blocking queue.
+This is useful for when another application would like to consume
+the DynamoDB entries but does not have a setup application for it.
+They can just retrieve the queue (consumer.getQueue()) and then continually pop() from it
 to then process the new entries.
 
 ```java
-import com.amazonaws.dynamodb.bootstrap.BlockingQueueConsumer;
-import com.amazonaws.dynamodb.bootstrap.DynamoDBBootstrapWorker;
+import com.amazonaws.dynamodb.bootstrap.consumer.BlockingQueueConsumer;
+import DynamoDBBootstrapWorker;
 import com.amazonaws.dynamodb.bootstrap.exception.NullReadCapacityException;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
@@ -143,29 +149,20 @@ import java.util.concurrent.ExecutionException;
 
 class TransferDataFromOneTableToBlockingQueue {
     public static void main(String[] args) {
-        AmazonDynamoDB client = AmazonDynamoDBClientBuilder.standard()
+        final AmazonDynamoDB client = AmazonDynamoDBClientBuilder.standard()
                 .withRegion(com.amazonaws.regions.Regions.US_WEST_1).build();
-
-        DynamoDBBootstrapWorker worker = null;
-
         try {
             // 100.0 read operations per second. 4 threads to scan the table.
-            worker = new DynamoDBBootstrapWorker(client, 100.0, "mySourceTable", 4);
+            final DynamoDBBootstrapWorker worker = new DynamoDBBootstrapWorker(client, 100.0,
+                    "mySourceTable", 4);
+            final BlockingQueueConsumer consumer = new BlockingQueueConsumer(8);
+            worker.pipe(consumer);
         } catch (NullReadCapacityException e) {
             System.err.println("The DynamoDB source table returned a null read capacity.");
             System.exit(1);
-        }
-
-        BlockingQueueConsumer consumer = new BlockingQueueConsumer(8);
-
-        try {
-            worker.pipe(consumer);
-        } catch (ExecutionException e) {
+        } catch (ExecutionException | InterruptedException e) {
             System.err.println("Encountered exception when executing transfer: " + e.getMessage());
             System.exit(1);
-        } catch (InterruptedException e){
-            System.err.println("Interrupted when executing transfer: " + e.getMessage());
-            System.exit(1);
         }
     }
 }

pom.xml

@@ -11,7 +11,7 @@
         <url>https://github.com/awslabs/dynamodb-import-export-tool.git</url>
     </scm>
     <properties>
-        <jdk.version>1.7</jdk.version>
+        <jdk.version>1.8</jdk.version>
         <aws.java.sdk.version>1.11.123</aws.java.sdk.version>
         <powermock.version>1.6.6</powermock.version>
         <jcommander.version>1.69</jcommander.version>

src/main/java/com/amazonaws/dynamodb/bootstrap/AbstractLogProvider.java

@@ -14,14 +14,14 @@
  */
 package com.amazonaws.dynamodb.bootstrap;
 
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
-
+import com.amazonaws.dynamodb.bootstrap.constants.BootstrapConstants;
+import com.amazonaws.dynamodb.bootstrap.consumer.AbstractLogConsumer;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
-import com.amazonaws.dynamodb.bootstrap.constants.BootstrapConstants;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Abstract class to send inputs from a source to a consumer.

src/main/java/com/amazonaws/dynamodb/bootstrap/CommandLineArgs.java

@@ -18,6 +18,8 @@
 import com.beust.jcommander.Parameter;
 import lombok.Getter;
 
+import java.util.List;
+
 /**
  * This class contains the parameters to input when executing the program from
  * command line.
@@ -60,6 +62,22 @@ public boolean getHelp() {
     @Parameter(names = CREATE_DESTINATION_TABLE_IF_MISSING, description = "Create destination table if it does not exist")
     private boolean createDestinationTableIfMissing;
 
+    public static final String CREATE_ALL_LSI = "--createAllLsi";
+    @Parameter(names = CREATE_ALL_LSI, description = "Create all LSI in destination table")
+    private boolean createAllLsi;
+
+    public static final String CREATE_ALL_GSI = "--createAllGsi";
+    @Parameter(names = CREATE_ALL_GSI, description = "Create all GSI in destination table")
+    private boolean createAllGsi;
+
+    public static final String INCLUDE_LSI = "--includeLsi";
+    @Parameter(names = INCLUDE_LSI, description = "Include the following LSI in the destination table")
+    private List<String> includeLsi;
+
+    public static final String INCLUDE_GSI = "--includeGsi";
+    @Parameter(names = INCLUDE_GSI, description = "Include the following GSI in the destination table")
+    private List<String> includeGsi;
+
     public static final String COPY_STREAM_SPECIFICATION_WHEN_CREATING = "--copyStreamSpecificationWhenCreating";
     @Parameter(names = COPY_STREAM_SPECIFICATION_WHEN_CREATING, description = "Use the source table stream specification for the destination table during its creation.")
     private boolean copyStreamSpecification;

src/main/java/com/amazonaws/dynamodb/bootstrap/CommandLineInterface.java

@@ -14,35 +14,40 @@
  */
 package com.amazonaws.dynamodb.bootstrap;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
 import com.amazonaws.AmazonServiceException;
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
 import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.dynamodb.bootstrap.constants.BootstrapConstants;
+import com.amazonaws.dynamodb.bootstrap.consumer.DynamoDBConsumer;
+import com.amazonaws.dynamodb.bootstrap.exception.NullReadCapacityException;
+import com.amazonaws.dynamodb.bootstrap.exception.SectionOutOfRangeException;
+import com.amazonaws.dynamodb.bootstrap.worker.DynamoDBBootstrapWorker;
 import com.amazonaws.regions.Region;
 import com.amazonaws.regions.Regions;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
-import com.amazonaws.services.dynamodbv2.model.*;
+import com.amazonaws.services.dynamodbv2.model.DescribeTableRequest;
+import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
+import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughputDescription;
+import com.amazonaws.services.dynamodbv2.model.ResourceNotFoundException;
+import com.amazonaws.services.dynamodbv2.model.TableDescription;
 import com.amazonaws.waiters.WaiterParameters;
 import com.amazonaws.waiters.WaiterTimedOutException;
 import com.amazonaws.waiters.WaiterUnrecoverableException;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Optional;
-import lombok.extern.log4j.Log4j;
-
-import com.amazonaws.ClientConfiguration;
-import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
-import com.amazonaws.dynamodb.bootstrap.constants.BootstrapConstants;
-import com.amazonaws.dynamodb.bootstrap.exception.NullReadCapacityException;
-import com.amazonaws.dynamodb.bootstrap.exception.SectionOutOfRangeException;
 import com.beust.jcommander.JCommander;
 import com.beust.jcommander.ParameterException;
+import com.google.common.base.Preconditions;
+import lombok.extern.log4j.Log4j;
+
+import java.util.Optional;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * The interface that parses the arguments, and begins to transfer data from one
@@ -54,7 +59,7 @@ public class CommandLineInterface {
     public static final String ENCOUNTERED_EXCEPTION_WHEN_EXECUTING_TRANSFER = "Encountered exception when executing transfer.";
 
     static AwsClientBuilder.EndpointConfiguration createEndpointConfiguration(Region region, Optional<String> endpoint, String endpointPrefix) {
-        return new AwsClientBuilder.EndpointConfiguration(endpoint.or("https://" + region.getServiceEndpoint(endpointPrefix)), region.getName());
+        return new AwsClientBuilder.EndpointConfiguration(endpoint.orElse("https://" + region.getServiceEndpoint(endpointPrefix)), region.getName());
     }
 
     private final AwsClientBuilder.EndpointConfiguration sourceEndpointConfiguration;
@@ -66,72 +71,37 @@ static AwsClientBuilder.EndpointConfiguration createEndpointConfiguration(Region
     private final int maxWriteThreads;
     private final boolean isConsistentScan;
     private final boolean createDestinationTableIfMissing;
+    private final boolean createAllGsi;
+    private final boolean createAllLsi;
+    private final SortedSet<String> includeGsi;
+    private final SortedSet<String> includeLsi;
     private final boolean copyStreamSpecification;
     private final int sectionNumber;
     private final int totalSections;
 
     private CommandLineInterface(final CommandLineArgs params) {
         sourceEndpointConfiguration = createEndpointConfiguration(Region.getRegion(Regions.fromName(params.getSourceSigningRegion())),
-                Optional.fromNullable(params.getSourceEndpoint()), AmazonDynamoDB.ENDPOINT_PREFIX);
+                Optional.ofNullable(params.getSourceEndpoint()), AmazonDynamoDB.ENDPOINT_PREFIX);
         sourceTable = params.getSourceTable();
         destinationEndpointConfiguration = createEndpointConfiguration(Region.getRegion(Regions.fromName(params.getDestinationSigningRegion())),
-                Optional.fromNullable(params.getDestinationEndpoint()), AmazonDynamoDB.ENDPOINT_PREFIX);
+                Optional.ofNullable(params.getDestinationEndpoint()), AmazonDynamoDB.ENDPOINT_PREFIX);
         destinationTable = params.getDestinationTable();
         readThroughputRatio = params.getReadThroughputRatio();
         writeThroughputRatio = params.getWriteThroughputRatio();
         maxWriteThreads = params.getMaxWriteThreads();
         isConsistentScan = params.isConsistentScan();
         createDestinationTableIfMissing = params.isCreateDestinationTableIfMissing();
+        createAllGsi = params.isCreateAllGsi();
+        createAllLsi = params.isCreateAllLsi();
+        includeLsi = new TreeSet<>(params.getIncludeLsi());
+        Preconditions.checkArgument(includeLsi.size() == params.getIncludeLsi().size(), "list of LSI names must be unique");
+        includeGsi = new TreeSet<>(params.getIncludeGsi());
+        Preconditions.checkArgument(includeGsi.size() == params.getIncludeGsi().size(), "list of GSI names must be unique");
         copyStreamSpecification = params.isCopyStreamSpecification();
         sectionNumber = params.getSection();
         totalSections = params.getTotalSections();
     }
 
-    static List<GlobalSecondaryIndex> convertGlobalSecondaryIndexDescriptions(List<GlobalSecondaryIndexDescription> list) {
-        final List<GlobalSecondaryIndex> result = new ArrayList<>(list.size());
-        for (GlobalSecondaryIndexDescription description : list) {
-            result.add(new GlobalSecondaryIndex()
-                    .withIndexName(description.getIndexName())
-                    .withKeySchema(description.getKeySchema())
-                    .withProjection(description.getProjection())
-                    .withProvisionedThroughput(getProvisionedThroughputFromDescription(description.getProvisionedThroughput())));
-        }
-        return result;
-    }
-
-    static List<LocalSecondaryIndex> convertLocalSecondaryIndexDescriptions(List<LocalSecondaryIndexDescription> list) {
-        final List<LocalSecondaryIndex> result = new ArrayList<>(list.size());
-        for (LocalSecondaryIndexDescription description : list) {
-            result.add(new LocalSecondaryIndex()
-                    .withIndexName(description.getIndexName())
-                    .withKeySchema(description.getKeySchema())
-                    .withProjection(description.getProjection()));
-        }
-        return result;
-    }
-
-    @VisibleForTesting
-    static CreateTableRequest convertTableDescriptionToCreateTableRequest(TableDescription description,
-                                                                          String newTableName,
-                                                                          boolean copyStreamSpecification) {
-        List<GlobalSecondaryIndexDescription> gsiDesc = description.getGlobalSecondaryIndexes();
-        List<GlobalSecondaryIndex> gsi = gsiDesc == null ? null : convertGlobalSecondaryIndexDescriptions(gsiDesc);
-        List<LocalSecondaryIndexDescription> lsiDesc = description.getLocalSecondaryIndexes();
-        List<LocalSecondaryIndex> lsi = lsiDesc == null ? null : convertLocalSecondaryIndexDescriptions(lsiDesc);
-        ProvisionedThroughput pt = getProvisionedThroughputFromDescription(description.getProvisionedThroughput());
-        CreateTableRequest ctr = new CreateTableRequest()
-                .withTableName(newTableName)
-                .withProvisionedThroughput(pt)
-                .withAttributeDefinitions(description.getAttributeDefinitions())
-                .withKeySchema(description.getKeySchema())
-                .withGlobalSecondaryIndexes(gsi)
-                .withLocalSecondaryIndexes(lsi);
-        if (copyStreamSpecification) {
-            ctr.withStreamSpecification(description.getStreamSpecification());
-        }
-        return ctr;
-    }
-
     private static ProvisionedThroughput getProvisionedThroughputFromDescription(ProvisionedThroughputDescription description) {
         return new ProvisionedThroughput(description.getReadCapacityUnits(), description.getWriteCapacityUnits());
     }
@@ -159,8 +129,16 @@ private void bootstrapTable() throws InterruptedException, ExecutionException, S
                 throw new IllegalArgumentException("Destination table " + destinationTable + " did not exist", e);
             }
             try {
-                destinationClient.createTable(convertTableDescriptionToCreateTableRequest(readTableDescription,
-                        destinationTable, copyStreamSpecification));
+                final TableDescriptionToCreateTableRequestConverter converter =
+                        TableDescriptionToCreateTableRequestConverter.builder()
+                        .copyStreamSpecification(copyStreamSpecification)
+                        .newTableName(destinationTable)
+                        .createAllGsi(createAllGsi)
+                        .gsiToInclude(includeGsi)
+                        .createAllLsi(createAllLsi)
+                        .lsiToInclude(includeLsi)
+                        .build();
+                destinationClient.createTable(converter.apply(readTableDescription));
                 destinationClient.waiters().tableExists().run(new WaiterParameters<>(new DescribeTableRequest(destinationTable)));
             } catch(WaiterUnrecoverableException | WaiterTimedOutException | AmazonServiceException ase) {
                 throw new IllegalArgumentException("Unable to create destination table", ase);

src/main/java/com/amazonaws/dynamodb/bootstrap/DynamoDBTableScan.java

@@ -16,6 +16,7 @@
 
 import java.util.concurrent.Executor;
 
+import com.amazonaws.dynamodb.bootstrap.worker.ScanSegmentWorker;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
 import com.amazonaws.services.dynamodbv2.model.ScanRequest;
 import com.google.common.util.concurrent.RateLimiter;

src/main/java/com/amazonaws/dynamodb/bootstrap/ParallelScanExecutor.java

@@ -14,6 +14,8 @@
  */
 package com.amazonaws.dynamodb.bootstrap;
 
+import com.amazonaws.dynamodb.bootstrap.worker.ScanSegmentWorker;
+
 import java.util.BitSet;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorCompletionService;