Increased logging verbosity around lease management. Also included additional javadocs for methods (#1040)

Co-authored-by: Ryan Pelaez <rmpelaez@amazon.com>

amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/PeriodicShardSyncManager.java

@@ -202,6 +202,10 @@ class PeriodicShardSyncManager {
                             log.warn(
                                     "Failed to submit shard sync task for stream {}. This could be due to the previous pending shard sync task.",
                                     shardSyncTaskManager.shardDetector().streamIdentifier().streamName());
+                        } else {
+                            log.info("Submitted shard sync task for stream {} because of reason {}",
+                                    shardSyncTaskManager.shardDetector().streamIdentifier().streamName(),
+                                    shardSyncResponse.reasonForDecision());
                         }
                     } else {
                         log.info("Skipping shard sync for {} due to the reason - {}", streamConfigEntry.getKey(),
@@ -222,6 +226,14 @@ class PeriodicShardSyncManager {
         }
     }
 
+    /**
+     * Retrieve all the streams, along with their associated leases
+     * @param streamIdentifiersToFilter
+     * @return
+     * @throws DependencyException
+     * @throws ProvisionedThroughputException
+     * @throws InvalidStateException
+     */
     private Map<StreamIdentifier, List<Lease>> getStreamToLeasesMap(
             final Set<StreamIdentifier> streamIdentifiersToFilter)
             throws DependencyException, ProvisionedThroughputException, InvalidStateException {
@@ -242,6 +254,13 @@ class PeriodicShardSyncManager {
         }
     }
 
+
+    /**
+     * Given a list of leases for a stream, determine if a shard sync is necessary.
+     * @param streamIdentifier
+     * @param leases
+     * @return
+     */
     @VisibleForTesting
     ShardSyncResponse checkForShardSync(StreamIdentifier streamIdentifier, List<Lease> leases) {
         if (CollectionUtils.isNullOrEmpty(leases)) {
@@ -272,12 +291,24 @@ class PeriodicShardSyncManager {
         }
     }
 
+    /**
+     * Object containing metadata about the state of a shard sync
+     */
     @Value
     @Accessors(fluent = true)
     @VisibleForTesting
     static class ShardSyncResponse {
+
+        /**
+         * Flag to determine if a shard sync is necessary or not
+         */
         private final boolean shouldDoShardSync;
+
         private final boolean isHoleDetected;
+
+        /**
+         * Reason behind the state of 'shouldDoShardSync' flag
+         */
         private final String reasonForDecision;
     }
 

amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/Scheduler.java

@@ -331,8 +331,7 @@ public class Scheduler implements Runnable {
 
             for (int i = 0; (!isDone) && (i < maxInitializationAttempts); i++) {
                 try {
-                    log.info("Initialization attempt {}", (i + 1));
+                    log.info("Initializing LeaseCoordinator attempt {}", (i + 1));
-                    log.info("Initializing LeaseCoordinator");
                     leaseCoordinator.initialize();
 
                     if (!skipShardSyncAtWorkerInitializationIfLeasesExist || leaseRefresher.isLeaseTableEmpty()) {

amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/HierarchicalShardSyncer.java

@@ -148,17 +148,24 @@ public class HierarchicalShardSyncer {
                 new NonEmptyLeaseTableSynchronizer(shardDetector, shardIdToShardMap, shardIdToChildShardIdsMap);
         final List<Lease> newLeasesToCreate = determineNewLeasesToCreate(leaseSynchronizer, latestShards, currentLeases,
                 initialPosition, inconsistentShardIds, multiStreamArgs);
-        log.debug("{} - Num new leases to create: {}", streamIdentifier, newLeasesToCreate.size());
+        log.info("{} - Number of new leases to create: {}", streamIdentifier, newLeasesToCreate.size());
+
+        final Set<Lease> createdLeases = new HashSet<>();
+
         for (Lease lease : newLeasesToCreate) {
             long startTime = System.currentTimeMillis();
             boolean success = false;
             try {
-                leaseRefresher.createLeaseIfNotExists(lease);
+                if(leaseRefresher.createLeaseIfNotExists(lease)) {
+                    createdLeases.add(lease);
+                }
                 success = true;
-            } finally {
+            }
+            finally {
                 MetricsUtil.addSuccessAndLatency(scope, "CreateLease", success, startTime, MetricsLevel.DETAILED);
             }
         }
+        log.info("{} - Newly created leases {}: {}", streamIdentifier, createdLeases.size(), createdLeases);
         final List<Lease> trackedLeases = new ArrayList<>(currentLeases);
         trackedLeases.addAll(newLeasesToCreate);
         return true;
@@ -398,6 +405,7 @@ public class HierarchicalShardSyncer {
                 isDescendant = true;
                 // We don't need to add leases of its ancestors,
                 // because we'd have done it when creating a lease for this shard.
+                log.debug("{} - Shard {} is a descendant shard of an existing shard. Skipping lease creation", streamIdentifier, shardId);
             } else {
 
                 final Shard shard = shardIdToShardMapOfAllKinesisShards.get(shardId);
@@ -474,9 +482,12 @@ public class HierarchicalShardSyncer {
                                 if (descendantParentShardIds.contains(parentShardId)
                                         && !initialPosition.getInitialPositionInStream()
                                         .equals(InitialPositionInStream.AT_TIMESTAMP)) {
+                                    log.info("Setting Lease '{}' checkpoint to 'TRIM_HORIZON'. Checkpoint was previously set to {}", lease.leaseKey(), lease.checkpoint());
                                     lease.checkpoint(ExtendedSequenceNumber.TRIM_HORIZON);
                                 } else {
-                                    lease.checkpoint(convertToCheckpoint(initialPosition));
+                                    final ExtendedSequenceNumber newCheckpoint = convertToCheckpoint(initialPosition);
+                                    log.info("Setting Lease '{}' checkpoint to '{}'. Checkpoint was previously set to {}", lease.leaseKey(), newCheckpoint, lease.checkpoint());
+                                    lease.checkpoint(newCheckpoint);
                                 }
                             }
                         }
@@ -538,6 +549,12 @@ public class HierarchicalShardSyncer {
                                                    : newKCLLeaseForChildShard(childShard);
     }
 
+    /**
+     * Generate a lease object for the given Child Shard. Checkpoint is set to TRIM_HORIZON
+     * @param childShard Shard for which a lease should be created
+     * @return Lease for the shard
+     * @throws InvalidStateException If the child shard has no parent shards
+     */
     private static Lease newKCLLeaseForChildShard(final ChildShard childShard) throws InvalidStateException {
         Lease newLease = new Lease();
         newLease.leaseKey(childShard.shardId());

amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseCleanupManager.java

@@ -145,10 +145,18 @@ public class LeaseCleanupManager {
         return deletionQueue.size();
     }
 
+    /**
+     *
+     * @return true if the 'Completed Lease Stopwatch' has elapsed more time than the 'Completed Lease Cleanup Interval'
+     */
     private boolean timeToCheckForCompletedShard() {
         return completedLeaseStopwatch.elapsed(TimeUnit.MILLISECONDS) >= completedLeaseCleanupIntervalMillis;
     }
 
+    /**
+     *
+     * @return true if the 'Garbage Lease Stopwatch' has elapsed more time than the 'Garbage Lease Cleanup Interval'
+     */
     private boolean timeToCheckForGarbageShard() {
         return garbageLeaseStopwatch.elapsed(TimeUnit.MILLISECONDS) >= garbageLeaseCleanupIntervalMillis;
     }
@@ -230,6 +238,15 @@ public class LeaseCleanupManager {
         return true;
     }
 
+    /**
+     * Check if the all of the parent shards for a given lease have an ongoing lease. If any one parent still has a lease, return false. Otherwise return true
+     * @param lease
+     * @param shardInfo
+     * @return
+     * @throws DependencyException
+     * @throws ProvisionedThroughputException
+     * @throws InvalidStateException
+     */
     private boolean allParentShardLeasesDeleted(Lease lease, ShardInfo shardInfo) throws DependencyException, ProvisionedThroughputException, InvalidStateException {
         for (String parentShard : lease.parentShardIds()) {
             final Lease parentLease = leaseCoordinator.leaseRefresher().getLease(ShardInfo.getLeaseKey(shardInfo, parentShard));

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

@@ -169,7 +169,7 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
             request = CreateTableRequest.builder().tableName(table).keySchema(serializer.getKeySchema())
                     .attributeDefinitions(serializer.getAttributeDefinitions())
                     .billingMode(billingMode).build();
-        }else{
+        } else {
             request = CreateTableRequest.builder().tableName(table).keySchema(serializer.getKeySchema())
                     .attributeDefinitions(serializer.getAttributeDefinitions()).provisionedThroughput(throughput)
                     .build();
@@ -429,7 +429,7 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
     @Override
     public boolean createLeaseIfNotExists(@NonNull final Lease lease)
             throws DependencyException, InvalidStateException, ProvisionedThroughputException {
-        log.debug("Creating lease {}", lease);
+        log.debug("Creating lease: {}", lease);
 
         PutItemRequest request = PutItemRequest.builder().tableName(table).item(serializer.toDynamoRecord(lease))
                 .expected(serializer.getDynamoNonexistantExpectation()).build();
@@ -452,6 +452,7 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
         } catch (DynamoDbException | TimeoutException e) {
             throw convertAndRethrowExceptions("create", lease.leaseKey(), e);
         }
+        log.info("Created lease: {}",lease);
         return true;
     }
 
@@ -476,7 +477,7 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
                     return null;
                 } else {
                     final Lease lease = serializer.fromDynamoRecord(dynamoRecord);
-                    log.debug("Got lease {}", lease);
+                    log.debug("Retrieved lease: {}", lease);
                     return lease;
                 }
             } catch (ExecutionException e) {
@@ -535,6 +536,7 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
         }
 
         lease.leaseCounter(lease.leaseCounter() + 1);
+        log.debug("Renewed lease with key {}", lease.leaseKey());
         return true;
     }
 
@@ -582,6 +584,8 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
             lease.ownerSwitchesSinceCheckpoint(lease.ownerSwitchesSinceCheckpoint() + 1);
         }
 
+        log.info("Transferred lease {} ownership from {} to {}", lease.leaseKey(), oldOwner, owner);
+
         return true;
     }
 
@@ -620,6 +624,8 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
 
         lease.leaseOwner(null);
         lease.leaseCounter(lease.leaseCounter() + 1);
+
+        log.info("Evicted lease with leaseKey {}", lease.leaseKey());
         return true;
     }
 
@@ -648,6 +654,7 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
             } catch (DynamoDbException | TimeoutException e) {
                 throw convertAndRethrowExceptions("deleteAll", lease.leaseKey(), e);
             }
+            log.debug("Deleted lease {} from table {}", lease.leaseKey(), table);
         }
     }
 
