'Version 1.5.1 of the DynamoDB Streams Kinesis Adapter'

Author: aggarwal

.github/PULL_REQUEST_TEMPLATE.md

@@ -1,2 +1,2 @@
 DynamoDB Streams Kinesis Adapter for Java
-Copyright 2014-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+Copyright 2014-2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.

CODE_OF_CONDUCT.md

@@ -14,7 +14,13 @@
 * 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
-### Latest Release (v1.5.x)
+### Latest Release (v1.5.1)
+* Restores compile compatibility with KCL 1.13.3.
+* Fixes a performance issue that arised when using v1.5.0 with KCL 1.12 through 1.13.2.
+* Fixes a defect where `MaxLeasesForWorker` configuration was not being propagated to `StreamsLeaseTaker`.
+* Finished (SHARD_END) leases will now only be delete after at least 6 hours have passed since the shard was created. This further reduces the chances of lineage replay.
+
+### Release (v1.5.0)
 * Introduces the implementation of periodic shard sync in conjunction with Amazon Kinesis Client Library v1.11.x (KCL). The default shard sync strategy is to discover new/child shards only when a consumer completes processing a shard. This default strategy constrains horizontal scaling of customer applications when consuming tables with  10,000+ partitions due to increased DescribeStream calls. Periodic shard sync guarantees that only a subset of the fleet (by default 10) will perform shard syncs, and decouples DescribeStream call volume from growth in fleet size.
 
 * Improves inconsistency handling in DescribeStream result aggregation by fixing any parent-open-child-open cases. This ensures that shard sync does not fail due to an assertion failure in KCL on this type of inconsistency.
@@ -24,8 +30,9 @@
 * Introduces `StreamsLeaseTaker` with improved load-balancing of leases among workers.
   * SHARD_END and non-SHARD_END check-pointed leases are balanced independently.
   * Leases are now stolen evenly from other workers instead of from only the most loaded worker. `MaxLeasesToStealAtOneTime` no longer needs to be specified by users. It is now determined automatically based on the number of leases held by the worker. The user-specified value for this is no longer used.
- 
+
 * Users should continue using factory methods from `StreamsWorkerFactory` to create KCL Worker as specified in the guidance of Release v1.4.x.
+* We strongly recommended that you create only one worker per host in your processing fleet to get optimal performance from DynamoDB Streams service.
 
 ### Release (v1.4.x)
 * This release fixes an issue of high propagation delay of streams records when processing streams on small tables. This issue occurs when KCL ShardSyncer is not discovering new shards due to server side delays in shard creation or in reporting new shard creation to internal services. The code is implemented in a new implementation of IKinesisProxy interface called DynamoDBStreamsProxy which is part of the latest release.
@@ -60,7 +67,7 @@ Add the following to your Maven pom file:
 <dependency>
     <groupId>com.amazonaws</groupId>
     <artifactId>dynamodb-streams-kinesis-adapter</artifactId>
