Spring/spring-integration-aws
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add TTL to DynamoDbMetaDataStore

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-aws/issues/92
This commit is contained in:
Artem Bilan
2018-05-22 16:51:50 -04:00
parent 50c1d6dcb7
commit 598384bc8c
4 changed files with 99 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java
View File

@@ -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();
}

115
src/main/java/org/springframework/integration/aws/metadata/DynamoDbMetaDataStore.java
View File

@@ -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 <a href="https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/TTL.html">DynamoDB TTL</a>
*/
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);
}

11
src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java
View File

@@ -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();