Feature/publish old image

README.md

@@ -21,8 +21,8 @@ In our implementation we opted to use Amazon Kinesis Client with DynamoDB Stream
 * Java 8
 * Gradlew 5.3.1
 * Kafka Connect Framework >= 2.1.1
-* Amazon Kinesis Client 1.9.1
-* DynamoDB Streams Kinesis Adapter 1.5.2
+* Amazon Kinesis Client 1.13.1
+* DynamoDB Streams Kinesis Adapter 1.5.3
 
 ## Documentation
 * [Getting started](docs/getting-started.md)
@@ -44,6 +44,7 @@ In our implementation we opted to use Amazon Kinesis Client with DynamoDB Stream
   * However you will only encounter this issue by running lots of tasks on one machine with really high load.
 
 * Synced(Source) DynamoDB table unit capacity must be large enough to ensure `INIT_SYNC` to be finished in around 16 hours. Otherwise there is a risk `INIT_SYNC` being restarted just as soon as it's finished because DynamoDB Streams store change events only for 24 hours.
+  * `INIT_SYNC` can be skipped with `init.sync.skip=true` configuration
 
 * Required AWS roles:
 ```json

build.gradle

@@ -20,6 +20,8 @@ task userWrapper(type: Wrapper) {
 
 description = "kafka-connect-dynamodb"
 
+shadowJar.archiveName = "${project.name}.jar"
+
 allprojects {
     group "com.trustpilot.kafka.connect.dynamodb"
     version = gitVersion()

docs/details.md

@@ -13,6 +13,8 @@ This connector can sync multiple DynamoDB tables at the same time and it does so
 
 `INIT_SYNC` is a process when all existing table data is scanned and pushed into Kafka destination topic. Usually this happens only once after source task for specific table is started for the first time. But it can be repeated in case of unexpected issues, e.g. if source connector was down for long period of time and it is possible that it has missed some of the change events from the table stream (DynamoDB streams store data for 24 hours only). 
 
+Using `init.sync.skip` will skip this process and the connector will only ever read from the LATEST position in the stream.
+
 ### 3. "SYNC"
 
 Once `INIT_SYNC` is finished source task switches into DynamoDB Streams consumer state. There all changes that happen to the source table are represented in this stream and copied over to the Kafka's destination topic. Consumers of this topic can recreate full state of the source table at any given time.

docs/options.md

@@ -20,6 +20,7 @@
     "aws.region": "eu-west-1",
     "aws.access.key.id": "",
     "aws.secret.key": "",
+    "aws.assume.role.arn": "",
 
     "dynamodb.table.env.tag.key": "environment",
     "dynamodb.table.env.tag.value": "dev",
@@ -35,9 +36,12 @@
     "tasks.max": "1",
 
     "init.sync.delay.period": 60,
+    "init.sync.skip": false,
     "connect.dynamodb.rediscovery.period": "60000"
 }
 ```
+`aws.assume.role.arn` - ARN identifier of an IAM role that the KCL and Dynamo Clients can assume for cross account access
+
 `dynamodb.table.env.tag.key` - tag key used to define environment. Useful if you have `staging` and `production` under same AWS account. Or if you want to use different Kafka Connect clusters to sync different tables.
 
 `dynamodb.table.env.tag.value` - defines from which environment to ingest tables. For e.g. 'staging' or 'production'...
@@ -50,6 +54,8 @@
 
 `init.sync.delay.period` - time interval in seconds. Defines how long `INIT_SYNC` should delay execution before starting. This is used to give time for Kafka Connect tasks to calm down after rebalance (Since multiple tasks rebalances can happen in quick succession and this would mean more duplicated data since `INIT_SYNC` process won't have time mark it's progress). 
 
+`init.sync.skip` - boolean to determine whether to start the connector reading the entire table or from the latest offset.
+
 `connect.dynamodb.rediscovery.period` - time interval in milliseconds. Defines how often connector should try to find new DynamoDB tables (or detect removed ones). If changes are found tasks are automatically reconfigured.
 
 `dynamodb.service.endpoint` - AWS DynamoDB API Endpoint. Will use default AWS if not set.

source/src/main/java/com/trustpilot/connector/dynamodb/DynamoDBSourceConnector.java

