diff --git a/README.md b/README.md index 6bb344b..abedab1 100644 --- a/README.md +++ b/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/ diff --git a/build.gradle b/build.gradle index 43ec23f..e4c935a 100644 --- a/build.gradle +++ b/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) diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar index c8bf763..fdcbf55 100644 Binary files a/gradle/wrapper/gradle-wrapper.jar and b/gradle/wrapper/gradle-wrapper.jar differ diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties index 4ba4026..a44070a 100644 --- a/gradle/wrapper/gradle-wrapper.properties +++ b/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 diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml index be24a0f..5313e15 100644 --- a/src/checkstyle/checkstyle-suppressions.xml +++ b/src/checkstyle/checkstyle-suppressions.xml @@ -4,8 +4,6 @@ "http://www.puppycrawl.com/dtds/suppressions_1_1.dtd"> - - diff --git a/src/main/java/org/springframework/integration/aws/inbound/S3InboundFileSynchronizer.java b/src/main/java/org/springframework/integration/aws/inbound/S3InboundFileSynchronizer.java index 9e51231..8e574b2 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/S3InboundFileSynchronizer.java +++ b/src/main/java/org/springframework/integration/aws/inbound/S3InboundFileSynchronizer.java @@ -84,9 +84,9 @@ public class S3InboundFileSynchronizer extends AbstractInboundFileSynchronizer session) throws IOException { - super.copyFileToLocalDirectory(((S3Session) session).normalizeBucketName(remoteDirectoryPath), + return super.copyFileToLocalDirectory(((S3Session) session).normalizeBucketName(remoteDirectoryPath), remoteFile, localDirectory, session); } diff --git a/src/main/java/org/springframework/integration/aws/metadata/DynamoDbMetaDataStore.java b/src/main/java/org/springframework/integration/aws/metadata/DynamoDbMetaDataStore.java new file mode 100644 index 0000000..7df6005 --- /dev/null +++ b/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() { + + @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 waiter = + DynamoDbMetaDataStore.this.dynamoDB.waiters() + .tableExists(); + + WaiterParameters waiterParameters = + new WaiterParameters<>( + new DescribeTableRequest(DynamoDbMetaDataStore.this.table.getTableName())) + .withPollingStrategy( + new PollingStrategy(new MaxAttemptsRetryStrategy(25), + new FixedDelayStrategy(1))); + + waiter.runAsync(waiterParameters, new WaiterHandler() { + + @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; + } + } + +} diff --git a/src/main/java/org/springframework/integration/aws/metadata/package-info.java b/src/main/java/org/springframework/integration/aws/metadata/package-info.java new file mode 100644 index 0000000..4f17105 --- /dev/null +++ b/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; diff --git a/src/test/java/org/springframework/integration/aws/DynamoDbRunning.java b/src/test/java/org/springframework/integration/aws/DynamoDbRunning.java new file mode 100644 index 0000000..f5c9e8b --- /dev/null +++ b/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 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); + } + +} diff --git a/src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java b/src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java new file mode 100644 index 0000000..827ffeb --- /dev/null +++ b/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 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"); + } + +}