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

Update the Worker shutdown logic to make sure that the LeaseCleanup Manager also terminates all the threads that it has started

Browse source
This commit is contained in:
kmz 2021-05-22 14:46:51 +00:00
parent 4f6882cefe
commit d0e21d7915
3 changed files with 32 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/Scheduler.java
View file

@ -825,6 +825,10 @@ public class Scheduler implements Runnable {
// Lost leases will force Worker to begin shutdown process for all shard consumers in
// Worker.run().
leaseCoordinator.stop();
// Stop the lease cleanup manager
leaseCleanupManager.shutdown();
leaderElectedPeriodicShardSyncManager.stop();
workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.SHUT_DOWN);
}

16
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseCleanupManager.java
View file

@ -93,6 +93,22 @@ public class LeaseCleanupManager {
}
}
/**
* Stops the lease cleanup thread, which is scheduled periodically as specified by
* {@link LeaseCleanupManager#leaseCleanupIntervalMillis}
*/
public void shutdown() {
if (isRunning) {
log.info("Stopping the lease cleanup thread.");
completedLeaseStopwatch.stop();
garbageLeaseStopwatch.stop();
deletionThreadPool.shutdown();
isRunning = false;
} else {
log.info("Lease cleanup thread already stopped.");
}
}
/**
* Enqueues a lease for deletion without check for duplicate entry. Use {@link #isEnqueuedForDeletion}
* for checking the duplicate entries.

12
amazon-kinesis-client/src/test/java/software/amazon/kinesis/leases/LeaseCleanupManagerTest.java
View file

@ -94,6 +94,18 @@ public class LeaseCleanupManagerTest {
leaseCleanupManager.start();
}
/**
* Tests subsequent calls to shutdown {@link LeaseCleanupManager}.
*/
@Test
public final void testSubsequentShutdowns() {
leaseCleanupManager.start();
Assert.assertTrue(leaseCleanupManager.isRunning());
leaseCleanupManager.shutdown();
Assert.assertFalse(leaseCleanupManager.isRunning());
leaseCleanupManager.shutdown();
}
/**
* Tests that when both child shard leases are present, we are able to delete the parent shard for the completed
* shard case.