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

Implement our own DynamoDb lock repository

Browse Source 
The existing `com.amazonaws:dynamodb-lock-client` does not implement
a locking algorithm properly and there is no easy way to determine if
lock has been abandoned according to the current `leaseDuration` behavior.

* Implement `DynamoDbLockRepository` to interact with lock table in this manner:
 - `lockKey` as a primary key
 - `lockOwner` as a unique owner client for the lock
 - `createdAt` just an info when the lock was created by the client
 - `expiredAt` the time in epoch seconds how long the lock is treated as valid
 - this `expiredAt` can be configured as a DynamoDb `TTL` feature
* The `DynamoDbLockRepository` uses a `leaseDuration` to calculate an `expiredAt`
and compares it with the current epoch seconds to see if lock is not valid anymore
* Rework `DynamoDbLockRegistry` to rely on the `DynamoDbLockRepository`
* Remove `com.amazonaws:dynamodb-lock-client` dependency as we don't need it anymore
* Implement `RenewableLockRegistry` contract for simple `expiredAt` update
This commit is contained in:
abilan
2023-02-27 12:00:46 -05:00
parent 2f2675dd6c
commit 594ea58f28
9 changed files with 716 additions and 664 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@@ -45,7 +45,6 @@ These dependencies are optional in the project:
* `com.amazonaws:amazon-kinesis-client` - for KCL-based inbound channel adapter
* `com.amazonaws:amazon-kinesis-producer` - for KPL-based `MessageHandler`
* `com.amazonaws:aws-java-sdk-dynamodb` - for `DynamoDbMetadataStore` and `DynamoDbLockRegistry`
* `com.amazonaws:dynamodb-lock-client` - for `DynamoDbLockRegistry`
Consider to include an appropriate dependency into your project when you use particular component from this project.
@@ -607,7 +606,6 @@ Certain components (for example aggregator and resequencer) use a lock obtained
The `DefaultLockRegistry` performs this function within a single component; you can now configure an external lock registry on these components.
When used with a shared `MessageGroupStore`, the `DynamoDbLockRegistry` can be used to provide this functionality across multiple application instances, such that only one instance can manipulate the group at a time.
This implementation can also be used for the distributed leader elections using a [LockRegistryLeaderInitiator][].
The `com.amazonaws:dynamodb-lock-client` dependency must be present to make a `DynamoDbLockRegistry` working.
[Spring Cloud AWS]: https://awspring.io/
[AWS SDK for Java]: https://aws.amazon.com/sdkforjava/

6
build.gradle
View File

