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

Clean up checkpoint state after successful checkpoint

Browse source
This commit is contained in:
Joshua Kim 2020-04-07 03:45:03 -04:00
parent 5355b4b7c5
commit 1794874e33
2 changed files with 2 additions and 0 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/checkpoint/dynamodb/DynamoDBCheckpointer.java
View file

@ -134,6 +134,7 @@ public class DynamoDBCheckpointer implements Checkpointer {
lease.checkpoint(checkpoint); lease.checkpoint(checkpoint);
lease.pendingCheckpoint(null); lease.pendingCheckpoint(null);
lease.pendingCheckpointState(null);
lease.ownerSwitchesSinceCheckpoint(0L); lease.ownerSwitchesSinceCheckpoint(0L);
return leaseCoordinator.updateLease(lease, concurrencyToken, operation, leaseKey); return leaseCoordinator.updateLease(lease, concurrencyToken, operation, leaseKey);

1
amazon-kinesis-client/src/test/java/software/amazon/kinesis/checkpoint/InMemoryCheckpointer.java
View file

@ -44,6 +44,7 @@ public class InMemoryCheckpointer implements Checkpointer {
checkpoints.put(leaseKey, checkpointValue); checkpoints.put(leaseKey, checkpointValue);
flushpoints.put(leaseKey, checkpointValue); flushpoints.put(leaseKey, checkpointValue);
pendingCheckpoints.remove(leaseKey); pendingCheckpoints.remove(leaseKey);
pendingCheckpointStates.remove(leaseKey);
if (log.isDebugEnabled()) { if (log.isDebugEnabled()) {
log.debug("shardId: {} checkpoint: {}", leaseKey, checkpointValue); log.debug("shardId: {} checkpoint: {}", leaseKey, checkpointValue);