GH-64: Add DynamoDbMetaDataStore implementation

Fixes spring-projects/spring-integration-aws#64

* Add `DynamoDbRunning` for testing against locally ran DynamoDB
* Upgrade to Gradle 4.0.1, SI-4.3.11
* Switch on some Checkstyle rules for tests

README.md

@@ -497,6 +497,13 @@ flag to `true` ot switch it to the gateway mode.
 By default the `SnsMessageHandler` is one-way `MessageHandler`.
 
 
+## Metadata Store for Amazon DynamoDB
+
+The `DynamoDbMetaDataStore`, a `ConcurrentMetadataStore` implementation, is provided to keep the metadata for Spring Integration components in the distributed Amazon DynamoDB store. 
+The implementation is based on a simple table with `KEY` and `VALUE` attributes, both are string types and the `KEY` is primary key of the table.
+By default the `SpringIntegrationMetadataStore` table is used and it is created during `DynamoDbMetaDataStore` initialization if that doesn't exist yet.
+The `DynamoDbMetaDataStore` can be used for the `KinesisMessageDrivenChannelAdapter` as a cloud-based `cehckpointStore`.
+
 [Spring Cloud AWS]: https://github.com/spring-cloud/spring-cloud-aws
 [AWS SDK for Java]: http://aws.amazon.com/sdkforjava/
 [Amazon Web Services]: http://aws.amazon.com/

build.gradle

