'Version 1.2.0 of the DynamoDB Streams Kinesis Adapter'

Author: amcp

META-INF/MANIFEST.MF

@@ -2,7 +2,7 @@ Manifest-Version: 1.0
 Bundle-ManifestVersion: 2
 Bundle-Name: Amazon DynamoDB Streams Kinesis Adapter for Java
 Bundle-SymbolicName: com.amazonaws.dynamodb.streams.kinesis.adapter;singleton:=true
-Bundle-Version: 1.1.1
+Bundle-Version: 1.2.0
 Bundle-Vendor: Amazon Technologies, Inc
 Bundle-RequiredExecutionEnvironment: JavaSE-1.7
 Export-Package: com.amazonaws.services.dynamodbv2.streamsadapter,

README.md

@@ -14,8 +14,8 @@
 * The KCL is designed to process streams from Amazon Kinesis, but by adding the DynamoDB Streams Kinesis Adapter, your application can process DynamoDB Streams instead, seamlessly and efficiently.
 
 ## Release Notes
-* Requires Kinesis Client Library version >= 1.6.4.
-* Requires AWS Java SDK version >= 1.11.14
+* Requires Kinesis Client Library version >= 1.7.5.
+* Requires AWS Java SDK version >= 1.11.115
 * It is highly recommended to [configure][kcl-configuration] Kinesis Client Library with `MaxRecords = 1000` and `IdleTimeInMillis = 500` to optimize DynamoDB Streams costs.
 
 ## Getting Started

pom.xml

@@ -6,7 +6,7 @@
     <artifactId>dynamodb-streams-kinesis-adapter</artifactId>
     <packaging>jar</packaging>
     <name>DynamoDB Streams Adapter for Java</name>
-    <version>1.1.1</version>
+    <version>1.2.0</version>
     <description>The DynamoDB Streams Adapter implements the AmazonKinesis interface so that your application can use KCL to consume and process data from a DynamoDB stream.</description>
     <url>https://aws.amazon.com/dynamodb</url>
 
@@ -23,8 +23,8 @@
     </licenses>
 
     <properties>
-        <aws-java-sdk.version>[1.11.14, 2.0.0)</aws-java-sdk.version>
-        <amazon-kinesis-client.version>[1.6.4, 1.7.0)</amazon-kinesis-client.version>
+        <aws-java-sdk.version>[1.11.115, 2.0.0)</aws-java-sdk.version>
+        <amazon-kinesis-client.version>[1.7.5, 1.8.0)</amazon-kinesis-client.version>
         <gpg.skip>true</gpg.skip>
     </properties>
 
@@ -112,6 +112,10 @@
             <name>Lei Ye</name>
             <email>leiye@amazon.com</email>
         </developer>
+        <developer>
+            <name>Alexander Patrikalakis</name>
+            <email>amcp@amazon.co.jp</email>
+        </developer>
     </developers>
 
     <build>

src/main/java/com/amazonaws/services/dynamodbv2/streamsadapter/AdapterRequestCache.java