@@ -32,13 +32,12 @@ repositories {
ext {
assertjVersion = '3.23.1'
awaitilityVersion = '4.2.0'
dynamodbLockClientVersion = '1.1.0'
jacksonVersion = '2.14.1'
junitVersion = '5.9.1'
servletApiVersion = '6.0.0'
log4jVersion = '2.19.0'
springCloudAwsVersion = '2.4.2'
springIntegrationVersion = '6.0.1'
springIntegrationVersion = '6.0.3'
kinesisClientVersion = '1.14.9'
kinesisProducerVersion = '0.14.13'
testcontainersVersion = '1.17.6'
@@ -103,7 +102,7 @@ jacoco {
checkstyle {
configDirectory.set(rootProject.file('src/checkstyle'))
toolVersion = '10.5.0'
toolVersion = '10.7.0'
}
dependencies {
@@ -120,7 +119,6 @@ dependencies {
optionalApi 'com.amazonaws:aws-java-sdk-kinesis'
optionalApi 'com.amazonaws:aws-java-sdk-dynamodb'
optionalApi "com.amazonaws:dynamodb-lock-client:$dynamodbLockClientVersion"
optionalApi "jakarta.servlet:jakarta.servlet-api:$servletApiVersion"

2
src/checkstyle/checkstyle.xml
View File

@@ -52,7 +52,7 @@
<module name="CovariantEquals"/>
<module name="EmptyStatement"/>
<module name="EqualsHashCode"/>
<module name="InnerAssignment"/>
<!-- <module name="InnerAssignment"/>-->
<module name="SimplifyBooleanExpression"/>
<module name="SimplifyBooleanReturn"/>
<module name="StringLiteralEquality"/>

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

@@ -74,6 +74,7 @@ import org.springframework.integration.support.AbstractIntegrationMessageBuilder
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.integration.support.locks.LockRegistry;
import org.springframework.integration.support.locks.RenewableLockRegistry;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
@@ -1547,15 +1548,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
Lock lock = this.locks.get(lockFuture.lockKey);
if (lock != null) {
try {
if (lock.tryLock()) {
try {
lockFuture.complete(true);
}
finally {
lock.unlock();
}
}
else {
if (!renewLockInRegistry(lockFuture)) {
lockFuture.complete(false);
this.locks.remove(lockFuture.lockKey);
}
@@ -1600,6 +1593,30 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
}
}
private boolean renewLockInRegistry(LockCompletableFuture renewLockFuture) {
if (KinesisMessageDrivenChannelAdapter.this.lockRegistry instanceof RenewableLockRegistry renewableLockRegistry) {
try {
renewableLockRegistry.renewLock(renewLockFuture.lockKey);
return renewLockFuture.complete(true);
}
catch (IllegalStateException ex) {
return false;
}
}
else {
Lock lock = this.locks.get(renewLockFuture.lockKey);
if (lock.tryLock()) {
try {
return renewLockFuture.complete(true);
}
finally {
lock.unlock();
}
}
}
return false;
}
@Override
public boolean isLongLived() {
return true;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2023 the original author or authors.
* Copyright 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.
@@ -17,342 +17,59 @@
package org.springframework.integration.aws.lock;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.time.Duration;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import com.amazonaws.services.dynamodbv2.AcquireLockOptions;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClientOptions;
import com.amazonaws.services.dynamodbv2.LockItem;
import com.amazonaws.services.dynamodbv2.model.AttributeDefinition;
import com.amazonaws.services.dynamodbv2.model.BillingMode;
import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
import com.amazonaws.services.dynamodbv2.model.KeyType;
import com.amazonaws.services.dynamodbv2.model.LockCurrentlyUnavailableException;
import com.amazonaws.services.dynamodbv2.model.LockNotGrantedException;
import com.amazonaws.services.dynamodbv2.model.LockTableDoesNotExistException;
import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
import com.amazonaws.services.dynamodbv2.model.ResourceInUseException;
import com.amazonaws.services.dynamodbv2.model.ScalarAttributeType;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import com.amazonaws.services.dynamodbv2.model.TransactionConflictException;
import org.springframework.dao.CannotAcquireLockException;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.integration.support.locks.ExpirableLockRegistry;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.integration.support.locks.RenewableLockRegistry;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
* An {@link ExpirableLockRegistry} implementation for the AWS DynamoDB. The algorithm is
* based on the {@link AmazonDynamoDBLockClient}.
* <p>
* Can create table in DynamoDB if an external {@link AmazonDynamoDBLockClient} is not
* provided.
* An {@link ExpirableLockRegistry} and {@link RenewableLockRegistry} implementation for the AWS DynamoDB.
* The algorithm is based on the {@link DynamoDbLockRepository}.
*
* @author Artem Bilan
* @author Karl Lessard
* @author Asiel Caballero
*
* @since 2.0
*/
public class DynamoDbLockRegistry implements ExpirableLockRegistry, InitializingBean, DisposableBean {
public class DynamoDbLockRegistry implements ExpirableLockRegistry, RenewableLockRegistry {
/**
* The {@value DEFAULT_TABLE_NAME} default name for the locks table in the DynamoDB.
*/
public static final String DEFAULT_TABLE_NAME = "SpringIntegrationLockRegistry";
/**
* The {@value DEFAULT_PARTITION_KEY_NAME} default name for the partition key in the
* table.
*/
public static final String DEFAULT_PARTITION_KEY_NAME = "lockKey";
/**
* The {@value DEFAULT_SORT_KEY_NAME} default name for the sort key in the table.
*/
public static final String DEFAULT_SORT_KEY_NAME = "sortKey";
/**
* The {@value DEFAULT_SORT_KEY} default value for the sort key in the table.
*/
public static final String DEFAULT_SORT_KEY = "SpringIntegrationLocks";
/**
* The {@value DEFAULT_REFRESH_PERIOD_MS} default period in milliseconds between DB
* polling requests.
*/
public static final long DEFAULT_REFRESH_PERIOD_MS = 1000L;
private static final Log logger = LogFactory.getLog(DynamoDbLockRegistry.class);
private final ThreadFactory customizableThreadFactory = new CustomizableThreadFactory("dynamodb-lock-registry-");
private static final int DEFAULT_IDLE = 100;
private final Map<String, DynamoDbLock> locks = new ConcurrentHashMap<>();
private final CountDownLatch createTableLatch = new CountDownLatch(1);
private final DynamoDbLockRepository dynamoDbLockRepository;
private final AmazonDynamoDB dynamoDB;
private Duration idleBetweenTries = Duration.ofMillis(DEFAULT_IDLE);
private final String tableName;
private AmazonDynamoDBLockClient dynamoDBLockClient;
private boolean dynamoDBLockClientExplicitlySet;
private BillingMode billingMode = BillingMode.PAY_PER_REQUEST;
private long readCapacity = 1L;
private long writeCapacity = 1L;
private String partitionKey = DEFAULT_PARTITION_KEY_NAME;
private String sortKeyName = DEFAULT_SORT_KEY_NAME;
private String sortKey = DEFAULT_SORT_KEY;
private long refreshPeriod = DEFAULT_REFRESH_PERIOD_MS;
private long leaseDuration = 20L;
private long heartbeatPeriod = 5L;
private volatile boolean initialized;
public DynamoDbLockRegistry(AmazonDynamoDB dynamoDB) {
this(dynamoDB, DEFAULT_TABLE_NAME);
public DynamoDbLockRegistry(DynamoDbLockRepository dynamoDbLockRepository) {
Assert.notNull(dynamoDbLockRepository, "'dynamoDbLockRepository' must not be null");
this.dynamoDbLockRepository = dynamoDbLockRepository;
}
public DynamoDbLockRegistry(AmazonDynamoDB dynamoDB, String tableName) {
Assert.notNull(dynamoDB, "'dynamoDB' must not be null");
Assert.hasText(tableName, "'tableName' must not be empty");
this.dynamoDB = dynamoDB;
this.tableName = tableName;
}
public DynamoDbLockRegistry(AmazonDynamoDBLockClient dynamoDBLockClient) {
Assert.notNull(dynamoDBLockClient, "'dynamoDBLockClient' must not be null");
this.dynamoDBLockClient = dynamoDBLockClient;
this.dynamoDBLockClientExplicitlySet = true;
this.dynamoDB = null;
this.tableName = null;
}
public void setBillingMode(BillingMode billingMode) {
Assert.notNull(billingMode, "'billingMode' must not be null");
this.billingMode = billingMode;
}
public void setReadCapacity(long readCapacity) {
this.readCapacity = readCapacity;
}
public void setWriteCapacity(long writeCapacity) {
this.writeCapacity = writeCapacity;
}
public void setPartitionKey(String partitionKey) {
Assert.hasText(partitionKey, "'partitionKey' must not be empty");
this.partitionKey = partitionKey;
}
/**
* Specify a name of the table attribute which is used as a sort key.
* @param sortKeyName the sort key attribute name to use.
* Specify a {@link Duration} to sleep between lock record insert/update attempts.
* Defaults to 100 milliseconds.
* @param idleBetweenTries the {@link Duration} to sleep between insert/update attempts.
* @since 3.0
*/
public void setSortKeyName(String sortKeyName) {
this.sortKeyName = sortKeyName;
}
/**
* Specify a value for the sort key attribute of the lock item.
* @param sortKey the sort key value to use.
*/
public void setSortKey(String sortKey) {
this.sortKey = sortKey;
}
public void setLeaseDuration(long leaseDuration) {
this.leaseDuration = leaseDuration;
}
/**
* Specify a period in seconds how often send locks renewal requests called heartbeat.
* When the value is less than or equal to {@code 0}, the heartbeat is disabled.
* @param heartbeatPeriod the heartbeat period for background thread to renew locks in DB
*/
public void setHeartbeatPeriod(long heartbeatPeriod) {
this.heartbeatPeriod = heartbeatPeriod;
}
public void setRefreshPeriod(long refreshPeriod) {
this.refreshPeriod = refreshPeriod;
}
/**
* Set the {@link Executor}, where is not provided then a default of cached thread
* pool Executor will be used.
* @param executor the executor service
* @deprecated with no-op in favor of internally created unmanaged threads.
*/
@Deprecated
public void setExecutor(Executor executor) {
}
@Override
public void afterPropertiesSet() {
if (!this.dynamoDBLockClientExplicitlySet) {
AmazonDynamoDBLockClientOptions dynamoDBLockClientOptions = AmazonDynamoDBLockClientOptions
.builder(this.dynamoDB, this.tableName).withPartitionKeyName(this.partitionKey)
.withSortKeyName(this.sortKeyName)
.withCreateHeartbeatBackgroundThread(this.heartbeatPeriod > 0)
.withHeartbeatPeriod(this.heartbeatPeriod)
.withLeaseDuration(this.leaseDuration).build();
this.dynamoDBLockClient = new AmazonDynamoDBLockClient(dynamoDBLockClientOptions);
}
this.leaseDuration = (long) new DirectFieldAccessor(this.dynamoDBLockClient)
.getPropertyValue("leaseDurationInMilliseconds");
this.customizableThreadFactory
.newThread(() -> {
try {
if (!this.dynamoDBLockClientExplicitlySet) {
try {
this.dynamoDBLockClient.assertLockTableExists();
return;
}
catch (LockTableDoesNotExistException e) {
if (logger.isInfoEnabled()) {
logger.info("No table '" + this.tableName + "'. Creating one...");
}
}
createLockTableInDynamoDB();
}
int i = 0;
// We need up to one minute to wait until table is created on AWS.
while (i++ < 60) {
if (this.dynamoDBLockClient.lockTableExists()) {
return;
}
else {
try {
// This is allowed minimum for constant AWS requests.
Thread.sleep(1000);
}
catch (InterruptedException e) {
ReflectionUtils.rethrowRuntimeException(e);
}
}
}
logger.error("Cannot describe DynamoDb table: " + this.tableName);
}
finally {
// Release create table barrier either way.
// If there is an error during creation/description,
// we deffer the actual ResourceNotFoundException to the end-user active
// calls.
this.createTableLatch.countDown();
}
})
.start();
this.initialized = true;
}
/**
* Creates a DynamoDB table with the right schema for it to be used by this locking library.
* The table should be set
* up in advance, because it takes a few minutes for DynamoDB to provision a new instance.
* <p>
* This method is a variation of {@link AmazonDynamoDBLockClient#createLockTableInDynamoDB} to support custom
* {@link BillingMode} for the lock table.
* <p>
* If table already exists no exception.
*/
private void createLockTableInDynamoDB() {
try {
KeySchemaElement partitionKeyElement = new KeySchemaElement();
partitionKeyElement.setAttributeName(this.partitionKey);
partitionKeyElement.setKeyType(KeyType.HASH);
List<KeySchemaElement> keySchema = new ArrayList<>();
keySchema.add(partitionKeyElement);
Collection<AttributeDefinition> attributeDefinitions = new ArrayList<>();
attributeDefinitions.add(new AttributeDefinition().withAttributeName(this.partitionKey)
.withAttributeType(ScalarAttributeType.S));
KeySchemaElement sortKeyElement = new KeySchemaElement();
sortKeyElement.setAttributeName(this.sortKeyName);
sortKeyElement.setKeyType(KeyType.RANGE);
keySchema.add(sortKeyElement);
attributeDefinitions.add(new AttributeDefinition().withAttributeName(this.sortKeyName)
.withAttributeType(ScalarAttributeType.S));
CreateTableRequest createTableRequest = new CreateTableRequest(this.tableName, keySchema)
.withAttributeDefinitions(attributeDefinitions)
.withBillingMode(this.billingMode);
if (BillingMode.PROVISIONED.equals(this.billingMode)) {
createTableRequest.setProvisionedThroughput(
new ProvisionedThroughput(this.readCapacity, this.writeCapacity));
}
this.dynamoDB.createTable(createTableRequest);
}
catch (ResourceInUseException ex) {
// Swallow an exception and you should check for table existence
}
}
private void awaitForActive() {
Assert.state(this.initialized,
() -> "The component has not been initialized: " + this + ".\n Is it declared as a bean?");
IllegalStateException illegalStateException = new IllegalStateException(
"The DynamoDb table " + this.tableName + " has not been created during " + 60 + " seconds");
try {
if (!this.createTableLatch.await(60, TimeUnit.SECONDS)) {
throw illegalStateException;
}
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw illegalStateException;
}
}
@Override
public void destroy() throws Exception {
if (!this.dynamoDBLockClientExplicitlySet) {
this.dynamoDBLockClient.close();
}
public void setIdleBetweenTries(Duration idleBetweenTries) {
Assert.notNull(idleBetweenTries, "'idleBetweenTries' must not be null");
this.idleBetweenTries = idleBetweenTries;
}
@Override
@@ -363,24 +80,33 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing
@Override
public void expireUnusedOlderThan(long age) {
Iterator<Map.Entry<String, DynamoDbLock>> iterator = this.locks.entrySet().iterator();
long now = System.currentTimeMillis();
while (iterator.hasNext()) {
Map.Entry<String, DynamoDbLock> entry = iterator.next();
DynamoDbLock lock = entry.getValue();
if (now - lock.lastUsed > age && !lock.delegate.isHeldByCurrentThread()) {
iterator.remove();
}
synchronized (this.locks) {
this.locks.entrySet()
.removeIf(entry -> {
DynamoDbLock lock = entry.getValue();
return now - lock.lastUsed > age && !lock.isAcquiredInThisProcess();
});
}
}
@Override
public void renewLock(Object lockKey) {
Assert.isInstanceOf(String.class, lockKey, "'lockKey' must of String type");
String lockId = (String) lockKey;
DynamoDbLock dynamoDbLock = this.locks.get(lockId);
if (dynamoDbLock == null) {
throw new IllegalStateException("Could not found mutex at " + lockId);
}
if (!dynamoDbLock.renew()) {
throw new IllegalStateException("Could not renew mutex at " + lockId);
}
}
@Override
public String toString() {
return "DynamoDbLockRegistry{" + "tableName='" + this.tableName + '\'' + ", billingMode=" + this.billingMode
+ ", readCapacity=" + this.readCapacity + ", writeCapacity=" + this.writeCapacity + ", partitionKey='"
+ this.partitionKey + '\'' + ", sortKeyName='" + this.sortKeyName + '\'' + ", sortKey='" + this.sortKey
+ '\'' + ", refreshPeriod=" + this.refreshPeriod + ", leaseDuration=" + this.leaseDuration
+ ", heartbeatPeriod=" + this.heartbeatPeriod + '}';
return "DynamoDbLockRegistry{" + "tableName='" + this.dynamoDbLockRepository.getTableName() + '\''
+ ", owner='" + this.dynamoDbLockRepository.getOwner() + '}';
}
private final class DynamoDbLock implements Lock {
@@ -389,18 +115,10 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing
private final String key;
// It is safe to use a shared instance - access is guaranteed by the delegate
// lock.
private final AcquireLockOptions.AcquireLockOptionsBuilder acquireLockOptionsBuilder;
private LockItem lockItem;
private volatile long lastUsed = System.currentTimeMillis();
private DynamoDbLock(String key) {
this.key = key;
this.acquireLockOptionsBuilder = AcquireLockOptions.builder(this.key).withReplaceData(false)
.withSortKey(DynamoDbLockRegistry.this.sortKey).withTimeUnit(TimeUnit.MILLISECONDS);
}
private void rethrowAsLockException(Exception e) {
@@ -409,74 +127,57 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing
@Override
public void lock() {
awaitForActive();
this.delegate.lock();
setupDefaultAcquireLockOptionsBuilder();
boolean wasInterruptedWhileUninterruptible = false;
try {
while (true) {
try {
while (!doLock()) {
Thread.sleep(100); // NOSONAR
}
break;
}
catch (InterruptedException e) {
/*
* This method must be uninterruptible so catch and ignore
* interrupts and only break out of the while loop when we get the
* lock.
*/
wasInterruptedWhileUninterruptible = true;
}
catch (Exception e) {
this.delegate.unlock();
rethrowAsLockException(e);
while (true) {
try {
while (!doLock()) {
sleepBetweenRetries();
}
break;
}
}
finally {
if (wasInterruptedWhileUninterruptible) {
Thread.currentThread().interrupt();
catch (TransactionConflictException ex) {
// try again
}
catch (InterruptedException ex) {
/*
* This method must be uninterruptible so catch and ignore
* interrupts and only break out of the while loop when
* we get the lock.
*/
}
catch (Exception ex) {
this.delegate.unlock();
rethrowAsLockException(ex);
}
}
}
private void setupDefaultAcquireLockOptionsBuilder() {
this.acquireLockOptionsBuilder
.withAdditionalTimeToWaitForLock(Long.MAX_VALUE - DynamoDbLockRegistry.this.leaseDuration)
.withRefreshPeriod(DynamoDbLockRegistry.this.refreshPeriod);
}
@Override
public void lockInterruptibly() throws InterruptedException {
awaitForActive();
this.delegate.lockInterruptibly();
setupDefaultAcquireLockOptionsBuilder();
try {
while (!doLock()) {
Thread.sleep(100); // NOSONAR
if (Thread.currentThread().isInterrupted()) {
throw new InterruptedException();
while (true) {
try {
while (!doLock()) {
sleepBetweenRetries();
if (Thread.currentThread().isInterrupted()) {
throw new InterruptedException();
}
}
break;
}
catch (TransactionConflictException ex) {
// try again
}
catch (InterruptedException ie) {
this.delegate.unlock();
Thread.currentThread().interrupt();
throw ie;
}
catch (Exception e) {
this.delegate.unlock();
rethrowAsLockException(e);
}
}
catch (InterruptedException ie) {
this.delegate.unlock();
Thread.currentThread().interrupt();
throw ie;
}
catch (Exception e) {
this.delegate.unlock();
rethrowAsLockException(e);
}
}
@@ -485,7 +186,7 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing
try {
return tryLock(0, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e) {
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
return false;
}
@@ -493,100 +194,117 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
long start = System.currentTimeMillis();
awaitForActive();
long now = System.currentTimeMillis();
if (!this.delegate.tryLock(time, unit)) {
return false;
}
long additionalTimeToWait =
TimeUnit.MILLISECONDS.convert(time, unit)
- System.currentTimeMillis() + start - DynamoDbLockRegistry.this.leaseDuration;
this.acquireLockOptionsBuilder.withAdditionalTimeToWaitForLock(additionalTimeToWait);
boolean acquired = false;
try {
acquired = doLock();
if (!acquired) {
long expire = now + TimeUnit.MILLISECONDS.convert(time, unit);
boolean acquired;
while (true) {
try {
while (!(acquired = doLock()) && System.currentTimeMillis() < expire) { //NOSONAR
sleepBetweenRetries();
}
if (!acquired) {
this.delegate.unlock();
}
return acquired;
}
catch (TransactionConflictException ex) {
// try again
}
catch (Exception ex) {
this.delegate.unlock();
}
else {
this.lastUsed = System.currentTimeMillis();
rethrowAsLockException(ex);
}
}
catch (LockCurrentlyUnavailableException ex) {
this.delegate.unlock();
logger.trace("The lock '" + this + "' cannot be acquired at the moment", ex);
}
catch (Exception e) {
this.delegate.unlock();
rethrowAsLockException(e);
}
return acquired;
}
private boolean doLock() throws InterruptedException {
boolean acquired = false;
if (this.lockItem != null) {
try {
this.lockItem.sendHeartBeat();
acquired = true;
}
catch (LockNotGrantedException ex) {
// May be no lock record in the DB - discard local holder and try to lock again
this.lockItem = null;
}
}
if (this.lockItem == null) {
this.lockItem = DynamoDbLockRegistry.this.dynamoDBLockClient
.tryAcquireLock(this.acquireLockOptionsBuilder.build()).orElse(null);
acquired = this.lockItem != null;
}
boolean acquired = DynamoDbLockRegistry.this.dynamoDbLockRepository.acquire(this.key);
if (acquired) {
this.lastUsed = System.currentTimeMillis();
}
return acquired;
}
@Override
public void unlock() {
if (!this.delegate.isHeldByCurrentThread()) {
throw new IllegalMonitorStateException("You do not own lock at " + this.key);
throw new IllegalMonitorStateException("The current thread doesn't own mutex at '" + this.key + "'");
}
if (this.delegate.getHoldCount() > 1) {
this.delegate.unlock();
return;
}
try {
if (Thread.currentThread().isInterrupted()) {
LockItem lockItemToRelease = this.lockItem;
DynamoDbLockRegistry.this.customizableThreadFactory
.newThread(() ->
DynamoDbLockRegistry.this.dynamoDBLockClient.releaseLock(lockItemToRelease))
.start();
while (true) {
try {
DynamoDbLockRegistry.this.dynamoDbLockRepository.delete(this.key);
return;
}
catch (TransactionConflictException ex) {
// try again
try {
sleepBetweenRetries();
}
catch (InterruptedException intEx) {
/*
* This method must be uninterruptible so catch and ignore
* interrupts and only break out of the while loop when
* we get 'renewed' result.
*/
}
}
catch (Exception ex) {
throw new DataAccessResourceFailureException("Failed to release mutex at " + this.key, ex);
}
}
else {
DynamoDbLockRegistry.this.dynamoDBLockClient.releaseLock(this.lockItem);
}
}
catch (Exception e) {
throw new DataAccessResourceFailureException("Failed to release lock at " + this.key, e);
}
finally {
this.lockItem = null;
this.delegate.unlock();
}
}
public boolean renew() {
if (!this.delegate.isHeldByCurrentThread()) {
throw new IllegalMonitorStateException("The current thread doesn't own mutex at " + this.key);
}
while (true) {
try {
boolean renewed = DynamoDbLockRegistry.this.dynamoDbLockRepository.renew(this.key);
if (renewed) {
this.lastUsed = System.currentTimeMillis();
}
return renewed;
}
catch (TransactionConflictException ex) {
// try again
try {
sleepBetweenRetries();
}
catch (InterruptedException intEx) {
/*
* This method must be uninterruptible so catch and ignore
* interrupts and only break out of the while loop when
* we get 'renewed' result.
*/
}
}
catch (Exception ex) {
throw new DataAccessResourceFailureException("Failed to renew mutex at " + this.key, ex);
}
}
}
public boolean isAcquiredInThisProcess() {
return DynamoDbLockRegistry.this.dynamoDbLockRepository.isAcquired(this.key);
}
private void sleepBetweenRetries() throws InterruptedException {
Thread.sleep(DynamoDbLockRegistry.this.idleBetweenTries.toMillis());
}
@Override
public Condition newCondition() {
throw new UnsupportedOperationException("DynamoDb locks don't support conditions.");
@@ -595,8 +313,7 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing
@Override
public String toString() {
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd@HH:mm:ss.SSS");
return "DynamoDbLock [lockKey=" + this.key + ",lockedAt=" + dateFormat.format(new Date(this.lastUsed))
+ ", lockItem=" + this.lockItem + "]";
return "DynamoDbLock [lockKey=" + this.key + ",lockedAt=" + dateFormat.format(new Date(this.lastUsed)) + "]";
}
}

445
src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRepository.java Normal file
View File

@@ -0,0 +1,445 @@
/*
* 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.
* You may obtain a copy of the License at
*
* https://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.lock;
import java.io.Closeable;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
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.PutItemSpec;
import com.amazonaws.services.dynamodbv2.document.spec.QuerySpec;
import com.amazonaws.services.dynamodbv2.document.spec.UpdateItemSpec;
import com.amazonaws.services.dynamodbv2.document.utils.ValueMap;
import com.amazonaws.services.dynamodbv2.model.AttributeDefinition;
import com.amazonaws.services.dynamodbv2.model.BillingMode;
import com.amazonaws.services.dynamodbv2.model.ConditionalCheckFailedException;
import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
import com.amazonaws.services.dynamodbv2.model.DescribeTableRequest;
import com.amazonaws.services.dynamodbv2.model.DescribeTableResult;
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.ResourceInUseException;
import com.amazonaws.services.dynamodbv2.model.ResourceNotFoundException;
import com.amazonaws.services.dynamodbv2.model.ScalarAttributeType;
import com.amazonaws.services.dynamodbv2.model.TableStatus;
import com.amazonaws.services.dynamodbv2.model.TimeToLiveSpecification;
import com.amazonaws.services.dynamodbv2.model.UpdateTimeToLiveRequest;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
* Encapsulation of the DynamoDB shunting that is needed for locks.
* <p>
* The DynamoDb table must have these attributes:
* <ul>
* <li> {@link DynamoDbLockRepository#KEY_ATTR} {@link ScalarAttributeType#S} - partition key {@link KeyType#HASH}
* <li> {@link DynamoDbLockRepository#OWNER_ATTR} {@link ScalarAttributeType#S}
* <li> {@link DynamoDbLockRepository#CREATED_ATTR} {@link ScalarAttributeType#N}
* <li> {@link DynamoDbLockRepository#TTL_ATTR} {@link ScalarAttributeType#N}
* </ul>
*
* @author Artem Bilan
*
* @since 3.0
*/
public class DynamoDbLockRepository implements InitializingBean, DisposableBean, Closeable {
/**
* The {@value DEFAULT_TABLE_NAME} default name for the locks table in the DynamoDB.
*/
public static final String DEFAULT_TABLE_NAME = "SpringIntegrationLockRegistry";
/**
* The {@value KEY_ATTR} name for the partition key in the table.
*/
public static final String KEY_ATTR = "lockKey";
/**
* The {@value OWNER_ATTR} name for the owner of lock in the table.
*/
public static final String OWNER_ATTR = "lockOwner";
/**
* The {@value CREATED_ATTR} date for lock item.
*/
public static final String CREATED_ATTR = "createdAt";
/**
* The {@value TTL_ATTR} for how long the lock is valid.
*/
public static final String TTL_ATTR = "expireAt";
private static final String LOCK_EXISTS_EXPRESSION =
String.format("attribute_exists(%s) AND %s = :owner", KEY_ATTR, OWNER_ATTR);
private static final String LOCK_NOT_EXISTS_EXPRESSION =
String.format("attribute_not_exists(%s) OR %s < :ttl OR (%s)", KEY_ATTR, TTL_ATTR, LOCK_EXISTS_EXPRESSION);
/**
* Default value for the {@link #leaseDuration} property.
*/
public static final Duration DEFAULT_LEASE_DURATION = Duration.ofSeconds(10);
private static final Log LOGGER = LogFactory.getLog(DynamoDbLockRegistry.class);
private final ThreadFactory customizableThreadFactory = new CustomizableThreadFactory("dynamodb-lock-registry-");
private final CountDownLatch createTableLatch = new CountDownLatch(1);
private final Set<String> heldLocks = Collections.synchronizedSet(new HashSet<>());
private final AmazonDynamoDB dynamoDB;
private final Table lockTable;
private BillingMode billingMode = BillingMode.PAY_PER_REQUEST;
private long readCapacity = 1L;
private long writeCapacity = 1L;
private String owner = UUID.randomUUID().toString();
private Duration leaseDuration = DEFAULT_LEASE_DURATION;
private Map<String, Object> ownerAttribute;
private volatile boolean initialized;
public DynamoDbLockRepository(AmazonDynamoDB dynamoDB) {
this(dynamoDB, DEFAULT_TABLE_NAME);
}
public DynamoDbLockRepository(AmazonDynamoDB dynamoDB, String tableName) {
this.dynamoDB = dynamoDB;
this.lockTable = new Table(this.dynamoDB, tableName);
}
public void setBillingMode(BillingMode billingMode) {
Assert.notNull(billingMode, "'billingMode' must not be null");
this.billingMode = billingMode;
}
public void setReadCapacity(long readCapacity) {
this.readCapacity = readCapacity;
}
public void setWriteCapacity(long writeCapacity) {
this.writeCapacity = writeCapacity;
}
/**
* Specify a custom client id (owner) for locks in DB.
* Must be unique per cluster to avoid interlocking between different instances.
* @param owner the client id to be associated with locks handled by the repository.
*/
public void setOwner(String owner) {
this.owner = owner;
}
/**
* How long to hold the item after last update.
* @param leaseDuration the duration for how long to keep the lock after last update.
*/
public void setLeaseDuration(Duration leaseDuration) {
this.leaseDuration = leaseDuration;
}
public String getTableName() {
return this.lockTable.getTableName();
}
public String getOwner() {
return this.owner;
}
@Override
public void afterPropertiesSet() {
this.customizableThreadFactory
.newThread(() -> {
try {
if (!lockTableExists()) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("No table '" + getTableName() + "'. Creating one...");
}
createLockTableInDynamoDB();
int i = 0;
// We need up to one minute to wait until table is created on AWS.
while (i++ < 60) {
if (lockTableExists()) {
this.dynamoDB.updateTimeToLive(
new UpdateTimeToLiveRequest()
.withTableName(getTableName())
.withTimeToLiveSpecification(
new TimeToLiveSpecification()
.withEnabled(true)
.withAttributeName(TTL_ATTR)));
return;
}
else {
try {
// This is allowed minimum for constant AWS requests.
Thread.sleep(1000);
}
catch (InterruptedException e) {
ReflectionUtils.rethrowRuntimeException(e);
}
}
}
LOGGER.error("Cannot describe DynamoDb table: " + getTableName());
}
}
finally {
// Release create table barrier either way.
// If there is an error during creation/description,
// we defer the actual ResourceNotFoundException to the end-user active
// calls.
this.createTableLatch.countDown();
}
})
.start();
this.ownerAttribute = Map.of(":owner", this.owner);
this.initialized = true;
}
private boolean lockTableExists() {
try {
DescribeTableResult result = this.dynamoDB.describeTable(new DescribeTableRequest(getTableName()));
return Set.of(TableStatus.ACTIVE, TableStatus.UPDATING)
.contains(TableStatus.fromValue(result.getTable().getTableStatus()));
}
catch (ResourceNotFoundException e) {
// This exception indicates the table doesn't exist.
return false;
}
}
/**
* Creates a DynamoDB table with the right schema for it to be used by this locking library.
* The table should be set up in advance,
* because it takes a few minutes for DynamoDB to provision a new instance.
* If table already exists no exception.
*/
private void createLockTableInDynamoDB() {
try {
CreateTableRequest createTableRequest =
new CreateTableRequest()
.withTableName(getTableName())
.withKeySchema(new KeySchemaElement(KEY_ATTR, KeyType.HASH))
.withAttributeDefinitions(new AttributeDefinition(KEY_ATTR, ScalarAttributeType.S))
.withBillingMode(this.billingMode);
if (BillingMode.PROVISIONED.equals(this.billingMode)) {
createTableRequest.setProvisionedThroughput(
new ProvisionedThroughput(this.readCapacity, this.writeCapacity));
}
this.dynamoDB.createTable(createTableRequest);
}
catch (ResourceInUseException ex) {
// Swallow an exception and you should check for table existence
}
}
private void awaitForActive() {
Assert.state(this.initialized,
() -> "The component has not been initialized: " + this + ".\n Is it declared as a bean?");
IllegalStateException illegalStateException = new IllegalStateException(
"The DynamoDb table " + getTableName() + " has not been created during " + 60 + " seconds");
try {
if (!this.createTableLatch.await(60, TimeUnit.SECONDS)) {
throw illegalStateException;
}
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw illegalStateException;
}
}
/**
* Check if a lock is held by this repository.
* @param lock the lock to check.
* @return acquired or not.
*/
public boolean isAcquired(String lock) {
awaitForActive();
if (this.heldLocks.contains(lock)) {
QuerySpec querySpec =
new QuerySpec()
.withHashKey(KEY_ATTR, lock)
.withProjectionExpression(KEY_ATTR)
.withMaxResultSize(1)
.withFilterExpression(OWNER_ATTR + " = :owner AND " + TTL_ATTR + " >= :ttl")
.withValueMap(ownerWithCurrentTimeValues());
return this.lockTable.query(querySpec).iterator().hasNext();
}
return false;
}
/**
* Remove a lock from this repository.
* @param lock the lock to remove.
*/
public void delete(String lock) {
awaitForActive();
if (this.heldLocks.remove(lock)) {
deleteFromDb(lock);
}
}
private void deleteFromDb(String lock) {
doDelete(
new DeleteItemSpec()
.withPrimaryKey(KEY_ATTR, lock)
.withConditionExpression(OWNER_ATTR + " = :owner")
.withValueMap(this.ownerAttribute));
}
private void doDelete(DeleteItemSpec deleteItemSpec) {
try {
this.lockTable.deleteItem(deleteItemSpec);
}
catch (ConditionalCheckFailedException ex) {
// Ignore - assuming no record in DB anymore.
}
}
/**
* Remove all the expired locks.
*/
public void deleteExpired() {
awaitForActive();
synchronized (this.heldLocks) {
this.heldLocks.forEach((lock) ->
doDelete(
new DeleteItemSpec()
.withPrimaryKey(KEY_ATTR, lock)
.withConditionExpression(OWNER_ATTR + " = :owner AND " + TTL_ATTR + " < :ttl")
.withValueMap(ownerWithCurrentTimeValues())));
this.heldLocks.clear();
}
}
private ValueMap ownerWithCurrentTimeValues() {
ValueMap valueMap =
new ValueMap()
.withNumber(":ttl", currentEpochSeconds());
valueMap.putAll(this.ownerAttribute);
return valueMap;
}
/**
* Acquire a lock for a key.
* @param lock the key for lock to acquire.
* @return acquired or not.
*/
public boolean acquire(String lock) {
awaitForActive();
PutItemSpec putItemSpec =
new PutItemSpec()
.withItem(
new Item()
.withPrimaryKey(KEY_ATTR, lock)
.withString(OWNER_ATTR, this.owner)
.withLong(CREATED_ATTR, currentEpochSeconds())
.withLong(TTL_ATTR, ttlEpochSeconds()))
.withConditionExpression(LOCK_NOT_EXISTS_EXPRESSION)
.withValueMap(ownerWithCurrentTimeValues());
try {
this.lockTable.putItem(putItemSpec);
this.heldLocks.add(lock);
return true;
}
catch (ConditionalCheckFailedException ex) {
return false;
}
}
/**
* Renew the lease for a lock.
* @param lock the lock to renew.
* @return renewed or not.
*/
public boolean renew(String lock) {
awaitForActive();
if (this.heldLocks.contains(lock)) {
UpdateItemSpec updateItemSpec =
new UpdateItemSpec()
.withPrimaryKey(KEY_ATTR, lock)
.withUpdateExpression("SET " + TTL_ATTR + " = :ttl")
.withConditionExpression(LOCK_EXISTS_EXPRESSION)
.withValueMap(ownerWithCurrentTimeValues());
try {
this.lockTable.updateItem(updateItemSpec);
return true;
}
catch (ConditionalCheckFailedException ex) {
return false;
}
}
return false;
}
@Override
public void destroy() {
close();
}
@Override
public void close() {
synchronized (this.heldLocks) {
this.heldLocks.forEach(this::deleteFromDb);
this.heldLocks.clear();
}
}
private long ttlEpochSeconds() {
return LocalDateTime.now().plus(this.leaseDuration).toEpochSecond(ZoneOffset.UTC);
}
private static long currentEpochSeconds() {
return LocalDateTime.now().toEpochSecond(ZoneOffset.UTC);
}
}

33
src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2022 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.
@@ -16,6 +16,7 @@
package org.springframework.integration.aws.leader;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
@@ -35,6 +36,7 @@ import org.junit.jupiter.api.Test;
import org.springframework.integration.aws.LocalstackContainerTest;
import org.springframework.integration.aws.lock.DynamoDbLockRegistry;
import org.springframework.integration.aws.lock.DynamoDbLockRepository;
import org.springframework.integration.leader.Context;
import org.springframework.integration.leader.DefaultCandidate;
import org.springframework.integration.leader.event.LeaderEventPublisher;
@@ -56,11 +58,11 @@ class DynamoDbLockRegistryLeaderInitiatorTests implements LocalstackContainerTes
static void init() {
DYNAMO_DB = LocalstackContainerTest.dynamoDbClient();
try {
DYNAMO_DB.deleteTableAsync(DynamoDbLockRegistry.DEFAULT_TABLE_NAME);
DYNAMO_DB.deleteTableAsync(DynamoDbLockRepository.DEFAULT_TABLE_NAME);
Waiter<DescribeTableRequest> waiter = DYNAMO_DB.waiters().tableNotExists();
waiter.run(new WaiterParameters<>(new DescribeTableRequest(DynamoDbLockRegistry.DEFAULT_TABLE_NAME))
waiter.run(new WaiterParameters<>(new DescribeTableRequest(DynamoDbLockRepository.DEFAULT_TABLE_NAME))
.withPollingStrategy(
new PollingStrategy(new MaxAttemptsRetryStrategy(25), new FixedDelayStrategy(1))));
}
@@ -71,19 +73,21 @@ class DynamoDbLockRegistryLeaderInitiatorTests implements LocalstackContainerTes
@AfterAll
static void destroy() {
DYNAMO_DB.deleteTable(DynamoDbLockRegistry.DEFAULT_TABLE_NAME);
DYNAMO_DB.deleteTable(DynamoDbLockRepository.DEFAULT_TABLE_NAME);
}
@Test
void testDistributedLeaderElection() throws Exception {
CountDownLatch granted = new CountDownLatch(1);
CountingPublisher countingPublisher = new CountingPublisher(granted);
List<DynamoDbLockRegistry> registries = new ArrayList<>();
List<DynamoDbLockRepository> repositories = new ArrayList<>();
List<LockRegistryLeaderInitiator> initiators = new ArrayList<>();
for (int i = 0; i < 2; i++) {
DynamoDbLockRegistry lockRepository = new DynamoDbLockRegistry(DYNAMO_DB);
lockRepository.afterPropertiesSet();
registries.add(lockRepository);
DynamoDbLockRepository dynamoDbLockRepository = new DynamoDbLockRepository(DYNAMO_DB);
dynamoDbLockRepository.setLeaseDuration(Duration.ofSeconds(1));
dynamoDbLockRepository.afterPropertiesSet();
repositories.add(dynamoDbLockRepository);
DynamoDbLockRegistry lockRepository = new DynamoDbLockRegistry(dynamoDbLockRepository);
LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(lockRepository,
new DefaultCandidate("foo#" + i, "bar"));
@@ -162,8 +166,8 @@ class DynamoDbLockRegistryLeaderInitiatorTests implements LocalstackContainerTes
initiator1.stop();
for (DynamoDbLockRegistry registry : registries) {
registry.destroy();
for (DynamoDbLockRepository dynamoDbLockRepository : repositories) {
dynamoDbLockRepository.close();
}
}
@@ -172,8 +176,9 @@ class DynamoDbLockRegistryLeaderInitiatorTests implements LocalstackContainerTes
CountDownLatch granted = new CountDownLatch(1);
CountingPublisher countingPublisher = new CountingPublisher(granted);
DynamoDbLockRegistry lockRepository = new DynamoDbLockRegistry(DYNAMO_DB);
lockRepository.afterPropertiesSet();
DynamoDbLockRepository dynamoDbLockRepository = new DynamoDbLockRepository(DYNAMO_DB);
dynamoDbLockRepository.afterPropertiesSet();
DynamoDbLockRegistry lockRepository = new DynamoDbLockRegistry(dynamoDbLockRepository);
LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(lockRepository);
initiator.setLeaderEventPublisher(countingPublisher);
@@ -192,13 +197,13 @@ class DynamoDbLockRegistryLeaderInitiatorTests implements LocalstackContainerTes
init();
lockRepository.afterPropertiesSet();
dynamoDbLockRepository.afterPropertiesSet();
assertThat(granted.await(20, TimeUnit.SECONDS)).isTrue();
initiator.stop();
lockRepository.destroy();
dynamoDbLockRepository.close();
}
private static class CountingPublisher implements LeaderEventPublisher {

123
src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryBuildTableTests.java
View File

@@ -1,123 +0,0 @@
/*
* Copyright 2020-2022 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
*
* https://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.lock;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import com.amazonaws.services.dynamodbv2.AbstractAmazonDynamoDB;
import com.amazonaws.services.dynamodbv2.model.BillingMode;
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.DescribeTableResult;
import com.amazonaws.services.dynamodbv2.model.GetItemRequest;
import com.amazonaws.services.dynamodbv2.model.GetItemResult;
import com.amazonaws.services.dynamodbv2.model.PutItemRequest;
import com.amazonaws.services.dynamodbv2.model.PutItemResult;
import com.amazonaws.services.dynamodbv2.model.ResourceNotFoundException;
import com.amazonaws.services.dynamodbv2.model.TableDescription;
import com.amazonaws.services.dynamodbv2.model.TableStatus;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Asiel Caballero
*
* @since 2.3.5
*/
class DynamoDbLockRegistryBuildTableTests {
private static final String TEST_TABLE
= "testLockRegistry" + DynamoDbLockRegistryBuildTableTests.class.getSimpleName();
private InMemoryAmazonDynamoDB client;
@BeforeEach
void setup() {
this.client = new InMemoryAmazonDynamoDB();
}
@Test
void onDemandIsSetup() throws InterruptedException {
assertsBillingMode(BillingMode.PAY_PER_REQUEST,
lockRegistry -> lockRegistry.setBillingMode(BillingMode.PAY_PER_REQUEST));
}
@Test
void provisionedIsSetup() throws InterruptedException {
assertsBillingMode(BillingMode.PROVISIONED,
lockRegistry -> lockRegistry.setBillingMode(BillingMode.PROVISIONED));
}
@Test
void defaultsToProvisioned() throws InterruptedException {
assertsBillingMode(BillingMode.PAY_PER_REQUEST, store -> { });
}
private void assertsBillingMode(com.amazonaws.services.dynamodbv2.model.BillingMode billingMode,
Consumer<DynamoDbLockRegistry> propertySetter) throws InterruptedException {
DynamoDbLockRegistry lockRegistry = new DynamoDbLockRegistry(this.client, TEST_TABLE);
propertySetter.accept(lockRegistry);
lockRegistry.afterPropertiesSet();
lockRegistry.obtain("test").tryLock(1, TimeUnit.SECONDS);
assertThat(billingMode.toString())
.isEqualTo(this.client.getCreateTableRequest().getBillingMode());
}
private static class InMemoryAmazonDynamoDB extends AbstractAmazonDynamoDB {
private CreateTableRequest createTableRequest;
private boolean wasCalled = false;
@Override
public synchronized CreateTableResult createTable(CreateTableRequest request) {
this.createTableRequest = request;
return null;
}
@Override
public GetItemResult getItem(GetItemRequest request) {
return new GetItemResult();
}
@Override
public PutItemResult putItem(PutItemRequest request) {
return new PutItemResult();
}
@Override
public synchronized DescribeTableResult describeTable(DescribeTableRequest request) {
if (this.wasCalled) {
return new DescribeTableResult()
.withTable(new TableDescription()
.withTableStatus(TableStatus.ACTIVE));
}
else {
this.wasCalled = true;
throw new ResourceNotFoundException(TEST_TABLE);
}
}
public synchronized CreateTableRequest getCreateTableRequest() {
return this.createTableRequest;
}
}
}

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

@@ -16,19 +16,15 @@
package org.springframework.integration.aws.lock;
import java.lang.reflect.Method;
import java.util.Map;
import java.time.Duration;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import com.amazonaws.services.dynamodbv2.AcquireLockOptions;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClientOptions;
import com.amazonaws.services.dynamodbv2.LockItem;
import com.amazonaws.services.dynamodbv2.model.DescribeTableRequest;
import com.amazonaws.waiters.FixedDelayStrategy;
import com.amazonaws.waiters.MaxAttemptsRetryStrategy;
@@ -48,9 +44,9 @@ import org.springframework.integration.aws.LocalstackContainerTest;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.ReflectionUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNoException;
/**
* @author Artem Bilan
@@ -65,6 +61,9 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
private final AsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
@Autowired
private DynamoDbLockRepository dynamoDbLockRepository;
@Autowired
private DynamoDbLockRegistry dynamoDbLockRegistry;
@@ -72,11 +71,11 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
static void setup() {
DYNAMO_DB = LocalstackContainerTest.dynamoDbClient();
try {
DYNAMO_DB.deleteTableAsync(DynamoDbLockRegistry.DEFAULT_TABLE_NAME);
DYNAMO_DB.deleteTableAsync(DynamoDbLockRepository.DEFAULT_TABLE_NAME);
Waiter<DescribeTableRequest> waiter = DYNAMO_DB.waiters().tableNotExists();
waiter.run(new WaiterParameters<>(new DescribeTableRequest(DynamoDbLockRegistry.DEFAULT_TABLE_NAME))
waiter.run(new WaiterParameters<>(new DescribeTableRequest(DynamoDbLockRepository.DEFAULT_TABLE_NAME))
.withPollingStrategy(
new PollingStrategy(new MaxAttemptsRetryStrategy(25), new FixedDelayStrategy(1))));
}
@@ -87,7 +86,7 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
@BeforeEach
void clear() {
this.dynamoDbLockRegistry.expireUnusedOlderThan(0);
this.dynamoDbLockRepository.close();
}
@Test
@@ -97,7 +96,7 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
Lock lock = this.dynamoDbLockRegistry.obtain("foo");
lock.lock();
try {
assertThat(TestUtils.getPropertyValue(this.dynamoDbLockRegistry, "locks", Map.class)).hasSize(1);
assertThat(TestUtils.getPropertyValue(this.dynamoDbLockRepository, "heldLocks", Set.class)).hasSize(1);
}
finally {
lock.unlock();
@@ -112,7 +111,7 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
Lock lock = this.dynamoDbLockRegistry.obtain("foo");
lock.lockInterruptibly();
try {
assertThat(TestUtils.getPropertyValue(this.dynamoDbLockRegistry, "locks", Map.class)).hasSize(1);
assertThat(TestUtils.getPropertyValue(this.dynamoDbLockRepository, "heldLocks", Set.class)).hasSize(1);
}
finally {
lock.unlock();
@@ -178,11 +177,11 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch = new CountDownLatch(1);
Future<Object> result = this.taskExecutor.submit(() -> {
DynamoDbLockRegistry registry2 = new DynamoDbLockRegistry(DYNAMO_DB);
registry2.setHeartbeatPeriod(1);
registry2.setRefreshPeriod(10);
registry2.setLeaseDuration(2);
registry2.afterPropertiesSet();
DynamoDbLockRepository dynamoDbLockRepository = new DynamoDbLockRepository(DYNAMO_DB);
dynamoDbLockRepository.setLeaseDuration(Duration.ofSeconds(10));
dynamoDbLockRepository.afterPropertiesSet();
DynamoDbLockRegistry registry2 = new DynamoDbLockRegistry(dynamoDbLockRepository);
registry2.setIdleBetweenTries(Duration.ofMillis(10));
Lock lock2 = registry2.obtain("foo");
locked.set(lock2.tryLock());
latch.countDown();
@@ -193,7 +192,7 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
return e;
}
finally {
registry2.destroy();
dynamoDbLockRepository.close();
}
return null;
});
@@ -202,7 +201,7 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
lock1.unlock();
Object ise = result.get(10, TimeUnit.SECONDS);
assertThat(ise).isInstanceOf(IllegalMonitorStateException.class);
assertThat(((Exception) ise).getMessage()).contains("You do not own");
assertThat(((Exception) ise).getMessage()).contains("The current thread doesn't own mutex at 'foo'");
}
@Test
@@ -214,11 +213,11 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
final CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
this.taskExecutor.submit(() -> {
DynamoDbLockRegistry registry2 = new DynamoDbLockRegistry(DYNAMO_DB);
registry2.setHeartbeatPeriod(1);
registry2.setRefreshPeriod(10);
registry2.setLeaseDuration(2);
registry2.afterPropertiesSet();
DynamoDbLockRepository dynamoDbLockRepository = new DynamoDbLockRepository(DYNAMO_DB);
dynamoDbLockRepository.setLeaseDuration(Duration.ofSeconds(10));
dynamoDbLockRepository.afterPropertiesSet();
DynamoDbLockRegistry registry2 = new DynamoDbLockRegistry(dynamoDbLockRepository);
registry2.setIdleBetweenTries(Duration.ofMillis(10));
Lock lock2 = registry2.obtain("foo");
try {
latch1.countDown();
@@ -232,7 +231,7 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
finally {
lock2.unlock();
latch3.countDown();
registry2.destroy();
dynamoDbLockRepository.close();
}
return null;
});
@@ -249,17 +248,14 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
@Test
void testTwoThreadsDifferentRegistries() throws Exception {
final DynamoDbLockRegistry registry1 = new DynamoDbLockRegistry(DYNAMO_DB);
registry1.setHeartbeatPeriod(1);
registry1.setRefreshPeriod(10);
registry1.setLeaseDuration(2);
registry1.afterPropertiesSet();
DynamoDbLockRepository dynamoDbLockRepository = new DynamoDbLockRepository(DYNAMO_DB);
dynamoDbLockRepository.setLeaseDuration(Duration.ofSeconds(10));
dynamoDbLockRepository.afterPropertiesSet();
final DynamoDbLockRegistry registry1 = new DynamoDbLockRegistry(dynamoDbLockRepository);
registry1.setIdleBetweenTries(Duration.ofMillis(10));
final DynamoDbLockRegistry registry2 = new DynamoDbLockRegistry(DYNAMO_DB);
registry2.setHeartbeatPeriod(1);
registry2.setRefreshPeriod(10);
registry2.setLeaseDuration(2);
registry2.afterPropertiesSet();
final DynamoDbLockRegistry registry2 = new DynamoDbLockRegistry(dynamoDbLockRepository);
registry2.setIdleBetweenTries(Duration.ofMillis(10));
final Lock lock1 = registry1.obtain("foo");
final AtomicBoolean locked = new AtomicBoolean();
@@ -292,8 +288,7 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
assertThat(latch3.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(locked.get()).isTrue();
registry1.destroy();
registry2.destroy();
dynamoDbLockRepository.close();
}
@Test
@@ -319,34 +314,15 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
lock.unlock();
Object imse = result.get(10, TimeUnit.SECONDS);
assertThat(imse).isInstanceOf(IllegalMonitorStateException.class);
assertThat(((Exception) imse).getMessage()).contains("You do not own");
assertThat(((Exception) imse).getMessage()).contains("The current thread doesn't own mutex at 'foo'");
}
@Test
void abandonedLock() throws Exception {
Method awaitForActiveMethod = ReflectionUtils.findMethod(DynamoDbLockRegistry.class, "awaitForActive");
ReflectionUtils.makeAccessible(awaitForActiveMethod);
ReflectionUtils.invokeMethod(awaitForActiveMethod, this.dynamoDbLockRegistry);
AmazonDynamoDBLockClientOptions lockClientOptions =
AmazonDynamoDBLockClientOptions.builder(DYNAMO_DB, DynamoDbLockRegistry.DEFAULT_TABLE_NAME)
.withPartitionKeyName(DynamoDbLockRegistry.DEFAULT_PARTITION_KEY_NAME)
.withSortKeyName(DynamoDbLockRegistry.DEFAULT_SORT_KEY_NAME)
.withCreateHeartbeatBackgroundThread(false)
.withLeaseDuration(2L)
.build();
AmazonDynamoDBLockClient lockClient = new AmazonDynamoDBLockClient(lockClientOptions);
AcquireLockOptions lockOptions =
AcquireLockOptions.builder("foo")
.withReplaceData(false)
.withSortKey(DynamoDbLockRegistry.DEFAULT_SORT_KEY)
.build();
LockItem lockItem = lockClient.acquireLock(lockOptions);
assertThat(lockItem).isNotNull();
lockClient.close();
DynamoDbLockRepository dynamoDbLockRepository = new DynamoDbLockRepository(DYNAMO_DB);
dynamoDbLockRepository.setLeaseDuration(Duration.ofSeconds(10));
dynamoDbLockRepository.afterPropertiesSet();
this.dynamoDbLockRepository.acquire("foo");
Lock lock = this.dynamoDbLockRegistry.obtain("foo");
int n = 0;
@@ -356,17 +332,36 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
assertThat(n).isLessThan(100);
lock.unlock();
dynamoDbLockRepository.close();
}
@Test
public void testLockRenew() {
final Lock lock = this.dynamoDbLockRegistry.obtain("foo");
assertThat(lock.tryLock()).isTrue();
try {
assertThatNoException().isThrownBy(() ->dynamoDbLockRegistry.renewLock("foo"));
}
finally {
lock.unlock();
}
}
@Configuration
public static class ContextConfiguration {
@Bean
public DynamoDbLockRepository dynamoDbLockRepository() {
DynamoDbLockRepository dynamoDbLockRepository = new DynamoDbLockRepository(DYNAMO_DB);
dynamoDbLockRepository.setLeaseDuration(Duration.ofSeconds(2));
return dynamoDbLockRepository;
}
@Bean
public DynamoDbLockRegistry dynamoDbLockRegistry() {
DynamoDbLockRegistry dynamoDbLockRegistry = new DynamoDbLockRegistry(DYNAMO_DB);
dynamoDbLockRegistry.setHeartbeatPeriod(1);
dynamoDbLockRegistry.setRefreshPeriod(10);
dynamoDbLockRegistry.setLeaseDuration(2);
DynamoDbLockRegistry dynamoDbLockRegistry = new DynamoDbLockRegistry(dynamoDbLockRepository());
dynamoDbLockRegistry.setIdleBetweenTries(Duration.ofMillis(10));
return dynamoDbLockRegistry;
}