Increased logging verbosity around lease management. Also included ad…

…ditional 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 @@ private void runShardSync() {
                             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 @@ private void runShardSync() {
         }
     }
 
+    /**
+     * 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 @@ private Map<StreamIdentifier, List<Lease>> getStreamToLeasesMap(
         }
     }
 
+
+    /**
+     * 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 @@ ShardSyncResponse checkForShardSync(StreamIdentifier streamIdentifier, List<Leas
         }
     }
 
+    /**
+     * 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 @@ void initialize() {
 
             for (int i = 0; (!isDone) && (i < maxInitializationAttempts); i++) {
                 try {
-                    log.info("Initialization attempt {}", (i + 1));
-                    log.info("Initializing LeaseCoordinator");
+                    log.info("Initializing LeaseCoordinator attempt {}", (i + 1));
                     leaseCoordinator.initialize();
 
                     if (!skipShardSyncAtWorkerInitializationIfLeasesExist || leaseRefresher.isLeaseTableEmpty()) {

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

@@ -148,17 +148,24 @@ public synchronized boolean checkAndCreateLeaseForNewShards(@NonNull final Shard
                 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 @@ static boolean checkIfDescendantAndAddNewLeasesForAncestors(final String shardId
                 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 @@ static boolean checkIfDescendantAndAddNewLeasesForAncestors(final String shardId
                                 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);
                                 }
                             }
                         }
@@ -512,7 +523,7 @@ static boolean checkIfDescendantAndAddNewLeasesForAncestors(final String shardId
      * Helper method to get parent shardIds of the current shard - includes the parent shardIds if:
      * a/ they are not null
      * b/ if they exist in the current shard map (i.e. haven't expired)
-     * 
+     *
      * @param shard Will return parents of this shard
      * @param shardIdToShardMapOfAllKinesisShards ShardId->Shard map containing all shards obtained via DescribeStream.
      * @return Set of parentShardIds
@@ -538,6 +549,12 @@ public synchronized Lease createLeaseForChildShard(final ChildShard childShard,
                                                    : 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());
@@ -571,7 +588,7 @@ private static Lease newKCLMultiStreamLeaseForChildShard(final ChildShard childS
     /**
      * Helper method to create a new Lease POJO for a shard.
      * Note: Package level access only for testing purposes
-     * 
+     *
      * @param shard
      * @return
      */
@@ -611,7 +628,7 @@ private static Lease newKCLMultiStreamLease(final Shard shard, final StreamIdent
 
     /**
      * Helper method to construct a shardId->Shard map for the specified list of shards.
-     * 
+     *
      * @param shards List of shards
      * @return ShardId->Shard map
      */
@@ -622,7 +639,7 @@ static Map<String, Shard> constructShardIdToShardMap(final List<Shard> shards) {
     /**
      * Helper method to return all the open shards for a stream.
      * Note: Package level access only for testing purposes.
-     * 
+     *
      * @param allShards All shards returved via DescribeStream. We assume this to represent a consistent shard list.
      * @return List of open shards (shards at the tip of the stream) - may include shards that are not yet active.
      */
@@ -633,15 +650,15 @@ static List<Shard> getOpenShards(final List<Shard> allShards, final String strea
 
     private static ExtendedSequenceNumber convertToCheckpoint(final InitialPositionInStreamExtended position) {
         ExtendedSequenceNumber checkpoint = null;
-        
+
         if (position.getInitialPositionInStream().equals(InitialPositionInStream.TRIM_HORIZON)) {
             checkpoint = ExtendedSequenceNumber.TRIM_HORIZON;
         } else if (position.getInitialPositionInStream().equals(InitialPositionInStream.LATEST)) {
             checkpoint = ExtendedSequenceNumber.LATEST;
         } else if (position.getInitialPositionInStream().equals(InitialPositionInStream.AT_TIMESTAMP)) {
             checkpoint = ExtendedSequenceNumber.AT_TIMESTAMP;
         }
-        
+
         return checkpoint;
     }
 
@@ -688,7 +705,7 @@ private static class StartingSequenceNumberAndShardIdBasedComparator implements
          * We assume that lease1 and lease2 are:
          *     a/ not null,
          *     b/ shards (if found) have non-null starting sequence numbers
-         * 
+         *
          * {@inheritDoc}
          */
         @Override
@@ -698,18 +715,18 @@ public int compare(final Lease lease1, final Lease lease2) {
             final String shardId2 = shardIdFromLeaseDeducer.apply(lease2, multiStreamArgs);
             final Shard shard1 = shardIdToShardMap.get(shardId1);
             final Shard shard2 = shardIdToShardMap.get(shardId2);
-            
+
             // If we found shards for the two leases, use comparison of the starting sequence numbers
             if (shard1 != null && shard2 != null) {
                 BigInteger sequenceNumber1 = new BigInteger(shard1.sequenceNumberRange().startingSequenceNumber());
                 BigInteger sequenceNumber2 = new BigInteger(shard2.sequenceNumberRange().startingSequenceNumber());
-                result = sequenceNumber1.compareTo(sequenceNumber2);                
+                result = sequenceNumber1.compareTo(sequenceNumber2);
             }
-            
+
             if (result == 0) {
                 result = shardId1.compareTo(shardId2);
             }
-            
+
             return result;
         }
     }

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

@@ -145,10 +145,18 @@ private int leasesPendingDeletion() {
         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 @@ private boolean cleanupLeaseForGarbageShard(Lease lease, Throwable e) throws Dep
         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 boolean createLeaseTableIfNotExists(@NonNull final Long readCapacity, @No
             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 @@ private List<Lease> list(Integer limit, Integer maxPages, StreamIdentifier strea
     @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 boolean createLeaseIfNotExists(@NonNull final Lease lease)
         } catch (DynamoDbException | TimeoutException e) {
             throw convertAndRethrowExceptions("create", lease.leaseKey(), e);
         }
+        log.info("Created lease: {}",lease);
         return true;
     }
 
@@ -476,7 +477,7 @@ public Lease getLease(@NonNull final String leaseKey)
                     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 boolean renewLease(@NonNull final Lease lease)
         }
 
         lease.leaseCounter(lease.leaseCounter() + 1);
+        log.debug("Renewed lease with key {}", lease.leaseKey());
         return true;
     }
 
@@ -582,6 +584,8 @@ public boolean takeLease(@NonNull final Lease lease, @NonNull final String owner
             lease.ownerSwitchesSinceCheckpoint(lease.ownerSwitchesSinceCheckpoint() + 1);
         }
 
+        log.info("Transferred lease {} ownership from {} to {}", lease.leaseKey(), oldOwner, owner);
+
         return true;
     }
 
