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

Change log level from debug to warn when update stale leases

Browse source 
Change log level from debug to warn when update stale leases
This commit is contained in:
Yu Zeng 2021-12-23 13:17:00 -08:00
parent a323272a97
commit 72aeac3819
2 changed files with 23 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/dynamodb/DynamoDBLeaseCoordinator.java
View file

@ -47,7 +47,8 @@ import software.amazon.kinesis.metrics.MetricsFactory;
import software.amazon.kinesis.metrics.MetricsLevel; import software.amazon.kinesis.metrics.MetricsLevel;
import software.amazon.kinesis.metrics.MetricsScope; import software.amazon.kinesis.metrics.MetricsScope;
import software.amazon.kinesis.metrics.MetricsUtil; import software.amazon.kinesis.metrics.MetricsUtil;
import static software.amazon.kinesis.common.CommonCalculations.*;
import static software.amazon.kinesis.common.CommonCalculations.getRenewerTakerIntervalMillis;
/** /**
* LeaseCoordinator abstracts away LeaseTaker and LeaseRenewer from the application code that's using leasing. It owns * LeaseCoordinator abstracts away LeaseTaker and LeaseRenewer from the application code that's using leasing. It owns

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

@ -41,7 +41,8 @@ import software.amazon.kinesis.metrics.MetricsFactory;
import software.amazon.kinesis.metrics.MetricsLevel; import software.amazon.kinesis.metrics.MetricsLevel;
import software.amazon.kinesis.metrics.MetricsScope; import software.amazon.kinesis.metrics.MetricsScope;
import software.amazon.kinesis.metrics.MetricsUtil; import software.amazon.kinesis.metrics.MetricsUtil;
import static software.amazon.kinesis.common.CommonCalculations.*;
import static software.amazon.kinesis.common.CommonCalculations.getRenewerTakerIntervalMillis;
/** /**
* An implementation of {@link LeaseTaker} that uses DynamoDB via {@link LeaseRefresher}. * An implementation of {@link LeaseTaker} that uses DynamoDB via {@link LeaseRefresher}.
@ -51,7 +52,7 @@ import static software.amazon.kinesis.common.CommonCalculations.*;
public class DynamoDBLeaseTaker implements LeaseTaker { public class DynamoDBLeaseTaker implements LeaseTaker {
private static final int TAKE_RETRIES = 3; private static final int TAKE_RETRIES = 3;
private static final int SCAN_RETRIES = 1; private static final int SCAN_RETRIES = 1;
private long veryOldLeaseDurationNanosMultiplier = 3; private static final double RENEWAL_SLACK_PERCENTAGE = 0.5;
// See note on TAKE_LEASES_DIMENSION(Callable) for why we have this callable. // See note on TAKE_LEASES_DIMENSION(Callable) for why we have this callable.
private static final Callable<Long> SYSTEM_CLOCK_CALLABLE = System::nanoTime; private static final Callable<Long> SYSTEM_CLOCK_CALLABLE = System::nanoTime;
@ -64,15 +65,15 @@ public class DynamoDBLeaseTaker implements LeaseTaker {
private final long leaseRenewalIntervalMillis; private final long leaseRenewalIntervalMillis;
private final MetricsFactory metricsFactory; private final MetricsFactory metricsFactory;
private final double RENEWAL_SLACK_PERCENTAGE = 0.5; final Map<String, Lease> allLeases = new HashMap<>();
private final Map<String, Lease> allLeases = new HashMap<>();
// TODO: Remove these defaults and use the defaults in the config // TODO: Remove these defaults and use the defaults in the config
private int maxLeasesForWorker = Integer.MAX_VALUE; private int maxLeasesForWorker = Integer.MAX_VALUE;
private int maxLeasesToStealAtOneTime = 1; private int maxLeasesToStealAtOneTime = 1;
private long veryOldLeaseDurationNanosMultiplier = 3;
private long lastScanTimeNanos = 0L; private long lastScanTimeNanos = 0L;
public DynamoDBLeaseTaker(LeaseRefresher leaseRefresher, String workerIdentifier, long leaseDurationMillis, public DynamoDBLeaseTaker(LeaseRefresher leaseRefresher, String workerIdentifier, long leaseDurationMillis,
final MetricsFactory metricsFactory) { final MetricsFactory metricsFactory) {
this.leaseRefresher = leaseRefresher; this.leaseRefresher = leaseRefresher;
@ -259,7 +260,8 @@ public class DynamoDBLeaseTaker implements LeaseTaker {
try { try {
return leaseRefresher.getLease(lease.leaseKey()); return leaseRefresher.getLease(lease.leaseKey());
} catch (DependencyException | InvalidStateException | ProvisionedThroughputException e) { } catch (DependencyException | InvalidStateException | ProvisionedThroughputException e) {
log.debug("Unable to retrieve the current lease, defaulting to existing lease", e); log.warn("Unable to retrieve the current lease while refreshing the stale lease, "
+ "defaulting to existing lease", e);
} }
} }
return lease; return lease;
@ -599,11 +601,7 @@ public class DynamoDBLeaseTaker implements LeaseTaker {
} }
// If I have no leases, I wasn't represented in leaseCounts. Let's fix that. // If I have no leases, I wasn't represented in leaseCounts. Let's fix that.
Integer myCount = leaseCounts.get(workerIdentifier); leaseCounts.putIfAbsent(workerIdentifier, 0);
if (myCount == null) {
myCount = 0;
leaseCounts.put(workerIdentifier, myCount);
}
return leaseCounts; return leaseCounts;
} }