@Test public void testCheckIfDescendantAndAddNewLeasesForAncestorsForShardWithCurrentLease() { final String shardId = "shardId-current"; final Set<String> shardIdsOfCurrentLeases = new HashSet<>(Collections.singletonList(shardId)); final Map<String, Lease> newLeaseMap = Collections.emptyMap(); final MemoizationContext memoizationContext = new MemoizationContext(); final Map<String, Shard> kinesisShards = new HashMap<>(); kinesisShards.put(shardId, ShardObjectHelper.newShard(shardId, null, null, null)); assertThat( HierarchicalShardSyncer.checkIfDescendantAndAddNewLeasesForAncestors(shardId, INITIAL_POSITION_LATEST, shardIdsOfCurrentLeases, kinesisShards, newLeaseMap, memoizationContext), equalTo(true)); assertThat(newLeaseMap.isEmpty(), equalTo(true)); }
static boolean checkIfDescendantAndAddNewLeasesForAncestors(final String shardId, final InitialPositionInStreamExtended initialPosition, final Set<String> shardIdsOfCurrentLeases, final Map<String, Shard> shardIdToShardMapOfAllKinesisShards, final Map<String, Lease> shardIdToLeaseMapOfNewShards, final MemoizationContext memoizationContext, final MultiStreamArgs multiStreamArgs) { final String streamIdentifier = getStreamIdentifier(multiStreamArgs); final Boolean previousValue = memoizationContext.isDescendant(shardId); if (previousValue != null) { return previousValue; } boolean isDescendant = false; final Set<String> descendantParentShardIds = new HashSet<>(); if (shardId != null && shardIdToShardMapOfAllKinesisShards.containsKey(shardId)) { if (shardIdsOfCurrentLeases.contains(shardId)) { isDescendant = true; } else { final Shard shard = shardIdToShardMapOfAllKinesisShards.get(shardId); final Set<String> parentShardIds = getParentShardIds(shard, shardIdToShardMapOfAllKinesisShards); for (String parentShardId : parentShardIds) { final boolean isParentDescendant = checkIfDescendantAndAddNewLeasesForAncestors(parentShardId, initialPosition, shardIdsOfCurrentLeases, shardIdToShardMapOfAllKinesisShards, shardIdToLeaseMapOfNewShards, memoizationContext, multiStreamArgs); if (isParentDescendant || memoizationContext.shouldCreateLease(parentShardId)) { isDescendant = true; descendantParentShardIds.add(parentShardId); log.debug("{} : Parent shard {} is a descendant.", streamIdentifier, parentShardId); } else { log.debug("{} : Parent shard {} is NOT a descendant.", streamIdentifier, parentShardId); } } if (isDescendant) { for (String parentShardId : parentShardIds) { if (!shardIdsOfCurrentLeases.contains(parentShardId)) { Lease lease = shardIdToLeaseMapOfNewShards.get(parentShardId); if (lease == null) { if (memoizationContext.shouldCreateLease(parentShardId) || !descendantParentShardIds.contains(parentShardId)) { log.debug("{} : Need to create a lease for shardId {}", streamIdentifier, parentShardId); lease = multiStreamArgs.isMultiStreamMode() ? newKCLMultiStreamLease(shardIdToShardMapOfAllKinesisShards.get(parentShardId), multiStreamArgs.streamIdentifier()) : newKCLLease(shardIdToShardMapOfAllKinesisShards.get(parentShardId)); shardIdToLeaseMapOfNewShards.put(parentShardId, lease); } } if (lease != null) { if (descendantParentShardIds.contains(parentShardId) && !initialPosition.getInitialPositionInStream() .equals(InitialPositionInStream.AT_TIMESTAMP)) { lease.checkpoint(ExtendedSequenceNumber.TRIM_HORIZON); } else { lease.checkpoint(convertToCheckpoint(initialPosition)); } } } } } else { if (initialPosition.getInitialPositionInStream().equals(InitialPositionInStream.TRIM_HORIZON) || initialPosition.getInitialPositionInStream() .equals(InitialPositionInStream.AT_TIMESTAMP)) { memoizationContext.setShouldCreateLease(shardId, true); } } } } memoizationContext.setIsDescendant(shardId, isDescendant); return isDescendant; }
HierarchicalShardSyncer { static boolean checkIfDescendantAndAddNewLeasesForAncestors(final String shardId, final InitialPositionInStreamExtended initialPosition, final Set<String> shardIdsOfCurrentLeases, final Map<String, Shard> shardIdToShardMapOfAllKinesisShards, final Map<String, Lease> shardIdToLeaseMapOfNewShards, final MemoizationContext memoizationContext, final MultiStreamArgs multiStreamArgs) { final String streamIdentifier = getStreamIdentifier(multiStreamArgs); final Boolean previousValue = memoizationContext.isDescendant(shardId); if (previousValue != null) { return previousValue; } boolean isDescendant = false; final Set<String> descendantParentShardIds = new HashSet<>(); if (shardId != null && shardIdToShardMapOfAllKinesisShards.containsKey(shardId)) { if (shardIdsOfCurrentLeases.contains(shardId)) { isDescendant = true; } else { final Shard shard = shardIdToShardMapOfAllKinesisShards.get(shardId); final Set<String> parentShardIds = getParentShardIds(shard, shardIdToShardMapOfAllKinesisShards); for (String parentShardId : parentShardIds) { final boolean isParentDescendant = checkIfDescendantAndAddNewLeasesForAncestors(parentShardId, initialPosition, shardIdsOfCurrentLeases, shardIdToShardMapOfAllKinesisShards, shardIdToLeaseMapOfNewShards, memoizationContext, multiStreamArgs); if (isParentDescendant || memoizationContext.shouldCreateLease(parentShardId)) { isDescendant = true; descendantParentShardIds.add(parentShardId); log.debug("{} : Parent shard {} is a descendant.", streamIdentifier, parentShardId); } else { log.debug("{} : Parent shard {} is NOT a descendant.", streamIdentifier, parentShardId); } } if (isDescendant) { for (String parentShardId : parentShardIds) { if (!shardIdsOfCurrentLeases.contains(parentShardId)) { Lease lease = shardIdToLeaseMapOfNewShards.get(parentShardId); if (lease == null) { if (memoizationContext.shouldCreateLease(parentShardId) || !descendantParentShardIds.contains(parentShardId)) { log.debug("{} : Need to create a lease for shardId {}", streamIdentifier, parentShardId); lease = multiStreamArgs.isMultiStreamMode() ? newKCLMultiStreamLease(shardIdToShardMapOfAllKinesisShards.get(parentShardId), multiStreamArgs.streamIdentifier()) : newKCLLease(shardIdToShardMapOfAllKinesisShards.get(parentShardId)); shardIdToLeaseMapOfNewShards.put(parentShardId, lease); } } if (lease != null) { if (descendantParentShardIds.contains(parentShardId) && !initialPosition.getInitialPositionInStream() .equals(InitialPositionInStream.AT_TIMESTAMP)) { lease.checkpoint(ExtendedSequenceNumber.TRIM_HORIZON); } else { lease.checkpoint(convertToCheckpoint(initialPosition)); } } } } } else { if (initialPosition.getInitialPositionInStream().equals(InitialPositionInStream.TRIM_HORIZON) || initialPosition.getInitialPositionInStream() .equals(InitialPositionInStream.AT_TIMESTAMP)) { memoizationContext.setShouldCreateLease(shardId, true); } } } } memoizationContext.setIsDescendant(shardId, isDescendant); return isDescendant; } }
HierarchicalShardSyncer { static boolean checkIfDescendantAndAddNewLeasesForAncestors(final String shardId, final InitialPositionInStreamExtended initialPosition, final Set<String> shardIdsOfCurrentLeases, final Map<String, Shard> shardIdToShardMapOfAllKinesisShards, final Map<String, Lease> shardIdToLeaseMapOfNewShards, final MemoizationContext memoizationContext, final MultiStreamArgs multiStreamArgs) { final String streamIdentifier = getStreamIdentifier(multiStreamArgs); final Boolean previousValue = memoizationContext.isDescendant(shardId); if (previousValue != null) { return previousValue; } boolean isDescendant = false; final Set<String> descendantParentShardIds = new HashSet<>(); if (shardId != null && shardIdToShardMapOfAllKinesisShards.containsKey(shardId)) { if (shardIdsOfCurrentLeases.contains(shardId)) { isDescendant = true; } else { final Shard shard = shardIdToShardMapOfAllKinesisShards.get(shardId); final Set<String> parentShardIds = getParentShardIds(shard, shardIdToShardMapOfAllKinesisShards); for (String parentShardId : parentShardIds) { final boolean isParentDescendant = checkIfDescendantAndAddNewLeasesForAncestors(parentShardId, initialPosition, shardIdsOfCurrentLeases, shardIdToShardMapOfAllKinesisShards, shardIdToLeaseMapOfNewShards, memoizationContext, multiStreamArgs); if (isParentDescendant || memoizationContext.shouldCreateLease(parentShardId)) { isDescendant = true; descendantParentShardIds.add(parentShardId); log.debug("{} : Parent shard {} is a descendant.", streamIdentifier, parentShardId); } else { log.debug("{} : Parent shard {} is NOT a descendant.", streamIdentifier, parentShardId); } } if (isDescendant) { for (String parentShardId : parentShardIds) { if (!shardIdsOfCurrentLeases.contains(parentShardId)) { Lease lease = shardIdToLeaseMapOfNewShards.get(parentShardId); if (lease == null) { if (memoizationContext.shouldCreateLease(parentShardId) || !descendantParentShardIds.contains(parentShardId)) { log.debug("{} : Need to create a lease for shardId {}", streamIdentifier, parentShardId); lease = multiStreamArgs.isMultiStreamMode() ? newKCLMultiStreamLease(shardIdToShardMapOfAllKinesisShards.get(parentShardId), multiStreamArgs.streamIdentifier()) : newKCLLease(shardIdToShardMapOfAllKinesisShards.get(parentShardId)); shardIdToLeaseMapOfNewShards.put(parentShardId, lease); } } if (lease != null) { if (descendantParentShardIds.contains(parentShardId) && !initialPosition.getInitialPositionInStream() .equals(InitialPositionInStream.AT_TIMESTAMP)) { lease.checkpoint(ExtendedSequenceNumber.TRIM_HORIZON); } else { lease.checkpoint(convertToCheckpoint(initialPosition)); } } } } } else { if (initialPosition.getInitialPositionInStream().equals(InitialPositionInStream.TRIM_HORIZON) || initialPosition.getInitialPositionInStream() .equals(InitialPositionInStream.AT_TIMESTAMP)) { memoizationContext.setShouldCreateLease(shardId, true); } } } } memoizationContext.setIsDescendant(shardId, isDescendant); return isDescendant; } HierarchicalShardSyncer();  HierarchicalShardSyncer(final boolean isMultiStreamMode, final String streamIdentifier); }
HierarchicalShardSyncer { static boolean checkIfDescendantAndAddNewLeasesForAncestors(final String shardId, final InitialPositionInStreamExtended initialPosition, final Set<String> shardIdsOfCurrentLeases, final Map<String, Shard> shardIdToShardMapOfAllKinesisShards, final Map<String, Lease> shardIdToLeaseMapOfNewShards, final MemoizationContext memoizationContext, final MultiStreamArgs multiStreamArgs) { final String streamIdentifier = getStreamIdentifier(multiStreamArgs); final Boolean previousValue = memoizationContext.isDescendant(shardId); if (previousValue != null) { return previousValue; } boolean isDescendant = false; final Set<String> descendantParentShardIds = new HashSet<>(); if (shardId != null && shardIdToShardMapOfAllKinesisShards.containsKey(shardId)) { if (shardIdsOfCurrentLeases.contains(shardId)) { isDescendant = true; } else { final Shard shard = shardIdToShardMapOfAllKinesisShards.get(shardId); final Set<String> parentShardIds = getParentShardIds(shard, shardIdToShardMapOfAllKinesisShards); for (String parentShardId : parentShardIds) { final boolean isParentDescendant = checkIfDescendantAndAddNewLeasesForAncestors(parentShardId, initialPosition, shardIdsOfCurrentLeases, shardIdToShardMapOfAllKinesisShards, shardIdToLeaseMapOfNewShards, memoizationContext, multiStreamArgs); if (isParentDescendant || memoizationContext.shouldCreateLease(parentShardId)) { isDescendant = true; descendantParentShardIds.add(parentShardId); log.debug("{} : Parent shard {} is a descendant.", streamIdentifier, parentShardId); } else { log.debug("{} : Parent shard {} is NOT a descendant.", streamIdentifier, parentShardId); } } if (isDescendant) { for (String parentShardId : parentShardIds) { if (!shardIdsOfCurrentLeases.contains(parentShardId)) { Lease lease = shardIdToLeaseMapOfNewShards.get(parentShardId); if (lease == null) { if (memoizationContext.shouldCreateLease(parentShardId) || !descendantParentShardIds.contains(parentShardId)) { log.debug("{} : Need to create a lease for shardId {}", streamIdentifier, parentShardId); lease = multiStreamArgs.isMultiStreamMode() ? newKCLMultiStreamLease(shardIdToShardMapOfAllKinesisShards.get(parentShardId), multiStreamArgs.streamIdentifier()) : newKCLLease(shardIdToShardMapOfAllKinesisShards.get(parentShardId)); shardIdToLeaseMapOfNewShards.put(parentShardId, lease); } } if (lease != null) { if (descendantParentShardIds.contains(parentShardId) && !initialPosition.getInitialPositionInStream() .equals(InitialPositionInStream.AT_TIMESTAMP)) { lease.checkpoint(ExtendedSequenceNumber.TRIM_HORIZON); } else { lease.checkpoint(convertToCheckpoint(initialPosition)); } } } } } else { if (initialPosition.getInitialPositionInStream().equals(InitialPositionInStream.TRIM_HORIZON) || initialPosition.getInitialPositionInStream() .equals(InitialPositionInStream.AT_TIMESTAMP)) { memoizationContext.setShouldCreateLease(shardId, true); } } } } memoizationContext.setIsDescendant(shardId, isDescendant); return isDescendant; } HierarchicalShardSyncer();  HierarchicalShardSyncer(final boolean isMultiStreamMode, final String streamIdentifier); synchronized boolean checkAndCreateLeaseForNewShards(@NonNull final ShardDetector shardDetector,
            final LeaseRefresher leaseRefresher, final InitialPositionInStreamExtended initialPosition,
            final MetricsScope scope, final boolean ignoreUnexpectedChildShards, final boolean isLeaseTableEmpty); synchronized boolean checkAndCreateLeaseForNewShards(@NonNull final ShardDetector shardDetector,
            final LeaseRefresher leaseRefresher, final InitialPositionInStreamExtended initialPosition,
            List<Shard> latestShards, final boolean ignoreUnexpectedChildShards, final MetricsScope scope, final boolean isLeaseTableEmpty); synchronized Lease createLeaseForChildShard(final ChildShard childShard, final StreamIdentifier streamIdentifier); }