@@ -620,6 +624,8 @@ public boolean evictLease(@NonNull final Lease lease)
 
         lease.leaseOwner(null);
         lease.leaseCounter(lease.leaseCounter() + 1);
+
+        log.info("Evicted lease with leaseKey {}", lease.leaseKey());
         return true;
     }
 
@@ -648,6 +654,7 @@ public void deleteAll() throws DependencyException, InvalidStateException, Provi
             } catch (DynamoDbException | TimeoutException e) {
                 throw convertAndRethrowExceptions("deleteAll", lease.leaseKey(), e);
             }
+            log.debug("Deleted lease {} from table {}", lease.leaseKey(), table);
         }
     }
 
@@ -675,6 +682,8 @@ public void deleteLease(@NonNull final Lease lease)
         } catch (DynamoDbException | TimeoutException e) {
             throw convertAndRethrowExceptions("delete", lease.leaseKey(), e);
         }
+
+        log.info("Deleted lease with leaseKey {}", lease.leaseKey());
     }
 
     /**
@@ -683,7 +692,7 @@ public void deleteLease(@NonNull final Lease lease)
     @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 boolean updateLease(@NonNull final Lease lease)
         }
 
         lease.leaseCounter(lease.leaseCounter() + 1);
+        log.info("Updated lease {}.", lease.leaseKey());
         return true;
     }
 
@@ -738,6 +748,8 @@ public void updateLeaseWithMetaInfo(Lease lease, UpdateField updateField)
         } 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

@@ -242,7 +242,7 @@ public Lease getCurrentlyHeldLease(String leaseKey) {
 
     /**
      * Internal method to return a lease with a specific lease key only if we currently hold it.
-     * 
+     *
      * @param leaseKey key of lease to return
      * @param now current timestamp for old-ness checking
      * @return non-authoritative copy of the held lease, or null if we don't currently hold it
@@ -309,6 +309,7 @@ public boolean updateLease(Lease lease, UUID concurrencyToken, @NonNull String o
         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);
@@ -325,15 +326,15 @@ public boolean updateLease(Lease lease, UUID concurrencyToken, @NonNull String o
                     /*
                      * Remove only if the value currently in the map is the same as the authoritative lease. We're
                      * guarding against a pause after the concurrency token check above. It plays out like so:
-                     * 
+                     *
                      * 1) Concurrency token check passes
                      * 2) Pause. Lose lease, re-acquire lease. This requires at least one lease counter update.
                      * 3) Unpause. leaseRefresher.updateLease fails conditional write due to counter updates, returns
                      * false.
                      * 4) ownedLeases.remove(key, value) doesn't do anything because authoritativeLease does not
                      * .equals() the re-acquired version in the map on the basis of lease counter. This is what we want.
                      * If we just used ownedLease.remove(key), we would have pro-actively removed a lease incorrectly.
-                     * 
+                     *
                      * Note that there is a subtlety here - Lease.equals() deliberately does not check the concurrency
                      * token, but it does check the lease counter, so this scheme works.
                      */