From 594ea58f28daea260d563723bfc70a819db1022a Mon Sep 17 00:00:00 2001 From: abilan Date: Mon, 27 Feb 2023 12:00:46 -0500 Subject: [PATCH] Implement our own DynamoDb lock repository 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 --- README.md | 2 - build.gradle | 6 +- src/checkstyle/checkstyle.xml | 2 +- .../KinesisMessageDrivenChannelAdapter.java | 35 +- .../aws/lock/DynamoDbLockRegistry.java | 609 +++++------------- .../aws/lock/DynamoDbLockRepository.java | 445 +++++++++++++ ...amoDbLockRegistryLeaderInitiatorTests.java | 33 +- .../DynamoDbLockRegistryBuildTableTests.java | 123 ---- .../aws/lock/DynamoDbLockRegistryTests.java | 125 ++-- 9 files changed, 716 insertions(+), 664 deletions(-) create mode 100644 src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRepository.java delete mode 100644 src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryBuildTableTests.java 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: + *

    + *
  • {@link DynamoDbLockRepository#KEY_ATTR} {@link ScalarAttributeType#S} - partition key {@link KeyType#HASH} + *
  • {@link DynamoDbLockRepository#OWNER_ATTR} {@link ScalarAttributeType#S} + *
  • {@link DynamoDbLockRepository#CREATED_ATTR} {@link ScalarAttributeType#N} + *
  • {@link DynamoDbLockRepository#TTL_ATTR} {@link ScalarAttributeType#N} + *
+ * + * @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 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 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); + } + +} diff --git a/src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java b/src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java index 21cdba5..3080df2 100644 --- a/src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java +++ b/src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java @@ -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 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 registries = new ArrayList<>(); + List repositories = new ArrayList<>(); List 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 { diff --git a/src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryBuildTableTests.java b/src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryBuildTableTests.java deleted file mode 100644 index 27917dc..0000000 --- a/src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryBuildTableTests.java +++ /dev/null @@ -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 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; - } - } -} diff --git a/src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryTests.java b/src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryTests.java index 534f062..7635ae4 100644 --- a/src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryTests.java +++ b/src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryTests.java @@ -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 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 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; }