@@ -35,7 +35,7 @@ ext {
 	servletApiVersion = '3.1.0'
 	slf4jVersion = '1.7.25'
 	springCloudAwsVersion = '1.2.1.RELEASE'
-	springIntegrationVersion = '4.3.10.RELEASE'
+	springIntegrationVersion = '4.3.11.RELEASE'
 
 	idPrefix = 'aws'
 
@@ -108,6 +108,7 @@ dependencies {
 	compile('org.springframework.integration:spring-integration-http', optional)
 
 	compile('com.amazonaws:aws-java-sdk-kinesis', optional)
+	compile('com.amazonaws:aws-java-sdk-dynamodb', optional)
 
 	compile("javax.servlet:javax.servlet-api:$servletApiVersion", provided)
 

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
-#Mon Jul 10 20:22:20 EDT 2017
+#Wed Jul 26 16:30:33 EDT 2017
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-4.0-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.0.1-bin.zip

src/checkstyle/checkstyle-suppressions.xml

@@ -4,8 +4,6 @@
     "http://www.puppycrawl.com/dtds/suppressions_1_1.dtd">
 <suppressions>
 	<suppress files="package-info\.java" checks=".*" />
-	<suppress files="[\\/]test[\\/]" checks="RequireThis" />
-	<suppress files="[\\/]test[\\/]" checks="FinalClass" />
 	<suppress files="[\\/]test[\\/]" checks="AvoidStaticImport" />
 	<suppress files="[\\/]test[\\/]" checks="InnerTypeLast" />
 	<suppress files="[\\/]test[\\/]" checks="Javadoc*" />

src/main/java/org/springframework/integration/aws/inbound/S3InboundFileSynchronizer.java

@@ -84,9 +84,9 @@ public class S3InboundFileSynchronizer extends AbstractInboundFileSynchronizer<S
 	}
 
 	@Override
-	protected void copyFileToLocalDirectory(String remoteDirectoryPath, S3ObjectSummary remoteFile,
+	protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, S3ObjectSummary remoteFile,
 			File localDirectory, Session<S3ObjectSummary> session) throws IOException {
-		super.copyFileToLocalDirectory(((S3Session) session).normalizeBucketName(remoteDirectoryPath),
+		return super.copyFileToLocalDirectory(((S3Session) session).normalizeBucketName(remoteDirectoryPath),
 				remoteFile, localDirectory, session);
 	}
 

src/main/java/org/springframework/integration/aws/metadata/DynamoDbMetaDataStore.java

@@ -0,0 +1,283 @@
+/*
+ * Copyright 2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.aws.metadata;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.metadata.ConcurrentMetadataStore;
+import org.springframework.util.Assert;
+
+import com.amazonaws.handlers.AsyncHandler;
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
+import com.amazonaws.services.dynamodbv2.document.AttributeUpdate;
+import com.amazonaws.services.dynamodbv2.document.DynamoDB;
+import com.amazonaws.services.dynamodbv2.document.Expected;
+import com.amazonaws.services.dynamodbv2.document.Item;
+import com.amazonaws.services.dynamodbv2.document.Table;
+import com.amazonaws.services.dynamodbv2.document.spec.DeleteItemSpec;
+import com.amazonaws.services.dynamodbv2.document.spec.UpdateItemSpec;
+import com.amazonaws.services.dynamodbv2.model.AttributeDefinition;
+import com.amazonaws.services.dynamodbv2.model.ConditionalCheckFailedException;
+import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
+import com.amazonaws.services.dynamodbv2.model.CreateTableResult;
+import com.amazonaws.services.dynamodbv2.model.DescribeTableRequest;
+import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
+import com.amazonaws.services.dynamodbv2.model.KeyType;
+import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
+import com.amazonaws.services.dynamodbv2.model.ResourceNotFoundException;
+import com.amazonaws.services.dynamodbv2.model.ReturnValue;
+import com.amazonaws.services.dynamodbv2.model.ScalarAttributeType;
+import com.amazonaws.waiters.FixedDelayStrategy;
+import com.amazonaws.waiters.MaxAttemptsRetryStrategy;
+import com.amazonaws.waiters.PollingStrategy;
+import com.amazonaws.waiters.Waiter;
+import com.amazonaws.waiters.WaiterHandler;
+import com.amazonaws.waiters.WaiterParameters;
+
+/**
+ * The {@link ConcurrentMetadataStore} for the {@link AmazonDynamoDB}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 1.1
+ */
+public class DynamoDbMetaDataStore implements ConcurrentMetadataStore, InitializingBean {
+
+	/**
+	 * The {@value DEFAULT_TABLE_NAME} default name for the metadata table in the DynamoDB.
+	 */
+	public static final String DEFAULT_TABLE_NAME = "SpringIntegrationMetadataStore";
+
+	private static final Log logger = LogFactory.getLog(DynamoDbMetaDataStore.class);
+
+	private static final String KEY = "KEY";
+
+	private static final String VALUE = "VALUE";
+
+	private final AmazonDynamoDBAsync dynamoDB;
+
+	private final Table table;
+
+	private final CountDownLatch createTableLatch = new CountDownLatch(1);
+
+	private Long readCapacity = 10000L;
+
+	private Long writeCapacity = 10000L;
+
+	public DynamoDbMetaDataStore(AmazonDynamoDBAsync dynamoDB) {
+		this(dynamoDB, DEFAULT_TABLE_NAME);
+	}
+
+	public DynamoDbMetaDataStore(AmazonDynamoDBAsync dynamoDB, String tableName) {
+		Assert.notNull(dynamoDB, "'dynamoDB' must not be null.");
+		Assert.hasText(tableName, "'tableName' must not be empty.");
+		this.dynamoDB = dynamoDB;
+		this.table =
+				new DynamoDB(this.dynamoDB)
+						.getTable(tableName);
+
+	}
+
+	public void setReadCapacity(Long readCapacity) {
+		this.readCapacity = readCapacity;
+	}
+
+	public void setWriteCapacity(Long writeCapacity) {
+		this.writeCapacity = writeCapacity;
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		try {
+			this.table.describe();
+			createTableLatch.countDown();
+			return;
+		}
+		catch (ResourceNotFoundException e) {
+			if (logger.isInfoEnabled()) {
+				logger.info("No table '" + this.table.getTableName() + "'. Creating one...");
+			}
+		}
+
+		CreateTableRequest createTableRequest =
+				new CreateTableRequest()
+						.withTableName(this.table.getTableName())
+						.withKeySchema(new KeySchemaElement(KEY, KeyType.HASH))
+						.withAttributeDefinitions(new AttributeDefinition(KEY, ScalarAttributeType.S))
+						.withProvisionedThroughput(new ProvisionedThroughput(this.readCapacity, this.writeCapacity));
+
+
+		this.dynamoDB.createTableAsync(createTableRequest,
+				new AsyncHandler<CreateTableRequest, CreateTableResult>() {
+
+					@Override
+					public void onError(Exception e) {
+						logger.error("Cannot create DynamoDb table: " +
+								DynamoDbMetaDataStore.this.table.getTableName(), e);
+						DynamoDbMetaDataStore.this.createTableLatch.countDown();
+					}
+
+					@Override
+					public void onSuccess(CreateTableRequest request, CreateTableResult createTableResult) {
+						Waiter<DescribeTableRequest> waiter =
+								DynamoDbMetaDataStore.this.dynamoDB.waiters()
+										.tableExists();
+
+						WaiterParameters<DescribeTableRequest> waiterParameters =
+								new WaiterParameters<>(
+										new DescribeTableRequest(DynamoDbMetaDataStore.this.table.getTableName()))
+										.withPollingStrategy(
+												new PollingStrategy(new MaxAttemptsRetryStrategy(25),
+														new FixedDelayStrategy(1)));
+
+						waiter.runAsync(waiterParameters, new WaiterHandler<DescribeTableRequest>() {
+
+							@Override
+							public void onWaitSuccess(DescribeTableRequest request) {
+								DynamoDbMetaDataStore.this.createTableLatch.countDown();
+								DynamoDbMetaDataStore.this.table.describe();
+							}
+
+							@Override
+							public void onWaitFailure(Exception e) {
+								logger.error("Cannot describe DynamoDb table: " +
+										DynamoDbMetaDataStore.this.table.getTableName(), e);
+								DynamoDbMetaDataStore.this.createTableLatch.countDown();
+							}
+
+						});
+					}
+
+				});
+	}
+
+	private void awaitForActive() {
+		try {
+			this.createTableLatch.await(10, TimeUnit.SECONDS);
+		}
+		catch (InterruptedException e) {
+
+		}
+	}
+
+	@Override
+	public void put(String key, String value) {
+		Assert.hasText(key, "'key' must not be empty.");
+		Assert.hasText(value, "'value' must not be empty.");
+
+		awaitForActive();
+
+		this.table.putItem(
+				new Item()
+						.withPrimaryKey(KEY, key)
+						.withString(VALUE, value));
+	}
+
+	@Override
+	public String get(String key) {
+		Assert.hasText(key, "'key' must not be empty.");
+
+
+		awaitForActive();
+
+		Item item = this.table.getItem(KEY, key);
+
+		return getValueIfAny(item);
+	}
+
+	@Override
+	public String putIfAbsent(String key, String value) {
+		Assert.hasText(key, "'key' must not be empty.");
+		Assert.hasText(value, "'value' must not be empty.");
+
+		awaitForActive();
+
+		try {
+			this.table.updateItem(
+					new UpdateItemSpec()
+							.withPrimaryKey(KEY, key)
+							.withAttributeUpdate(
+									new AttributeUpdate(VALUE)
+											.put(value))
+							.withExpected(
+									new Expected(KEY)
+											.notExist()));
+
+			return null;
+		}
+		catch (ConditionalCheckFailedException e) {
+			return get(key);
+		}
+	}
+
+	@Override
+	public boolean replace(String key, String oldValue, String newValue) {
+		Assert.hasText(key, "'key' must not be empty.");
+		Assert.hasText(oldValue, "'value' must not be empty.");
+		Assert.hasText(newValue, "'newValue' must not be empty.");
+
+		awaitForActive();
+
+		try {
+			return this.table.updateItem(
+					new UpdateItemSpec()
+							.withPrimaryKey(KEY, key)
+							.withAttributeUpdate(
+									new AttributeUpdate(VALUE)
+											.put(newValue))
+							.withExpected(
+									new Expected(VALUE)
+											.eq(oldValue))
+							.withReturnValues(ReturnValue.UPDATED_NEW))
+					.getItem() != null;
+		}
+		catch (ConditionalCheckFailedException e) {
+			return false;
+		}
+	}
+
+	@Override
+	public String remove(String key) {
+		Assert.hasText(key, "'key' must not be empty.");
+
+		awaitForActive();
+
+		Item item = this.table.deleteItem(
+				new DeleteItemSpec()
+						.withPrimaryKey(KEY, key)
+						.withReturnValues(ReturnValue.ALL_OLD))
+				.getItem();
+
+		return getValueIfAny(item);
+	}
+
+	private static String getValueIfAny(Item item) {
+		if (item != null) {
+			return item.getString(VALUE);
+		}
+		else {
+			return null;
+		}
+	}
+
+}

src/main/java/org/springframework/integration/aws/metadata/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides classes supporting metadata stores.
+ */
+package org.springframework.integration.aws.metadata;

src/test/java/org/springframework/integration/aws/DynamoDbRunning.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.aws;
+
+import static org.junit.Assume.assumeNoException;
+import static org.junit.Assume.assumeTrue;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.SdkClientException;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.regions.Regions;
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClientBuilder;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 1.1
+ */
+public final class DynamoDbRunning extends TestWatcher {
+
+	private static Log logger = LogFactory.getLog(DynamoDbRunning.class);
+
+	// Static so that we only test once on failure: speeds up test suite
+	private static Map<Integer, Boolean> dynamoDbOnline = new HashMap<>();
+
+	private final int port;
+
+	private AmazonDynamoDBAsync amazonDynamoDB;
+
+	private DynamoDbRunning(int port) {
+		this.port = port;
+		dynamoDbOnline.put(port, true);
+	}
+
+	public AmazonDynamoDBAsync getDynamoDB() {
+		return this.amazonDynamoDB;
+	}
+
+	@Override
+	public Statement apply(Statement base, Description description) {
+		assumeTrue(dynamoDbOnline.get(this.port));
+
+		String url = "http://localhost:" + this.port;
+
+		this.amazonDynamoDB = AmazonDynamoDBAsyncClientBuilder.standard()
+				.withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("", "")))
+				.withClientConfiguration(
+						new ClientConfiguration()
+								.withMaxErrorRetry(0)
+								.withConnectionTimeout(1000))
+				.withEndpointConfiguration(
+						new AwsClientBuilder.EndpointConfiguration(url, Regions.DEFAULT_REGION.getName()))
+				.build();
+
+		try {
+			this.amazonDynamoDB.listTables();
+		}
+		catch (SdkClientException e) {
+			logger.warn("Tests not running because no DynamoDb on " + url, e);
+			assumeNoException(e);
+		}
+		return super.apply(base, description);
+	}
+
+
+	public static DynamoDbRunning isRunning(int port) {
+		return new DynamoDbRunning(port);
+	}
+
+}