@@ -675,6 +682,8 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
         } catch (DynamoDbException | TimeoutException e) {
             throw convertAndRethrowExceptions("delete", lease.leaseKey(), e);
         }
+
+        log.info("Deleted lease with leaseKey {}", lease.leaseKey());
     }
 
     /**
@@ -683,7 +692,7 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
     @Override
     public boolean updateLease(@NonNull final Lease lease)
             throws DependencyException, InvalidStateException, ProvisionedThroughputException {
-        log.debug("Updating lease {}", lease);
+        log.debug("Updating lease: {}", lease);
 
         final AWSExceptionManager exceptionManager = createExceptionManager();
         exceptionManager.add(ConditionalCheckFailedException.class, t -> t);
@@ -711,6 +720,7 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
         }
 
         lease.leaseCounter(lease.leaseCounter() + 1);
+        log.info("Updated lease {}.", lease.leaseKey());
         return true;
     }
 
@@ -738,6 +748,8 @@ public class DynamoDBLeaseRefresher implements LeaseRefresher {
         } catch (DynamoDbException | TimeoutException e) {
             throw convertAndRethrowExceptions("update", lease.leaseKey(), e);
         }
+
+        log.info("Updated lease without expectation {}.", lease);
     }
 
     /**

amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/dynamodb/DynamoDBLeaseRenewer.java

@@ -309,6 +309,7 @@ public class DynamoDBLeaseRenewer implements LeaseRenewer {
         long startTime = System.currentTimeMillis();
         boolean success = false;
         try {
+            log.info("Updating lease from {} to {}", authoritativeLease, lease);
             synchronized (authoritativeLease) {
                 authoritativeLease.update(lease);
                 boolean updatedLease = leaseRefresher.updateLease(authoritativeLease);

amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/dynamodb/DynamoDBLeaseTaker.java

@@ -259,6 +259,7 @@ public class DynamoDBLeaseTaker implements LeaseTaker {
             leasesToTake = leasesToTake.stream().map(lease -> {
                 if (lease.isMarkedForLeaseSteal()) {
                     try {
+                        log.debug("Updating stale lease {}.", lease.leaseKey());
                         return leaseRefresher.getLease(lease.leaseKey());
                     } catch (DependencyException | InvalidStateException | ProvisionedThroughputException e) {
                         log.warn("Failed to fetch latest state of the lease {} that needs to be stolen, "
@@ -408,7 +409,7 @@ public class DynamoDBLeaseTaker implements LeaseTaker {
                 target = 1;
             } else {
                 /*
-                 * numWorkers must be < numLeases.
+                 * if we have made it here, it means there are more leases than workers
                  *
                  * Our target for each worker is numLeases / numWorkers (+1 if numWorkers doesn't evenly divide numLeases)
                  */

amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShutdownTask.java

@@ -267,13 +267,14 @@ public class ShutdownTask implements ConsumerTask {
                 }
             }
         }
-        // Attempt create leases for child shards.
         for(ChildShard childShard : childShards) {
             final String leaseKey = ShardInfo.getLeaseKey(shardInfo, childShard.shardId());
             if(leaseCoordinator.leaseRefresher().getLease(leaseKey) == null) {
+                log.debug("{} - Shard {} - Attempting to create lease for child shard {}", shardDetector.streamIdentifier(), shardInfo.shardId(), leaseKey);
                 final Lease leaseToCreate = hierarchicalShardSyncer.createLeaseForChildShard(childShard, shardDetector.streamIdentifier());
                 leaseCoordinator.leaseRefresher().createLeaseIfNotExists(leaseToCreate);
-                log.info("Shard {}: Created child shard lease: {}", shardInfo.shardId(), leaseToCreate.leaseKey());
+
+                log.info("{} - Shard {}: Created child shard lease: {}", shardDetector.streamIdentifier(), shardInfo.shardId(), leaseToCreate);
             }
         }
     }