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

Some fixes for the DynamoDbLockRepository

Browse Source 
* Increase a `DEFAULT_LEASE_DURATION` to `1 min`
* Upgrade to `localstack:1.4.0` Docker image
This commit is contained in:
abilan
2023-02-28 10:32:59 -05:00
parent 88cc93018e
commit ff658f2f84
5 changed files with 30 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java
View File

@@ -1594,7 +1594,9 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
}
private boolean renewLockInRegistry(LockCompletableFuture renewLockFuture) {
if (KinesisMessageDrivenChannelAdapter.this.lockRegistry instanceof RenewableLockRegistry renewableLockRegistry) {
if (KinesisMessageDrivenChannelAdapter.this.lockRegistry
instanceof RenewableLockRegistry renewableLockRegistry) {
try {
renewableLockRegistry.renewLock(renewLockFuture.lockKey);
return renewLockFuture.complete(true);

2
src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2023 the original author or authors.
* Copyright 2018-2023 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.

22
src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRepository.java
View File

@@ -112,7 +112,7 @@ public class DynamoDbLockRepository implements InitializingBean, DisposableBean,
/**
* Default value for the {@link #leaseDuration} property.
*/
public static final Duration DEFAULT_LEASE_DURATION = Duration.ofSeconds(10);
public static final Duration DEFAULT_LEASE_DURATION = Duration.ofSeconds(60);
private static final Log LOGGER = LogFactory.getLog(DynamoDbLockRegistry.class);
@@ -309,7 +309,7 @@ public class DynamoDbLockRepository implements InitializingBean, DisposableBean,
.withProjectionExpression(KEY_ATTR)
.withMaxResultSize(1)
.withFilterExpression(OWNER_ATTR + " = :owner AND " + TTL_ATTR + " >= :ttl")
.withValueMap(ownerWithCurrentTimeValues());
.withValueMap(ownerWithTtlValues(currentEpochSeconds()));
return this.lockTable.query(querySpec).iterator().hasNext();
}
@@ -355,15 +355,15 @@ public class DynamoDbLockRepository implements InitializingBean, DisposableBean,
new DeleteItemSpec()
.withPrimaryKey(KEY_ATTR, lock)
.withConditionExpression(OWNER_ATTR + " = :owner AND " + TTL_ATTR + " < :ttl")
.withValueMap(ownerWithCurrentTimeValues())));
.withValueMap(ownerWithTtlValues(currentEpochSeconds()))));
this.heldLocks.clear();
}
}
private ValueMap ownerWithCurrentTimeValues() {
private ValueMap ownerWithTtlValues(long epochSeconds) {
ValueMap valueMap =
new ValueMap()
.withNumber(":ttl", currentEpochSeconds());
.withNumber(":ttl", epochSeconds);
valueMap.putAll(this.ownerAttribute);
return valueMap;
}
@@ -375,16 +375,17 @@ public class DynamoDbLockRepository implements InitializingBean, DisposableBean,
*/
public boolean acquire(String lock) {
awaitForActive();
long currentTime = currentEpochSeconds();
PutItemSpec putItemSpec =
new PutItemSpec()
.withItem(
new Item()
.withPrimaryKey(KEY_ATTR, lock)
.withString(OWNER_ATTR, this.owner)
.withLong(CREATED_ATTR, currentEpochSeconds())
.withLong(CREATED_ATTR, currentTime)
.withLong(TTL_ATTR, ttlEpochSeconds()))
.withConditionExpression(LOCK_NOT_EXISTS_EXPRESSION)
.withValueMap(ownerWithCurrentTimeValues());
.withValueMap(ownerWithTtlValues(currentTime));
try {
this.lockTable.putItem(putItemSpec);
this.heldLocks.add(lock);
@@ -393,7 +394,6 @@ public class DynamoDbLockRepository implements InitializingBean, DisposableBean,
catch (ConditionalCheckFailedException ex) {
return false;
}
}
/**
@@ -404,16 +404,12 @@ public class DynamoDbLockRepository implements InitializingBean, DisposableBean,
public boolean renew(String lock) {
awaitForActive();
if (this.heldLocks.contains(lock)) {
ValueMap valueMap =
new ValueMap()
.withNumber(":ttl", ttlEpochSeconds());
valueMap.putAll(this.ownerAttribute);
UpdateItemSpec updateItemSpec =
new UpdateItemSpec()
.withPrimaryKey(KEY_ATTR, lock)
.withUpdateExpression("SET " + TTL_ATTR + " = :ttl")
.withConditionExpression(LOCK_EXISTS_EXPRESSION)
.withValueMap(valueMap);
.withValueMap(ownerWithTtlValues(ttlEpochSeconds()));
try {
this.lockTable.updateItem(updateItemSpec);
return true;

3
src/test/java/org/springframework/integration/aws/LocalstackContainerTest.java
View File

@@ -39,6 +39,7 @@ import org.testcontainers.utility.DockerImageName;
* started only once per JVM, therefore the target Docker container is reused automatically.
*
* @author Artem Bilan
*
* @since 3.0
*/
@Testcontainers(disabledWithoutDocker = true)
@@ -46,7 +47,7 @@ public interface LocalstackContainerTest {
LocalStackContainer LOCAL_STACK_CONTAINER =
new LocalStackContainer(
DockerImageName.parse("localstack/localstack:1.2.0"))
DockerImageName.parse("localstack/localstack:1.4.0"))
.withServices(
LocalStackContainer.Service.DYNAMODB,
LocalStackContainer.Service.KINESIS,

15
src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryTests.java
View File

@@ -17,6 +17,9 @@
package org.springframework.integration.aws.lock;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
@@ -25,12 +28,14 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
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;
import org.assertj.core.data.Percentage;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -341,10 +346,20 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
assertThat(lock.tryLock()).isTrue();
try {
this.dynamoDbLockRepository.setLeaseDuration(Duration.ofSeconds(60));
assertThatNoException().isThrownBy(() -> this.dynamoDbLockRegistry.renewLock("foo"));
String ttl =
DYNAMO_DB.getItem(DynamoDbLockRepository.DEFAULT_TABLE_NAME,
Map.of(DynamoDbLockRepository.KEY_ATTR, new AttributeValue("foo")))
.getItem()
.get(DynamoDbLockRepository.TTL_ATTR).getN();
assertThat(Long.parseLong(ttl))
.isCloseTo(LocalDateTime.now().plusSeconds(60).toEpochSecond(ZoneOffset.UTC),
Percentage.withPercentage(10));
}
finally {
lock.unlock();
this.dynamoDbLockRepository.setLeaseDuration(Duration.ofSeconds(2));
}
}