diff --git a/README.md b/README.md
index 0fa318e..ddf0e4c 100644
--- a/README.md
+++ b/README.md
@@ -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/
diff --git a/build.gradle b/build.gradle
index 1c7743b..a307395 100644
--- a/build.gradle
+++ b/build.gradle
@@ -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"
diff --git a/src/checkstyle/checkstyle.xml b/src/checkstyle/checkstyle.xml
index 3f79348..4c4e840 100644
--- a/src/checkstyle/checkstyle.xml
+++ b/src/checkstyle/checkstyle.xml
@@ -52,7 +52,7 @@
-
+
diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java
index 993045f..3062f5b 100644
--- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java
+++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java
@@ -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;
diff --git a/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java b/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java
index fbf8925..35a3e93 100644
--- a/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java
+++ b/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java
@@ -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}.
- *
- * 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 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.
- *
- * This method is a variation of {@link AmazonDynamoDBLockClient#createLockTableInDynamoDB} to support custom
- * {@link BillingMode} for the lock table.
- *
- * If table already exists no exception.
- */
- private void createLockTableInDynamoDB() {
- try {
- KeySchemaElement partitionKeyElement = new KeySchemaElement();
- partitionKeyElement.setAttributeName(this.partitionKey);
- partitionKeyElement.setKeyType(KeyType.HASH);
-
- List keySchema = new ArrayList<>();
- keySchema.add(partitionKeyElement);
-
- Collection 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> iterator = this.locks.entrySet().iterator();
long now = System.currentTimeMillis();
- while (iterator.hasNext()) {
- Map.Entry 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)) + "]";
}
}
diff --git a/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRepository.java b/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRepository.java
new file mode 100644
index 0000000..2bfb363
--- /dev/null
+++ b/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRepository.java
@@ -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.
+ *
+ * The DynamoDb table must have these attributes:
+ *