From 598384bc8cf37ed01dc6a90aa029c0ba2a7f91df Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 22 May 2018 16:51:50 -0400 Subject: [PATCH] Add TTL to `DynamoDbMetaDataStore` Fixes https://github.com/spring-projects/spring-integration-aws/issues/92 --- README.md | 7 +- .../inbound/kinesis/ShardCheckpointer.java | 2 +- .../aws/metadata/DynamoDbMetaDataStore.java | 115 +++++++++++++----- .../metadata/DynamoDbMetadataStoreTests.java | 11 +- 4 files changed, 99 insertions(+), 36 deletions(-) diff --git a/README.md b/README.md index 8014c73..3de3bab 100644 --- a/README.md +++ b/README.md @@ -497,7 +497,11 @@ this.amazonDynamoDB = AmazonDynamoDBAsyncClientBuilder.standard() ```` Where you should specify the port on which you have ran the Dynalite service. -Also you can use for you testing purpose a copy of `org.springframework.integration.aws.DynamoDbLocalRunning` in the `/test` directory of this project. +Also you can use for your testing purpose a copy of `org.springframework.integration.aws.DynamoDbLocalRunning` in the `/test` directory of this project. + +Starting with _version 2.0_, the ``DynamoDbMetaDataStore` can be configured with the `timeToLive` option to enable the [DynamoDB TTL][] feature. +The `TTL` attribute is added to each item with the value based on the sum of current time and provided `timeToLive` in seconds. +If the provided `timeToLive` value is non-positive, the TTL functionality is disable on the table. ## Amazon Kinesis @@ -623,6 +627,7 @@ Also you can use for you testing purpose a copy of `org.springframework.integrat [Pull requests]: http://help.github.com/send-pull-requests [contributor guidelines]: https://github.com/spring-projects/spring-integration/blob/master/CONTRIBUTING.adoc [Dynalite]: https://github.com/mhart/dynalite +[DynamoDB TTL]: https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/TTL.html [Kinesis Client Library]: https://github.com/awslabs/amazon-kinesis-client [Kinesalite]: https://github.com/mhart/kinesalite [Amazon SQS Message Attributes]: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-attributes.html diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java index 9d43c81..b8d7a16 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java @@ -95,7 +95,7 @@ class ShardCheckpointer implements Checkpointer { if (new BigInteger(sequenceNumber).compareTo(new BigInteger(this.lastCheckpointValue)) <= 0) { if (logger.isDebugEnabled()) { logger.debug("Removing record with sequenceNumber " + sequenceNumber + - " because the sequenceNumber is <= checkpoint(" + this.lastCheckpointValue + ")"); + " because it is <= checkpoint(" + this.lastCheckpointValue + ")"); } iterator.remove(); } diff --git a/src/main/java/org/springframework/integration/aws/metadata/DynamoDbMetaDataStore.java b/src/main/java/org/springframework/integration/aws/metadata/DynamoDbMetaDataStore.java index cbabbac..0aac58e 100644 --- a/src/main/java/org/springframework/integration/aws/metadata/DynamoDbMetaDataStore.java +++ b/src/main/java/org/springframework/integration/aws/metadata/DynamoDbMetaDataStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -47,6 +47,8 @@ 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.services.dynamodbv2.model.TimeToLiveSpecification; +import com.amazonaws.services.dynamodbv2.model.UpdateTimeToLiveRequest; import com.amazonaws.waiters.FixedDelayStrategy; import com.amazonaws.waiters.MaxAttemptsRetryStrategy; import com.amazonaws.waiters.PollingStrategy; @@ -74,6 +76,8 @@ public class DynamoDbMetaDataStore implements ConcurrentMetadataStore, Initializ private static final String VALUE = "VALUE"; + private static final String TTL = "TTL"; + private final AmazonDynamoDBAsync dynamoDB; private final Table table; @@ -88,6 +92,8 @@ public class DynamoDbMetaDataStore implements ConcurrentMetadataStore, Initializ private long writeCapacity = 1L; + private Integer timeToLive; + public DynamoDbMetaDataStore(AmazonDynamoDBAsync dynamoDB) { this(dynamoDB, DEFAULT_TABLE_NAME); } @@ -118,10 +124,22 @@ public class DynamoDbMetaDataStore implements ConcurrentMetadataStore, Initializ this.createTableDelay = createTableDelay; } + /** + * Configure a period in seconds for items expiration. + * If it is configured to non-positive value (<= 0), the TTL is disabled on the table. + * @param timeToLive period in seconds for items expiration. + * @since 2.0 + * @see DynamoDB TTL + */ + public void setTimeToLive(int timeToLive) { + this.timeToLive = timeToLive; + } + @Override - public void afterPropertiesSet() throws Exception { + public void afterPropertiesSet() { try { this.table.describe(); + updateTimeToLiveIfAny(); this.createTableLatch.countDown(); return; } @@ -167,6 +185,7 @@ public class DynamoDbMetaDataStore implements ConcurrentMetadataStore, Initializ @Override public void onWaitSuccess(DescribeTableRequest request) { + updateTimeToLiveIfAny(); DynamoDbMetaDataStore.this.createTableLatch.countDown(); DynamoDbMetaDataStore.this.table.describe(); } @@ -184,6 +203,20 @@ public class DynamoDbMetaDataStore implements ConcurrentMetadataStore, Initializ }); } + private void updateTimeToLiveIfAny() { + if (this.timeToLive != null) { + UpdateTimeToLiveRequest updateTimeToLiveRequest = + new UpdateTimeToLiveRequest() + .withTableName(this.table.getTableName()) + .withTimeToLiveSpecification( + new TimeToLiveSpecification() + .withAttributeName(TTL) + .withEnabled(this.timeToLive > 0)); + + this.dynamoDB.updateTimeToLive(updateTimeToLiveRequest); + } + } + private void awaitForActive() { try { this.createTableLatch.await(this.createTableRetries * this.createTableDelay, TimeUnit.SECONDS); @@ -202,10 +235,16 @@ public class DynamoDbMetaDataStore implements ConcurrentMetadataStore, Initializ awaitForActive(); - this.table.putItem( + Item item = new Item() .withPrimaryKey(KEY, key) - .withString(VALUE, value)); + .withString(VALUE, value); + + if (this.timeToLive != null && this.timeToLive > 0) { + item = item.withLong(TTL, (System.currentTimeMillis() + this.timeToLive) / 1000); + } + + this.table.putItem(item); } @Override @@ -226,17 +265,25 @@ public class DynamoDbMetaDataStore implements ConcurrentMetadataStore, Initializ awaitForActive(); - try { - this.table.updateItem( - new UpdateItemSpec() - .withPrimaryKey(KEY, key) - .withAttributeUpdate( - new AttributeUpdate(VALUE) - .put(value)) - .withExpected( - new Expected(KEY) - .notExist())); + UpdateItemSpec updateItemSpec = + new UpdateItemSpec() + .withPrimaryKey(KEY, key) + .withAttributeUpdate( + new AttributeUpdate(VALUE) + .put(value)) + .withExpected( + new Expected(KEY) + .notExist()); + if (this.timeToLive != null && this.timeToLive > 0) { + updateItemSpec = + updateItemSpec.addAttributeUpdate( + new AttributeUpdate(TTL) + .put((System.currentTimeMillis() + this.timeToLive) / 1000)); + } + + try { + this.table.updateItem(updateItemSpec); return null; } catch (ConditionalCheckFailedException e) { @@ -252,17 +299,26 @@ public class DynamoDbMetaDataStore implements ConcurrentMetadataStore, Initializ awaitForActive(); + UpdateItemSpec updateItemSpec = + new UpdateItemSpec() + .withPrimaryKey(KEY, key) + .withAttributeUpdate( + new AttributeUpdate(VALUE) + .put(newValue)) + .withExpected( + new Expected(VALUE) + .eq(oldValue)) + .withReturnValues(ReturnValue.UPDATED_NEW); + + if (this.timeToLive != null && this.timeToLive > 0) { + updateItemSpec = + updateItemSpec.addAttributeUpdate( + new AttributeUpdate(TTL) + .put((System.currentTimeMillis() + this.timeToLive) / 1000)); + } + 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)) + return this.table.updateItem(updateItemSpec) .getItem() != null; } catch (ConditionalCheckFailedException e) { @@ -276,11 +332,12 @@ public class DynamoDbMetaDataStore implements ConcurrentMetadataStore, Initializ awaitForActive(); - Item item = this.table.deleteItem( - new DeleteItemSpec() - .withPrimaryKey(KEY, key) - .withReturnValues(ReturnValue.ALL_OLD)) - .getItem(); + Item item = + this.table.deleteItem( + new DeleteItemSpec() + .withPrimaryKey(KEY, key) + .withReturnValues(ReturnValue.ALL_OLD)) + .getItem(); return getValueIfAny(item); } diff --git a/src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java b/src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java index 86fd8d3..9efd621 100644 --- a/src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java +++ b/src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -58,7 +58,7 @@ public class DynamoDbMetadataStoreTests { private final String file1Id = "12345"; @BeforeClass - public static void setup() throws Exception { + public static void setup() { AmazonDynamoDBAsync dynamoDB = DYNAMO_DB_RUNNING.getDynamoDB(); try { @@ -77,6 +77,7 @@ public class DynamoDbMetadataStoreTests { } store = new DynamoDbMetaDataStore(dynamoDB, TEST_TABLE); + store.setTimeToLive(10); // Dynalite doesn't support TTL store.afterPropertiesSet(); } @@ -106,7 +107,7 @@ public class DynamoDbMetadataStoreTests { } @Test - public void testPutIfAbsent() throws Exception { + public void testPutIfAbsent() { String fileID = store.get(this.file1); assertThat(fileID).describedAs("Get First time, Value must not exist").isNull(); @@ -121,7 +122,7 @@ public class DynamoDbMetadataStoreTests { } @Test - public void testRemove() throws Exception { + public void testRemove() { String fileID = store.remove(this.file1); assertThat(fileID).isNull(); @@ -137,7 +138,7 @@ public class DynamoDbMetadataStoreTests { } @Test - public void testReplace() throws Exception { + public void testReplace() { boolean removedValue = store.replace(this.file1, this.file1Id, "4567"); assertThat(removedValue).isFalse();