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

Fix flaky HashRangesAreAlwaysComplete test

Browse source 
Updates depth value for Merge and Reshard with some In-Progress Parents tests to prevent invalid hierarchy trees.
This commit is contained in:
furq-aws 2023-03-16 04:52:56 -07:00
parent c23d83dc9a
commit 28a21d776b
1 changed files with 5 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
amazon-kinesis-client/src/test/java/software/amazon/kinesis/coordinator/PeriodicShardSyncManagerTest.java
View file

@ -58,6 +58,8 @@ import static software.amazon.kinesis.leases.LeaseManagementConfig.DEFAULT_CONSE
public class PeriodicShardSyncManagerTest { public class PeriodicShardSyncManagerTest {
private static final int MAX_DEPTH_WITH_IN_PROGRESS_PARENTS = 1;
private StreamIdentifier streamIdentifier; private StreamIdentifier streamIdentifier;
private PeriodicShardSyncManager periodicShardSyncManager; private PeriodicShardSyncManager periodicShardSyncManager;
@Mock @Mock
@ -446,7 +448,7 @@ public class PeriodicShardSyncManagerTest {
for (int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
int maxInitialLeaseCount = 100; int maxInitialLeaseCount = 100;
List<Lease> leases = generateInitialLeases(maxInitialLeaseCount); List<Lease> leases = generateInitialLeases(maxInitialLeaseCount);
reshard(leases, 5, ReshardType.MERGE, maxInitialLeaseCount, true); reshard(leases, MAX_DEPTH_WITH_IN_PROGRESS_PARENTS, ReshardType.MERGE, maxInitialLeaseCount, true);
Collections.shuffle(leases); Collections.shuffle(leases);
Assert.assertFalse(periodicShardSyncManager.hasHoleInLeases(streamIdentifier, leases).isPresent()); Assert.assertFalse(periodicShardSyncManager.hasHoleInLeases(streamIdentifier, leases).isPresent());
} }
@ -457,17 +459,9 @@ public class PeriodicShardSyncManagerTest {
for (int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
int maxInitialLeaseCount = 100; int maxInitialLeaseCount = 100;
List<Lease> leases = generateInitialLeases(maxInitialLeaseCount); List<Lease> leases = generateInitialLeases(maxInitialLeaseCount);
reshard(leases, 5, ReshardType.ANY, maxInitialLeaseCount, true); reshard(leases, MAX_DEPTH_WITH_IN_PROGRESS_PARENTS, ReshardType.ANY, maxInitialLeaseCount, true);
Collections.shuffle(leases); Collections.shuffle(leases);
boolean isHoleInHashRanges = periodicShardSyncManager.hasHoleInLeases(streamIdentifier, leases).isPresent(); Assert.assertFalse(periodicShardSyncManager.hasHoleInLeases(streamIdentifier, leases).isPresent());
if (isHoleInHashRanges) {
// In-progress parents may result in a lease having the highest startingHashKey value while not having
// an endingHashKey value of MAX_HASH_KEY. This is detected as a "hole" in the hash key range,
// so mark the in-progress parents as finished (SHARD_END) and recheck for holes.
finishInProgressParents(leases);
isHoleInHashRanges = periodicShardSyncManager.hasHoleInLeases(streamIdentifier, leases).isPresent();
}
Assert.assertFalse(isHoleInHashRanges);
} }
} }
@ -577,13 +571,6 @@ public class PeriodicShardSyncManagerTest {
return leaseCounter; return leaseCounter;
} }
private void finishInProgressParents(List<Lease> leases) {
leases.stream()
.filter(l -> l.checkpoint() != null && !l.checkpoint().equals(ExtendedSequenceNumber.SHARD_END) &&
l.childShardIds() != null && !l.childShardIds().isEmpty())
.forEach(l -> l.checkpoint(ExtendedSequenceNumber.SHARD_END));
}
private boolean isHeads() { private boolean isHeads() {
return Math.random() <= 0.5; return Math.random() <= 0.5;
} }