Handle spurious lease renewal failures gracefully.

If the request to conditionally update a lease counter in DynamoDB fails, it's
considered a failure to renew the lease. This is a good thing, except if the
request failure was just because of connectivity problems. In this case the
counter *did* update in DynamoDB, but the Dynamo client retries the request
which then fails the update condition (since the lease counter no longer
matches expected value).

To handle this gracefully we opt to get the lease record from Dynamo and
examine the lease owner and counter. If it matches what we were expecting,
then we consider renewal a success.

src/main/java/com/amazonaws/services/kinesis/leases/impl/LeaseManager.java

@@ -19,6 +19,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
+import com.amazonaws.services.kinesis.leases.util.DynamoUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -367,7 +368,19 @@ public class LeaseManager<T extends Lease> implements ILeaseManager<T> {
                         + " because the lease counter was not " + lease.getLeaseCounter());
             }
 
+            // 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.getLeaseOwner();
+            Long expectedCounter = lease.getLeaseCounter() + 1;
+            T updatedLease = getLease(lease.getLeaseKey());
+            if (updatedLease == null || !expectedOwner.equals(updatedLease.getLeaseOwner()) ||
+                    !expectedCounter.equals(updatedLease.getLeaseCounter())) {
                 return false;
+            }
+
+            LOG.info("Detected spurious renewal failure for lease with key " + lease.getLeaseKey()
+                    + ", but recovered");
         } catch (AmazonClientException e) {
             throw convertAndRethrowExceptions("renew", lease.getLeaseKey(), e);
         }