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

Code refactoring - 1

Browse source
This commit is contained in:
Ashwin Giridharan 2020-06-15 21:20:33 -07:00
parent 37281e9493
commit b60dd60f35
2 changed files with 9 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/dynamodb/DynamoDBLeaseRenewer.java
View file

@ -298,14 +298,12 @@ public class DynamoDBLeaseRenewer implements LeaseRenewer {
}
final MetricsScope scope = MetricsUtil.createMetricsWithOperation(metricsFactory, operation);
if (StringUtils.isNotEmpty(singleStreamShardId)) {
if(lease instanceof MultiStreamLease) {
MetricsUtil.addStreamId(scope,
StreamIdentifier.multiStreamInstance(((MultiStreamLease) lease).streamIdentifier()));
MetricsUtil.addShardId(scope, ((MultiStreamLease) lease).shardId());
} else {
MetricsUtil.addShardId(scope, singleStreamShardId);
}
if (lease instanceof MultiStreamLease) {
MetricsUtil.addStreamId(scope,
StreamIdentifier.multiStreamInstance(((MultiStreamLease) lease).streamIdentifier()));
MetricsUtil.addShardId(scope, ((MultiStreamLease) lease).shardId());
} else if (StringUtils.isNotEmpty(singleStreamShardId)) {
MetricsUtil.addShardId(scope, singleStreamShardId);
}
long startTime = System.currentTimeMillis();

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

@ -132,9 +132,6 @@ public class ShutdownTask implements ConsumerTask {
final Lease leaseFromDdb = Optional.ofNullable(leaseCoordinator.leaseRefresher().getLease(leaseKeyProvider.apply(shardInfo)))
.orElseThrow(() -> new IllegalStateException("Lease for shard " + leaseKeyProvider.apply(shardInfo) + " does not exist."));
if (!leaseFromDdb.checkpoint().equals(ExtendedSequenceNumber.SHARD_END)) {
recordProcessorCheckpointer
.sequenceNumberAtShardEnd(recordProcessorCheckpointer.largestPermittedCheckpointValue());
recordProcessorCheckpointer.largestPermittedCheckpointValue(ExtendedSequenceNumber.SHARD_END);
// Call the shardRecordsProcessor to checkpoint with SHARD_END sequence number.
// The shardEnded is implemented by customer. We should validate if the SHARD_END checkpointing is successful after calling shardEnded.
throwOnApplicationException(() -> applicationCheckpointAndVerification(), scope, startTime);
@ -167,8 +164,6 @@ public class ShutdownTask implements ConsumerTask {
} catch (Exception e) {
log.error("Unable to cleanup garbage lease {} for {}", currentShardLease.leaseKey(),
streamIdentifier, e);
} finally {
}
}
}
@ -203,6 +198,9 @@ public class ShutdownTask implements ConsumerTask {
}
private void applicationCheckpointAndVerification() {
recordProcessorCheckpointer
.sequenceNumberAtShardEnd(recordProcessorCheckpointer.largestPermittedCheckpointValue());
recordProcessorCheckpointer.largestPermittedCheckpointValue(ExtendedSequenceNumber.SHARD_END);
shardRecordProcessor.shardEnded(ShardEndedInput.builder().checkpointer(recordProcessorCheckpointer).build());
final ExtendedSequenceNumber lastCheckpointValue = recordProcessorCheckpointer.lastCheckpointValue();
if (lastCheckpointValue == null