HierarchicalShardSyncer { static boolean checkIfDescendantAndAddNewLeasesForAncestors(final String shardId, final InitialPositionInStreamExtended initialPosition, final Set<String> shardIdsOfCurrentLeases, final Map<String, Shard> shardIdToShardMapOfAllKinesisShards, final Map<String, Lease> shardIdToLeaseMapOfNewShards, final MemoizationContext memoizationContext, final MultiStreamArgs multiStreamArgs) { final String streamIdentifier = getStreamIdentifier(multiStreamArgs); final Boolean previousValue = memoizationContext.isDescendant(shardId); if (previousValue != null) { return previousValue; } boolean isDescendant = false; final Set<String> descendantParentShardIds = new HashSet<>(); if (shardId != null && shardIdToShardMapOfAllKinesisShards.containsKey(shardId)) { if (shardIdsOfCurrentLeases.contains(shardId)) { isDescendant = true; } else { final Shard shard = shardIdToShardMapOfAllKinesisShards.get(shardId); final Set<String> parentShardIds = getParentShardIds(shard, shardIdToShardMapOfAllKinesisShards); for (String parentShardId : parentShardIds) { final boolean isParentDescendant = checkIfDescendantAndAddNewLeasesForAncestors(parentShardId, initialPosition, shardIdsOfCurrentLeases, shardIdToShardMapOfAllKinesisShards, shardIdToLeaseMapOfNewShards, memoizationContext, multiStreamArgs); if (isParentDescendant || memoizationContext.shouldCreateLease(parentShardId)) { isDescendant = true; descendantParentShardIds.add(parentShardId); log.debug("{} : Parent shard {} is a descendant.", streamIdentifier, parentShardId); } else { log.debug("{} : Parent shard {} is NOT a descendant.", streamIdentifier, parentShardId); } } if (isDescendant) { for (String parentShardId : parentShardIds) { if (!shardIdsOfCurrentLeases.contains(parentShardId)) { Lease lease = shardIdToLeaseMapOfNewShards.get(parentShardId); if (lease == null) { if (memoizationContext.shouldCreateLease(parentShardId) || !descendantParentShardIds.contains(parentShardId)) { log.debug("{} : Need to create a lease for shardId {}", streamIdentifier, parentShardId); lease = multiStreamArgs.isMultiStreamMode() ? newKCLMultiStreamLease(shardIdToShardMapOfAllKinesisShards.get(parentShardId), multiStreamArgs.streamIdentifier()) : newKCLLease(shardIdToShardMapOfAllKinesisShards.get(parentShardId)); shardIdToLeaseMapOfNewShards.put(parentShardId, lease); } } if (lease != null) { if (descendantParentShardIds.contains(parentShardId) && !initialPosition.getInitialPositionInStream() .equals(InitialPositionInStream.AT_TIMESTAMP)) { lease.checkpoint(ExtendedSequenceNumber.TRIM_HORIZON); } else { lease.checkpoint(convertToCheckpoint(initialPosition)); } } } } } else { if (initialPosition.getInitialPositionInStream().equals(InitialPositionInStream.TRIM_HORIZON) || initialPosition.getInitialPositionInStream() .equals(InitialPositionInStream.AT_TIMESTAMP)) { memoizationContext.setShouldCreateLease(shardId, true); } } } } memoizationContext.setIsDescendant(shardId, isDescendant); return isDescendant; } HierarchicalShardSyncer();  HierarchicalShardSyncer(final boolean isMultiStreamMode, final String streamIdentifier); synchronized boolean checkAndCreateLeaseForNewShards(@NonNull final ShardDetector shardDetector,
            final LeaseRefresher leaseRefresher, final InitialPositionInStreamExtended initialPosition,
            final MetricsScope scope, final boolean ignoreUnexpectedChildShards, final boolean isLeaseTableEmpty); synchronized boolean checkAndCreateLeaseForNewShards(@NonNull final ShardDetector shardDetector,
            final LeaseRefresher leaseRefresher, final InitialPositionInStreamExtended initialPosition,
            List<Shard> latestShards, final boolean ignoreUnexpectedChildShards, final MetricsScope scope, final boolean isLeaseTableEmpty); synchronized Lease createLeaseForChildShard(final ChildShard childShard, final StreamIdentifier streamIdentifier);  }