From dd52e8f304d20bbdc46d3f9da77e26dd7977f8b7 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 19 Jul 2018 18:21:23 -0400 Subject: [PATCH] Fix DynamoDbLockReg for LockClient requirements https://stackoverflow.com/questions/51428196/spring-aws-kinesis-binder-acquiring-and-releasing-lock-issues-in-dynamo-db-while According the logic in the `AmazonDynamoDBLockClient.acquireLock()` we need to let the loop to iterate at least twice to really get access to the existing lock and check its expiration status * Fix the `DynamoDbLockRegistry.tryLock()` to set a proper `.withRefreshPeriod(0L)` and `.withRefreshPeriod(0L)` to let it pull the DB until we reach a `leaseDuration` limit on the lock or its expiration status --- .../aws/lock/DynamoDbLockRegistry.java | 24 +++++++++---------- 1 file changed, 11 insertions(+), 13 deletions(-) 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 5fd6db3..1471704 100644 --- a/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java +++ b/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java @@ -366,8 +366,7 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing AcquireLockOptions.builder(this.key) .withReplaceData(false) .withSortKey(DynamoDbLockRegistry.this.sortKey) - .withTimeUnit(TimeUnit.MILLISECONDS) - .withRefreshPeriod(DynamoDbLockRegistry.this.refreshPeriod); + .withTimeUnit(TimeUnit.MILLISECONDS); } private void rethrowAsLockException(Exception e) { @@ -380,8 +379,7 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing this.delegate.lock(); - this.acquireLockOptionsBuilder - .withAdditionalTimeToWaitForLock(Long.MAX_VALUE - DynamoDbLockRegistry.this.leaseDuration); + setupDefaultAcquireLockOptionsBuilder(); boolean wasInterruptedWhileUninterruptible = false; @@ -415,14 +413,19 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing } + 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(); - this.acquireLockOptionsBuilder - .withAdditionalTimeToWaitForLock(Long.MAX_VALUE - DynamoDbLockRegistry.this.leaseDuration); + setupDefaultAcquireLockOptionsBuilder(); try { while (!doLock()) { @@ -460,18 +463,13 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { awaitForActive(); - long start = System.currentTimeMillis(); - if (!this.delegate.tryLock(time, unit)) { return false; } - long refreshPeriod = System.currentTimeMillis() - start + TimeUnit.MILLISECONDS.convert(time, unit); - long timeToWaitForLock = refreshPeriod - DynamoDbLockRegistry.this.leaseDuration; - this.acquireLockOptionsBuilder - .withAdditionalTimeToWaitForLock(timeToWaitForLock) - .withRefreshPeriod(refreshPeriod); + .withAdditionalTimeToWaitForLock(0L) + .withRefreshPeriod(0L); boolean acquired = false; try {