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

Fix DynamoDbLockReg for LockClient requirements

Browse Source 
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
This commit is contained in:
Artem Bilan
2018-07-19 18:21:23 -04:00
parent 0e02b0d470
commit dd52e8f304
1 changed files with 11 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 {