@@ -57,12 +57,14 @@ public void start(Map<String, String> properties) {
                 AwsClients.buildAWSResourceGroupsTaggingAPIClient(config.getAwsRegion(),
                                                                   config.getResourceTaggingServiceEndpoint(),
                                                                   config.getAwsAccessKeyIdValue(),
-                                                                  config.getAwsSecretKeyValue());
+                                                                  config.getAwsSecretKeyValue(),
+                                                                  config.getAwsAssumeRoleArn());
 
         AmazonDynamoDB dynamoDBClient = AwsClients.buildDynamoDbClient(config.getAwsRegion(),
                                                                        config.getDynamoDBServiceEndpoint(),
                                                                        config.getAwsAccessKeyIdValue(),
-                                                                       config.getAwsSecretKeyValue());
+                                                                       config.getAwsSecretKeyValue(),
+                                                                       config.getAwsAssumeRoleArn());
 
         if (tablesProvider == null) {
             if (config.getWhitelistTables() != null) {

source/src/main/java/com/trustpilot/connector/dynamodb/DynamoDBSourceConnectorConfig.java

@@ -17,6 +17,11 @@ public class DynamoDBSourceConnectorConfig extends AbstractConfig {
 	public static final String SRC_INIT_SYNC_DELAY_DISPLAY = "INIT_SYNC delay";
 	public static final int SRC_INIT_SYNC_DELAY_DEFAULT = 60;
 
+	public static final String SRC_INIT_SYNC_SKIP_CONFIG = "init.sync.skip";
+	public static final String SRC_INIT_SYNC_SKIP_DOC = "Define whether to skip INIT_SYNC of table.";
+	public static final String SRC_INIT_SYNC_SKIP_DISPLAY = "Skip INIT_SYNC";
+	public static final boolean SRC_INIT_SYNC_SKIP_DEFAULT = false;
+
   	public static final String AWS_REGION_CONFIG = "aws.region";
 	public static final String AWS_REGION_DOC = "Define AWS region.";
 	public static final String AWS_REGION_DISPLAY = "Region";
@@ -57,6 +62,11 @@ public class DynamoDBSourceConnectorConfig extends AbstractConfig {
 	public static final String SRC_KCL_TABLE_BILLING_MODE_DISPLAY = "KCL table billing mode";
 	public static final String SRC_KCL_TABLE_BILLING_MODE_DEFAULT = "PROVISIONED";
 
+	public static final String AWS_ASSUME_ROLE_ARN_CONFIG = "aws.assume.role.arn";
+	public static final String AWS_ASSUME_ROLE_ARN_DOC = "Define which role arn the KCL/Dynamo Client should assume.";
+	public static final String AWS_ASSUME_ROLE_ARN_DISPLAY = "Assume Role Arn";
+	public static final String AWS_ASSUME_ROLE_ARN_DEFAULT = null;
+
 	public static final String DST_TOPIC_PREFIX_CONFIG = "kafka.topic.prefix";
 	public static final String DST_TOPIC_PREFIX_DOC = "Define Kafka topic destination prefix. End will be the name of a table.";
 	public static final String DST_TOPIC_PREFIX_DISPLAY = "Topic prefix";
@@ -181,6 +191,15 @@ public static ConfigDef baseConfigDef() {
 						ConfigDef.Width.MEDIUM,
 						SRC_KCL_TABLE_BILLING_MODE_DISPLAY)
 
+				.define(AWS_ASSUME_ROLE_ARN_CONFIG,
+						ConfigDef.Type.STRING,
+						AWS_ASSUME_ROLE_ARN_DEFAULT,
+						ConfigDef.Importance.LOW,
+						AWS_ASSUME_ROLE_ARN_DOC,
+						AWS_GROUP, 10,
+						ConfigDef.Width.LONG,
+						AWS_ASSUME_ROLE_ARN_DISPLAY)
+
 				.define(DST_TOPIC_PREFIX_CONFIG,
 						ConfigDef.Type.STRING,
 						DST_TOPIC_PREFIX_DEFAULT,
@@ -190,12 +209,21 @@ public static ConfigDef baseConfigDef() {
 						ConfigDef.Width.MEDIUM,
 						DST_TOPIC_PREFIX_DISPLAY)
 
+				.define(SRC_INIT_SYNC_SKIP_CONFIG,
+						ConfigDef.Type.BOOLEAN,
+						SRC_INIT_SYNC_SKIP_DEFAULT,
+						ConfigDef.Importance.LOW,
+						SRC_INIT_SYNC_SKIP_DOC,
+						CONNECTOR_GROUP, 2,
+						ConfigDef.Width.MEDIUM,
+						SRC_INIT_SYNC_SKIP_DISPLAY)
+
 				.define(SRC_INIT_SYNC_DELAY_CONFIG,
 						ConfigDef.Type.INT,
 						SRC_INIT_SYNC_DELAY_DEFAULT,
 						ConfigDef.Importance.LOW,
 						SRC_INIT_SYNC_DELAY_DOC,
-						CONNECTOR_GROUP, 2,
+						CONNECTOR_GROUP, 3,
 						ConfigDef.Width.MEDIUM,
 						SRC_INIT_SYNC_DELAY_DISPLAY)
 
@@ -253,6 +281,10 @@ public long getRediscoveryPeriod() {
 		return getLong(REDISCOVERY_PERIOD_CONFIG);
 	}
 
+	public boolean getInitSyncSkip() {
+		return (boolean)get(SRC_INIT_SYNC_SKIP_CONFIG);
+	}
+
 	public int getInitSyncDelay() {
 		return (int)get(SRC_INIT_SYNC_DELAY_CONFIG);
 	}
@@ -272,4 +304,8 @@ public List<String> getWhitelistTables() {
 	public BillingMode getKCLTableBillingMode() {
 		return BillingMode.fromValue(getString(SRC_KCL_TABLE_BILLING_MODE_CONFIG));
 	}
+
+	public String getAwsAssumeRoleArn() {
+		return getString(AWS_ASSUME_ROLE_ARN_CONFIG);
+	}
 }

source/src/main/java/com/trustpilot/connector/dynamodb/DynamoDBSourceTask.java

@@ -89,6 +89,7 @@ public class DynamoDBSourceTask extends SourceTask {
     private SourceInfo sourceInfo;
     private TableDescription tableDesc;
     private int initSyncDelay;
+    private boolean initSyncSkip;
 
     @SuppressWarnings("unused")
     //Used by Confluent platform to initialize connector
@@ -118,11 +119,13 @@ public void start(Map<String, String> configProperties) {
                     config.getAwsRegion(),
                     config.getDynamoDBServiceEndpoint(),
                     config.getAwsAccessKeyIdValue(),
-                    config.getAwsSecretKeyValue());
+                    config.getAwsSecretKeyValue(),
+                    config.getAwsAssumeRoleArn());
         }
         tableDesc = client.describeTable(config.getTableName()).getTable();
 
         initSyncDelay = config.getInitSyncDelay();
+        initSyncSkip = config.getInitSyncSkip();
 
         LOGGER.debug("Getting offset for table: {}", tableDesc.getTableName());
         setStateFromOffset();
@@ -142,15 +145,23 @@ public void start(Map<String, String> configProperties) {
                 config.getAwsRegion(),
                 config.getDynamoDBServiceEndpoint(),
                 config.getAwsAccessKeyIdValue(),
-                config.getAwsSecretKeyValue());
+                config.getAwsSecretKeyValue(),
+                config.getAwsAssumeRoleArn());
 
         if (kclWorker == null) {
             kclWorker = new KclWorkerImpl(
-                    AwsClients.getCredentials(config.getAwsAccessKeyIdValue(), config.getAwsSecretKeyValue()),
+                    AwsClients.getCredentials(config.getAwsAccessKeyIdValue(), config.getAwsSecretKeyValue(), config.getAwsAssumeRoleArn()),
                     eventsQueue,
                     shardRegister);
         }
-        kclWorker.start(client, dynamoDBStreamsClient, tableDesc.getTableName(), config.getTaskID(), config.getDynamoDBServiceEndpoint(), config.getKCLTableBillingMode());
+        kclWorker.start(client,
+                dynamoDBStreamsClient,
+                tableDesc.getTableName(),
+                config.getTaskID(),
+                config.getDynamoDBServiceEndpoint(),
+                config.getInitSyncSkip(),
+                config.getKCLTableBillingMode()
+        );
 
         shutdown = false;
     }
@@ -160,10 +171,17 @@ private void setStateFromOffset() {
                                             .offset(Collections.singletonMap("table_name", tableDesc.getTableName()));
         if (offset != null) {
             sourceInfo = SourceInfo.fromOffset(offset, clock);
+            if (initSyncSkip) {
+                sourceInfo.skipInitSync();
+            }
         } else {
             LOGGER.debug("No stored offset found for table: {}", tableDesc.getTableName());
             sourceInfo = new SourceInfo(tableDesc.getTableName(), clock);
-            sourceInfo.startInitSync(); // InitSyncStatus always needs to run after adding new table
+            if (initSyncSkip) {
+                sourceInfo.skipInitSync();
+            } else {
+                sourceInfo.startInitSync();
+            }
         }
     }
 
@@ -194,6 +212,9 @@ public List<SourceRecord> poll() throws InterruptedException {
             if (sourceInfo.initSyncStatus == InitSyncStatus.FINISHED) {
                 return sync();
             }
+            if (sourceInfo.initSyncStatus == InitSyncStatus.SKIPPED) {
+                return sync();
+            }
             throw new Exception("Invalid SourceInfo InitSyncStatus state: " + sourceInfo.initSyncStatus);
         } catch (InterruptedException ex) {
             LOGGER.error("Failed to handle incoming records. Records dropped!", ex);
@@ -236,6 +257,7 @@ private LinkedList<SourceRecord> initSync() throws Exception {
             result.add(converter.toSourceRecord(sourceInfo,
                                                 Envelope.Operation.READ,
                                                 record,
+                                                null,
                                                 sourceInfo.lastInitSyncStart,
                                                 null,
                                                 null));
@@ -254,6 +276,7 @@ private LinkedList<SourceRecord> initSync() throws Exception {
             result.add(converter.toSourceRecord(sourceInfo,
                                                 Envelope.Operation.READ,
                                                 lastRecord,
+                                                null,
                                                 sourceInfo.lastInitSyncStart,
                                                 null,
                                                 null));
@@ -277,6 +300,7 @@ private List<SourceRecord> sync() throws Exception {
         LOGGER.debug("Waiting for records from eventsQueue for table: {}", tableDesc.getTableName());
         KclRecordsWrapper dynamoDBRecords = eventsQueue.poll(500, TimeUnit.MILLISECONDS);
         if (dynamoDBRecords == null) {
+            LOGGER.debug("null dynamoDBRecords");
             return null; // returning thread control at regular intervals
         }
 
@@ -310,12 +334,12 @@ private List<SourceRecord> sync() throws Exception {
                 }
 
                 // Received record which is behind "safe" zone. Indicating that "potentially" we lost some records.
-                // Need to resync...
+                // Need to resync if sync hasn't been skipped...
                 // This happens if:
                 // * connector was down for some time
                 // * connector is lagging
                 // * connector failed to finish init sync in acceptable time frame
-                if (recordIsInDangerZone(arrivalTimestamp)) {
+                if (recordIsInDangerZone(arrivalTimestamp) && sourceInfo.initSyncStatus != InitSyncStatus.SKIPPED) {
                     sourceInfo.startInitSync();
 
                     LOGGER.info(
@@ -342,9 +366,12 @@ private List<SourceRecord> sync() throws Exception {
                     attributes = dynamoDbRecord.getDynamodb().getKeys();
                 }
 
+                Map<String, AttributeValue> oldImage = dynamoDbRecord.getDynamodb().getOldImage();
+
                 SourceRecord sourceRecord = converter.toSourceRecord(sourceInfo,
                                                                      op,
                                                                      attributes,
+                                                                     oldImage,
                                                                      arrivalTimestamp.toInstant(),
                                                                      dynamoDBRecords.getShardId(),
                                                                      record.getSequenceNumber());

source/src/main/java/com/trustpilot/connector/dynamodb/Envelope.java

@@ -56,6 +56,10 @@ public static final class FieldName {
          * The {@code after} field is used to store the state of a record after an operation.
          */
         public static final String DOCUMENT = "document";
+        /**
+         * The {@code oldDocument} field is used to store the state of a record before an operation.
+         */
+        public static final String OLD_DOCUMENT = "old_document";
         /**
          * The {@code op} field is used to store the kind of operation on a record.
          */

source/src/main/java/com/trustpilot/connector/dynamodb/InitSyncStatus.java

@@ -3,5 +3,6 @@
 public enum InitSyncStatus {
     UNDEFINED,
     RUNNING,
-    FINISHED
+    FINISHED,
+    SKIPPED
 }

source/src/main/java/com/trustpilot/connector/dynamodb/SourceInfo.java

@@ -71,6 +71,14 @@ public void endInitSync() {
         lastInitSyncEnd = Instant.now(clock);
     }
 
+    public void skipInitSync() {
+        initSyncStatus = InitSyncStatus.SKIPPED;
+        lastInitSyncStart = Instant.ofEpochSecond(0);
+        lastInitSyncEnd = Instant.ofEpochSecond(0);;
+        exclusiveStartKey = null;
+        initSyncCount = 0L;
+    }
+
     private static final Schema STRUCT_SCHEMA = SchemaBuilder.struct()
                                                              .name(SchemaNameAdjuster
                                                                            .defaultAdjuster()