@@ -47,10 +47,8 @@ public AdapterRequestCache(int capacity) {
     /**
      * Adds an entry to the cache.
      *
-     * @param request
-     *            Kinesis request
-     * @param requestAdapter
-     *            DynamoDB adapter client wrapper for the Kinesis request
+     * @param request        Kinesis request
+     * @param requestAdapter DynamoDB adapter client wrapper for the Kinesis request
      */
     public synchronized void addEntry(AmazonWebServiceRequest request, AmazonWebServiceRequest requestAdapter) {
         if (null == request || null == requestAdapter) {
@@ -67,8 +65,7 @@ public synchronized void addEntry(AmazonWebServiceRequest request, AmazonWebServ
     /**
      * Gets the actual DynamoDB Streams request made for a Kinesis request.
      *
-     * @param request
-     *            Kinesis request
+     * @param request Kinesis request
      * @return actual DynamoDB Streams request made for the associated Kinesis request
      */
     public synchronized AmazonWebServiceRequest getEntry(AmazonWebServiceRequest request) {

src/main/java/com/amazonaws/services/dynamodbv2/streamsadapter/AmazonDynamoDBStreamsAdapterClient.java

@@ -29,7 +29,7 @@
 /**
  * Main app that launches the worker that runs the multi-language record processor.
  * Requires a properties file containing configuration for this daemon and the KCL.
- *
+ * <p>
  * This version extends the KCL's MultiLangDaemon to use DynamoDB Streams instead of
  * Kinesis.
  */
@@ -39,7 +39,7 @@ public class StreamsMultiLangDaemon {
 
     /**
      * @param args Accepts a single argument, that argument is a properties file which provides KCL configuration as
-     *        well as the name of an executable.
+     *             well as the name of an executable.
      */
     public static void main(String[] args) {
 
@@ -55,17 +55,13 @@ public static void main(String[] args) {
             System.exit(1);
         }
 
-        ExecutorService executorService = config.getExecutorService();
-
-        Worker worker = new StreamsWorker(
-                config.getRecordProcessorFactory(),
-                config.getKinesisClientLibConfiguration(),
-                executorService);
+        final ExecutorService executorService = config.getExecutorService();
+        final Worker worker = StreamsWorkerFactory.createDynamoDbStreamsWorker(config.getRecordProcessorFactory(), config.getKinesisClientLibConfiguration(), executorService);
 
         // Daemon
-        MultiLangDaemon daemon = new MultiLangDaemon(worker);
+        final MultiLangDaemon daemon = new MultiLangDaemon(worker);
 
-        Future<Integer> future = executorService.submit(daemon);
+        final Future<Integer> future = executorService.submit(daemon);
         try {
             System.exit(future.get());
         } catch (InterruptedException | ExecutionException e) {

src/main/java/com/amazonaws/services/dynamodbv2/streamsadapter/StreamsMultiLangDaemon.java

@@ -17,10 +17,15 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import com.amazonaws.services.dynamodbv2.streamsadapter.model.RecordAdapter;
-import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor;
 import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
-import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownReason;
+import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
+import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
+import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
+import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownInput;
 import com.amazonaws.services.kinesis.model.Record;
 
 /**
@@ -30,17 +35,21 @@
  */
 public abstract class StreamsRecordProcessor implements IRecordProcessor {
 
+    private static final Log LOG = LogFactory.getLog(StreamsRecordProcessor.class);
+
     /**
      * {@inheritDoc}
      */
-    public abstract void initialize(String shardId);
+    public abstract void initialize(InitializationInput initializationInput);
 
-    public void processRecords(List<Record> records,
-            IRecordProcessorCheckpointer checkpointer) {
-        List<com.amazonaws.services.dynamodbv2.model.Record> streamsRecords =
-                new ArrayList<com.amazonaws.services.dynamodbv2.model.Record>();
-        for(Record record : records) {
-            if(record instanceof RecordAdapter) {
+    public void processRecords(ProcessRecordsInput processRecordsInput) {
+        final List<com.amazonaws.services.dynamodbv2.model.Record> streamsRecords = new ArrayList<com.amazonaws.services.dynamodbv2.model.Record>();
+        if (processRecordsInput.getRecords() == null) {
+            LOG.warn("ProcessRecordsInput's list of Records was null. Skipping.");
+            return;
+        }
+        for (Record record : processRecordsInput.getRecords()) {
+            if (record instanceof RecordAdapter) {
                 streamsRecords.add(((RecordAdapter) record).getInternalObject());
             } else {
                 // This record processor is not being used with the
@@ -49,7 +58,7 @@ public void processRecords(List<Record> records,
                 throw new IllegalArgumentException("Record is not an instance of RecordAdapter");
             }
         }
-        processStreamsRecords(streamsRecords, checkpointer);
+        processStreamsRecords(streamsRecords, processRecordsInput.getCheckpointer());
     }
 
     /**
@@ -58,16 +67,14 @@ public void processRecords(List<Record> records,
      * Upon fail over, the new instance will get records with sequence number &gt; checkpoint position
      * for each partition key.
      *
-     * @param records Data records to be processed
+     * @param records      Data records to be processed
      * @param checkpointer RecordProcessor should use this instance to checkpoint their progress.
      */
-    public abstract void processStreamsRecords(List<com.amazonaws.services.dynamodbv2.model.Record> records,
-            IRecordProcessorCheckpointer checkpointer);
+    public abstract void processStreamsRecords(List<com.amazonaws.services.dynamodbv2.model.Record> records, IRecordProcessorCheckpointer checkpointer);
 
     /**
      * {@inheritDoc}
      */
-    public abstract void shutdown(IRecordProcessorCheckpointer checkpointer,
-            ShutdownReason reason);
+    public abstract void shutdown(ShutdownInput shutdownInput);
 
 }