-    <version>1.5.0</version>
+    <version>1.5.1</version>
 </dependency>
 ```
 

CONTRIBUTING.md

@@ -6,7 +6,7 @@
     <artifactId>dynamodb-streams-kinesis-adapter</artifactId>
     <packaging>jar</packaging>
     <name>DynamoDB Streams Adapter for Java</name>
-    <version>1.5.0</version>
+    <version>1.5.1</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,10 +23,10 @@
     </licenses>
 
     <properties>
-        <aws-java-sdk.version>1.11.603</aws-java-sdk.version>
-        <amazon-kinesis-client.version>[1.11.0,)</amazon-kinesis-client.version>
+        <aws-java-sdk.version>1.11.728</aws-java-sdk.version>
+        <amazon-kinesis-client.version>1.13.3</amazon-kinesis-client.version>
         <powermock.version>1.6.2</powermock.version>
-        <aws.dynamodblocal.version>1.11.86</aws.dynamodblocal.version>
+        <aws.dynamodblocal.version>[1.12,2.0)</aws.dynamodblocal.version>
         <maven.dependency.version>3.0.0</maven.dependency.version>
         <sqlite4java.version>1.0.392</sqlite4java.version>
         <gpg.skip>true</gpg.skip>
@@ -62,7 +62,7 @@
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.9.10.3</version>
+            <version>2.6.6</version>
         </dependency>
 
         <dependency>
@@ -114,6 +114,26 @@
     </dependencies>
 
     <developers>
+        <developer>
+            <name>Parijat Sinha</name>
+            <email>parijas@amazon.com</email>
+        </developer>
+        <developer>
+            <name>Shitanshu Aggarwal</name>
+            <email>shitansh@amazon.com</email>
+        </developer>
+        <developer>
+            <name>Abhishek Khanna</name>
+            <email>akkhanna@amazon.com</email>
+        </developer>
+        <developer>
+            <name>Debjyoti Roy</name>
+            <email>debjyotr@amazon.com</email>
+        </developer>
+        <developer>
+            <name>Shantanu Bhoyar</name>
+            <email>shbhoyar@amazon.com</email>
+        </developer>
         <developer>
             <name>Holger Jansohn</name>
             <email>holgs@live.ca</email>

NOTICE.txt

@@ -11,6 +11,7 @@
 import com.amazonaws.services.dynamodbv2.streamsadapter.utils.ThreadSleeper;
 import com.amazonaws.services.kinesis.AmazonKinesis;
 import com.amazonaws.services.kinesis.clientlibrary.proxies.IKinesisProxyExtended;
+import com.amazonaws.services.kinesis.clientlibrary.proxies.ShardClosureVerificationResponse;
 
 import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
 import com.amazonaws.services.kinesis.model.DescribeStreamResult;
@@ -258,6 +259,18 @@ public synchronized List<Shard> getShardList() {
         return listOfShardsSinceLastGet.get();
     }
 
+    @Override
+    /**
+     * This method gets invoked from ShutdownTask when the shard consumer is shutting down.
+     * Kinesis modified KCL to verify that the shard being closed has children, and this requires listing all shards.
+     * Since DynamoDB streams can have a large number of shards, this verification delays shard closure, and can severely
+     * degrade processing performance. For large streams, this can even cause stream processing to completely halt.
+     * Therefore, we skip performing this validation in ShutdownTask by simply returning true.
+     */
+    public ShardClosureVerificationResponse verifyShardClosure(String shardId) {
+        return () -> true; // isShardClosed -> true
+    }
+
     private ShardGraphProcessingResult buildShardGraphSnapshot() {
 
         DescribeStreamResult response;

README.md

@@ -7,6 +7,8 @@
 
 import java.io.Serializable;
 import java.math.BigInteger;
+import java.time.Duration;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -48,6 +50,15 @@
 public class DynamoDBStreamsShardSyncer implements ShardSyncer {
 
     private static final Log LOG = LogFactory.getLog(DynamoDBStreamsShardSyncer.class);
+    private static final String SHARD_ID_SEPARATOR = "-";
+
+    /* This retention period mostly will protect race conditions that are triggered by shards getting sealed
+     * immediately after creation. Average active lifetime of a shard is around 4 hours today, so setting retention to
+     * slightly higher helps us retain the leases for shards with active lifetime close to average for investigations
+     * and visibility. Lower values on the order of minutes may also work, but reduce operational auditability.
+     */
+    private static final Duration MIN_LEASE_RETENTION = Duration.ofHours(6);
+
     private final LeaseCleanupValidator leaseCleanupValidator;
 
     public DynamoDBStreamsShardSyncer(final LeaseCleanupValidator leaseCleanupValidator) {
@@ -714,11 +725,20 @@ synchronized void cleanupLeaseForClosedShard(String closedShardId,
             boolean okayToDelete = true;
             for (KinesisClientLease lease : childShardLeases) {
                 if (!lease.getCheckpoint().equals(ExtendedSequenceNumber.SHARD_END)) {
-                    okayToDelete = false;
+                    okayToDelete = false; // if any child is still being processed, don't delete lease for parent
                     break;
                 }
             }
 
+            try {
+                if (Instant.now().isBefore(getShardCreationTime(closedShardId).plus(MIN_LEASE_RETENTION))) {
+                    okayToDelete = false; // if parent was created within lease retention period, don't delete lease for parent
+                }
+            } catch (RuntimeException e) {
+                LOG.info("Could not extract creation time from ShardId [" + closedShardId +"]");
+                LOG.debug(e);
+            }
+
             if (okayToDelete) {
                 LOG.info("Deleting lease for shard " + leaseForClosedShard.getLeaseKey()
                     + " as it is eligible for cleanup - its child shard is check-pointed at SHARD_END.");
@@ -727,6 +747,16 @@ synchronized void cleanupLeaseForClosedShard(String closedShardId,
         }
     }
 
+    /**
+     * This method extracts the shard creation time from the ShardId
+     *
+     * @param shardId
+     * @return instant at which the shard was created
+     */
+    private Instant getShardCreationTime(String shardId) {
+        return Instant.ofEpochMilli(Long.parseLong(shardId.split(SHARD_ID_SEPARATOR)[1]));
+    }
+
     /**
      * Helper method to create a new KinesisClientLease POJO for a shard.
      * Note: Package level access only for testing purposes

pom.xml

@@ -61,7 +61,8 @@ public static Worker createDynamoDbStreamsWorker(IRecordProcessorFactory recordP
             .shardSyncer(new DynamoDBStreamsShardSyncer(new StreamsLeaseCleanupValidator()))
             .shardPrioritization(config.getShardPrioritizationStrategy())
             .leaseManager(kinesisClientLeaseManager)
-            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis()))
+            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis())
+                    .maxLeasesForWorker(config.getMaxLeasesForWorker()))
             .leaderDecider(new StreamsDeterministicShuffleShardSyncLeaderDecider(config, kinesisClientLeaseManager))
             .build();
     }
@@ -88,7 +89,8 @@ public static Worker createDynamoDbStreamsWorker(IRecordProcessorFactory recordP
             .shardSyncer(new DynamoDBStreamsShardSyncer(new StreamsLeaseCleanupValidator()))
             .shardPrioritization(config.getShardPrioritizationStrategy())
             .leaseManager(kinesisClientLeaseManager)
-            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis()))
+            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis())
+                    .maxLeasesForWorker(config.getMaxLeasesForWorker()))
             .leaderDecider(new StreamsDeterministicShuffleShardSyncLeaderDecider(config, kinesisClientLeaseManager))
             .build();
     }
@@ -118,7 +120,8 @@ public static Worker createDynamoDbStreamsWorker(IRecordProcessorFactory recordP
             .shardSyncer(new DynamoDBStreamsShardSyncer(new StreamsLeaseCleanupValidator()))
             .shardPrioritization(config.getShardPrioritizationStrategy())
             .leaseManager(kinesisClientLeaseManager)
-            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis()))
+            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis())
+                    .maxLeasesForWorker(config.getMaxLeasesForWorker()))
             .leaderDecider(new StreamsDeterministicShuffleShardSyncLeaderDecider(config, kinesisClientLeaseManager))
             .build();
     }
@@ -148,7 +151,8 @@ public static Worker createDynamoDbStreamsWorker(IRecordProcessorFactory recordP
             .shardSyncer(new DynamoDBStreamsShardSyncer(new StreamsLeaseCleanupValidator()))
             .shardPrioritization(config.getShardPrioritizationStrategy())
             .leaseManager(kinesisClientLeaseManager)
-            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis()))
+            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis())
+                    .maxLeasesForWorker(config.getMaxLeasesForWorker()))
             .leaderDecider(new StreamsDeterministicShuffleShardSyncLeaderDecider(config, kinesisClientLeaseManager))
             .build();
     }
@@ -175,7 +179,8 @@ public static Worker createDynamoDbStreamsWorker(IRecordProcessorFactory recordP
             .shardSyncer(new DynamoDBStreamsShardSyncer(new StreamsLeaseCleanupValidator()))
             .shardPrioritization(config.getShardPrioritizationStrategy())
             .leaseManager(kinesisClientLeaseManager)
-            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis()))
+            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis())
+                    .maxLeasesForWorker(config.getMaxLeasesForWorker()))
             .leaderDecider(new StreamsDeterministicShuffleShardSyncLeaderDecider(config, kinesisClientLeaseManager))
             .build();
     }
@@ -205,7 +210,8 @@ public static Worker createDynamoDbStreamsWorker(IRecordProcessorFactory recordP
             .shardSyncer(new DynamoDBStreamsShardSyncer(new StreamsLeaseCleanupValidator()))
             .shardPrioritization(config.getShardPrioritizationStrategy())
             .leaseManager(kinesisClientLeaseManager)
-            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis()))
+            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis())
+                    .maxLeasesForWorker(config.getMaxLeasesForWorker()))
             .leaderDecider(new StreamsDeterministicShuffleShardSyncLeaderDecider(config, kinesisClientLeaseManager))
             .build();
     }
@@ -235,7 +241,8 @@ public static Worker createDynamoDbStreamsWorker(IRecordProcessorFactory recordP
             .shardSyncer(new DynamoDBStreamsShardSyncer(new StreamsLeaseCleanupValidator()))
             .shardPrioritization(config.getShardPrioritizationStrategy())
             .leaseManager(kinesisClientLeaseManager)
-            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis()))
+            .leaseTaker(new StreamsLeaseTaker<>(kinesisClientLeaseManager, config.getWorkerIdentifier(), config.getFailoverTimeMillis())
+                    .maxLeasesForWorker(config.getMaxLeasesForWorker()))
             .leaderDecider(new StreamsDeterministicShuffleShardSyncLeaderDecider(config, kinesisClientLeaseManager))
             .build();
     }