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

Quick fix

Browse source
This commit is contained in:
Chunxue Yang 2020-04-05 13:36:25 -07:00
parent 2f9ce0ce4f
commit 4b3c717c53
2 changed files with 2 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -173,6 +173,7 @@ public class HierarchicalShardSyncer {
} else { } else {
throw new InvalidStateException("Unable to populate new lease for child shard " + childShard.shardId() + "because parent shards cannot be found."); throw new InvalidStateException("Unable to populate new lease for child shard " + childShard.shardId() + "because parent shards cannot be found.");
} }
newLease.checkpoint(ExtendedSequenceNumber.TRIM_HORIZON);
newLease.ownerSwitchesSinceCheckpoint(0L); newLease.ownerSwitchesSinceCheckpoint(0L);
return newLease; return newLease;
} }

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

@ -71,8 +71,6 @@ public class ShutdownTask implements ConsumerTask {
@NonNull @NonNull
private final InitialPositionInStreamExtended initialPositionInStream; private final InitialPositionInStreamExtended initialPositionInStream;
private final boolean cleanupLeasesOfCompletedShards; private final boolean cleanupLeasesOfCompletedShards;
private final boolean garbageCollectLeases = false;
private final boolean isLeaseTableEmpty = false;
private final boolean ignoreUnexpectedChildShards; private final boolean ignoreUnexpectedChildShards;
@NonNull @NonNull
private final LeaseCoordinator leaseCoordinator; private final LeaseCoordinator leaseCoordinator;
@ -174,7 +172,7 @@ public class ShutdownTask implements ConsumerTask {
private void createLeasesForChildShardsIfNotExist() private void createLeasesForChildShardsIfNotExist()
throws DependencyException, InvalidStateException, ProvisionedThroughputException { throws DependencyException, InvalidStateException, ProvisionedThroughputException {
for(ChildShard childShard : childShards) { for(ChildShard childShard : childShards) {
if(leaseCoordinator.getCurrentlyHeldLease(shardInfo.shardId()) == null) { if(leaseCoordinator.getCurrentlyHeldLease(childShard.shardId()) == null) {
final Lease leaseToCreate = hierarchicalShardSyncer.createLeaseForChildShard(childShard); final Lease leaseToCreate = hierarchicalShardSyncer.createLeaseForChildShard(childShard);
leaseCoordinator.leaseRefresher().createLeaseIfNotExists(leaseToCreate); leaseCoordinator.leaseRefresher().createLeaseIfNotExists(leaseToCreate);
} }