luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add conditional check for spurious failures from DDB on lease put

Browse source
This commit is contained in:
Eric Zhu 2021-01-25 17:11:55 -08:00
parent e03285508e
commit 81939b599c
1 changed files with 52 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/dynamodb/DynamoDBLeaseRefresher.java
View file

@ -430,7 +430,19 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
}
} catch (ConditionalCheckFailedException e) {
log.debug("Did not create lease {} because it already existed", lease);
return false;
// If we had a spurious retry during the Dynamo put, then this conditional PUT failure
// might be incorrect. So, we get the item straight away and check if the lease owner + lease
// counter are what we expected.
String expectedOwner = lease.leaseOwner();
Long expectedCounter = lease.leaseCounter();
final Lease updatedLease = getLease(lease.leaseKey());
if (updatedLease == null || !expectedOwner.equals(updatedLease.leaseOwner())
|| !expectedCounter.equals(updatedLease.leaseCounter())) {
return false;
}
log.info("Detected spurious renewal failure for lease with key {}, but recovered", lease.leaseKey());
} catch (DynamoDbException | TimeoutException e) {
throw convertAndRethrowExceptions("create", lease.leaseKey(), e);
}
@ -552,7 +564,19 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
} catch (ConditionalCheckFailedException e) {
log.debug("Lease renewal failed for lease with key {} because the lease counter was not {}",
lease.leaseKey(), lease.leaseCounter());
return false;
// If we had a spurious retry during the Dynamo update, then this conditional PUT failure
// might be incorrect. So, we get the item straight away and check if the lease owner + lease
// counter are what we expected.
String expectedOwner = lease.leaseOwner();
Long expectedCounter = lease.leaseCounter() + 1;
final Lease updatedLease = getLease(lease.leaseKey());
if (updatedLease == null || !expectedOwner.equals(updatedLease.leaseOwner())
|| !expectedCounter.equals(updatedLease.leaseCounter())) {
return false;
}
log.info("Detected spurious renewal failure for lease with key {}, but recovered", lease.leaseKey());
} catch (DynamoDbException | TimeoutException e) {
throw convertAndRethrowExceptions("take", lease.leaseKey(), e);
}
@ -595,7 +619,19 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
} catch (ConditionalCheckFailedException e) {
log.debug("Lease eviction failed for lease with key {} because the lease owner was not {}",
lease.leaseKey(), lease.leaseOwner());
return false;
// If we had a spurious retry during the Dynamo update, then this conditional PUT failure
// might be incorrect. So, we get the item straight away and check if the lease owner + lease
// counter are what we expected.
String expectedOwner = lease.leaseOwner();
Long expectedCounter = lease.leaseCounter() + 1;
final Lease updatedLease = getLease(lease.leaseKey());
if (updatedLease == null || !expectedOwner.equals(updatedLease.leaseOwner())
|| !expectedCounter.equals(updatedLease.leaseCounter())) {
return false;
}
log.info("Detected spurious renewal failure for lease with key {}, but recovered", lease.leaseKey());
} catch (DynamoDbException | TimeoutException e) {
throw convertAndRethrowExceptions("evict", lease.leaseKey(), e);
}
@ -687,7 +723,19 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
} catch (ConditionalCheckFailedException e) {
log.debug("Lease update failed for lease with key {} because the lease counter was not {}",
lease.leaseKey(), lease.leaseCounter());
return false;
// If we had a spurious retry during the Dynamo update, then this conditional PUT failure
// might be incorrect. So, we get the item straight away and check if the lease owner + lease
// counter are what we expected.
String expectedOwner = lease.leaseOwner();
Long expectedCounter = lease.leaseCounter() + 1;
final Lease updatedLease = getLease(lease.leaseKey());
if (updatedLease == null || !expectedOwner.equals(updatedLease.leaseOwner())
|| !expectedCounter.equals(updatedLease.leaseCounter())) {
return false;
}
log.info("Detected spurious renewal failure for lease with key {}, but recovered", lease.leaseKey());
} catch (DynamoDbException | TimeoutException e) {
throw convertAndRethrowExceptions("update", lease.leaseKey(), e);
}