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

Merge 81939b599c into b9ec494b02

Browse source
This commit is contained in:
zhuer 2024-03-04 15:13:30 -08:00 committed by GitHub
commit 5f36cd1591
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
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

@ -469,7 +469,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);
}
@ -593,7 +605,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);
}
@ -638,7 +662,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);
}
@ -735,7 +771,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);
}