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

Add some more comments around possible race conditions.

Browse source
This commit is contained in:
Pfifer, Justin 2017-02-14 09:52:00 -08:00
parent bd77f9a7ac
commit bcfc325364
1 changed files with 10 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/Worker.java
View file

@ -538,10 +538,12 @@ public class Worker implements Runnable {
*/ */
public Future<Void> requestShutdown() { public Future<Void> requestShutdown() {
//
// Stop accepting new leases. Once we do this we can be sure that
// no more leases will be acquired.
//
leaseCoordinator.stopLeaseTaker(); leaseCoordinator.stopLeaseTaker();
//
// Stop accepting new leases
//
Collection<KinesisClientLease> leases = leaseCoordinator.getAssignments(); Collection<KinesisClientLease> leases = leaseCoordinator.getAssignments();
if (leases == null || leases.isEmpty()) { if (leases == null || leases.isEmpty()) {
// //
@ -560,6 +562,11 @@ public class Worker implements Runnable {
if (consumer != null) { if (consumer != null) {
consumer.notifyShutdownRequested(shutdownNotification); consumer.notifyShutdownRequested(shutdownNotification);
} else { } else {
//
// There is a race condition between retrieving the current assignments, and creating the
// notification. If the a lease is lost in between these two points, we explicitly decrement the
// notification latches to clear the shutdown.
//
notificationCompleteLatch.countDown(); notificationCompleteLatch.countDown();
shutdownCompleteLatch.countDown(); shutdownCompleteLatch.countDown();
} }