src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java

@@ -0,0 +1,158 @@
+/*
+ * Copyright 2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.aws.metadata;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import org.springframework.integration.aws.DynamoDbRunning;
+import org.springframework.integration.test.util.TestUtils;
+
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
+import com.amazonaws.services.dynamodbv2.model.AttributeValue;
+import com.amazonaws.services.dynamodbv2.model.DescribeTableRequest;
+import com.amazonaws.waiters.FixedDelayStrategy;
+import com.amazonaws.waiters.MaxAttemptsRetryStrategy;
+import com.amazonaws.waiters.PollingStrategy;
+import com.amazonaws.waiters.Waiter;
+import com.amazonaws.waiters.WaiterParameters;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 1.1
+ *
+ */
+public class DynamoDbMetadataStoreTests {
+
+	@ClassRule
+	public static final DynamoDbRunning DYNAMO_DB_RUNNING = DynamoDbRunning.isRunning(4567);
+
+	private static final String TEST_TABLE = "testMetadataStore";
+
+	private static DynamoDbMetaDataStore store;
+
+	private final String file1 = "/remotepath/filesTodownload/file-1.txt";
+
+	private final String file1Id = "12345";
+
+	@BeforeClass
+	public static void setup() throws Exception {
+		AmazonDynamoDBAsync dynamoDB = DYNAMO_DB_RUNNING.getDynamoDB();
+
+		try {
+			dynamoDB.deleteTableAsync(TEST_TABLE);
+
+			Waiter<DescribeTableRequest> waiter =
+					dynamoDB.waiters()
+							.tableNotExists();
+
+			waiter.run(new WaiterParameters<>(new DescribeTableRequest(TEST_TABLE))
+					.withPollingStrategy(new PollingStrategy(new MaxAttemptsRetryStrategy(25),
+							new FixedDelayStrategy(1))));
+		}
+		catch (Exception e) {
+
+		}
+
+		store = new DynamoDbMetaDataStore(dynamoDB, TEST_TABLE);
+		store.afterPropertiesSet();
+	}
+
+	@Before
+	public void clear() throws InterruptedException {
+		CountDownLatch createTableLatch = TestUtils.getPropertyValue(store, "createTableLatch", CountDownLatch.class);
+
+		createTableLatch.await();
+
+		DYNAMO_DB_RUNNING.getDynamoDB()
+				.deleteItem(TEST_TABLE,
+						Collections.singletonMap("KEY",
+								new AttributeValue()
+										.withS(this.file1)));
+	}
+
+	@Test
+	public void testGetFromStore() {
+		String fileID = store.get(this.file1);
+		assertThat(fileID).isNull();
+
+		store.put(this.file1, this.file1Id);
+
+		fileID = store.get(this.file1);
+		assertThat(fileID).isNotNull();
+		assertThat(fileID).isEqualTo(this.file1Id);
+	}
+
+	@Test
+	public void testPutIfAbsent() throws Exception {
+		String fileID = store.get(file1);
+		assertThat(fileID).describedAs("Get First time, Value must not exist").isNull();
+
+		fileID = store.putIfAbsent(file1, file1Id);
+		assertThat(fileID).describedAs("Insert First time, Value must return null").isNull();
+
+		fileID = store.putIfAbsent(file1, "56789");
+		assertThat(fileID).describedAs("Key Already Exists - Insertion Failed, ol value must be returned").isNotNull();
+		assertThat(fileID).describedAs("The Old Value must be equal to returned").isEqualTo(this.file1Id);
+
+		assertThat(store.get(this.file1)).describedAs("The Old Value must return").isEqualTo(this.file1Id);
+	}
+
+	@Test
+	public void testRemove() throws Exception {
+		String fileID = store.remove(this.file1);
+		assertThat(fileID).isNull();
+
+		fileID = store.putIfAbsent(this.file1, this.file1Id);
+		assertThat(fileID).isNull();
+
+		fileID = store.remove(this.file1);
+		assertThat(fileID).isNotNull();
+		assertThat(fileID).isEqualTo(this.file1Id);
+
+		fileID = store.get(this.file1);
+		assertThat(fileID).isNull();
+	}
+
+	@Test
+	public void testReplace() throws Exception {
+		boolean removedValue = store.replace(file1, file1Id, "4567");
+		assertThat(removedValue).isFalse();
+
+		String fileID = store.get(this.file1);
+		assertThat(fileID).isNull();
+
+		fileID = store.putIfAbsent(this.file1, file1Id);
+		assertThat(fileID).isNull();
+
+		removedValue = store.replace(this.file1, file1Id, "4567");
+		assertThat(removedValue).isTrue();
+
+		fileID = store.get(this.file1);
+		assertThat(fileID).isNotNull();
+		assertThat(fileID).isEqualTo("4567");
+	}
+
+}