KCL version 1.6.3 doesn't shutdown worker cleanly

[matthewbogner]

I saw in 1.6.2 release notes that the ability to cleanly shutdown a worker was implemented.

I've put together a reproduction that will demonstrate that this still throws an exception by the worker when attempting to perform the final checkpoint.

See the README for instructions on running the repro scenario:
https://github.com/matthewbogner/kinesis-stress-example

During shutdown of the consumer, you'll see the following exception which shouldn't be occurring:

2016-06-22 17:03:47,979 ERROR [pool-2-thread-1] [net.ibogner.kinesis.RecordProcessor:41] Failed to checkpoint
com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException: Can't update checkpoint - instance doesn't hold the lease for this shard
    at com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibLeaseCoordinator.setCheckpoint(KinesisClientLibLeaseCoordinator.java:172) ~[amazon-kinesis-client-1.6.3.jar:na]
    at com.amazonaws.services.kinesis.clientlibrary.lib.worker.RecordProcessorCheckpointer.advancePosition(RecordProcessorCheckpointer.java:216) ~[amazon-kinesis-client-1.6.3.jar:na]
    at com.amazonaws.services.kinesis.clientlibrary.lib.worker.RecordProcessorCheckpointer.checkpoint(RecordProcessorCheckpointer.java:77) ~[amazon-kinesis-client-1.6.3.jar:na]
    at net.ibogner.kinesis.RecordProcessor.checkpoint(RecordProcessor.java:39) [classes/:na]
    at net.ibogner.kinesis.RecordProcessor.processRecords(RecordProcessor.java:25) [classes/:na]
    at com.amazonaws.services.kinesis.clientlibrary.lib.worker.ProcessTask.call(ProcessTask.java:169) [amazon-kinesis-client-1.6.3.jar:na]
    at com.amazonaws.services.kinesis.clientlibrary.lib.worker.MetricsCollectingTaskDecorator.call(MetricsCollectingTaskDecorator.java:49) [amazon-kinesis-client-1.6.3.jar:na]
    at com.amazonaws.services.kinesis.clientlibrary.lib.worker.MetricsCollectingTaskDecorator.call(MetricsCollectingTaskDecorator.java:24) [amazon-kinesis-client-1.6.3.jar:na]
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_71]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [na:1.8.0_71]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [na:1.8.0_71]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_71]

[rankinc]

I have also experienced shutdown problems with 1.6.3, but I don't think that this is one of them. Does this change to your example program help? (You might also want to apply pull #82 to the library first).

--- a/src/main/java/net/ibogner/kinesis/KinesisStressExample.java
+++ b/src/main/java/net/ibogner/kinesis/KinesisStressExample.java
@@ -53,6 +53,7 @@ public class KinesisStressExample {
     private final AmazonKinesisClient kinesisClient;

     private Worker worker;
+    private Thread workerThread;

     public KinesisStressExample(String streamName, String regionName, Integer numThreads, Integer numRecordsPerPut, Long numMessagesToProduce, String credentialsProfileName) {
         this.streamName = streamName;
@@ -92,7 +93,8 @@ public class KinesisStressExample {
                                                           kclWorkerId)
                                 .withRegionName(regionName))
                 .build();
-        new Thread(worker).start();
+        workerThread = new Thread(worker);
+        workerThread.start();
     }

     public void awaitCompletion() throws InterruptedException {
@@ -100,6 +102,7 @@ public class KinesisStressExample {
             Thread.sleep(1000);
         }
         worker.shutdown();
+        workerThread.join();
         logger.info("Done shutting down worker");

         keepGoing = false;

[pfifer]

Thanks for reporting this.

I'm looking into this, and the provided repro should help.

[matthewbogner]

Thanks!

[matthewbogner]

How is the investigation going ?

[pfifer]

The shutdown of the worker effectively triggers that instance of the worker to lose all it's leases. Lease loss causes the event loop to call shutdown, instead of process records, on the next iteration. The event loop in worker will only dispatch a single task for a shard at a time. So if a processRecords call is being handled when shutdown is called, nothing will happen until the processRecords call completes. At the same time the lease manager marks the lease as no longer held. So if the record processor calls checkpoint it will receive an exception. This is the body of the race condition, and why your test program shows it reasonably well. The test program calls checkpoint on every processRecords calls. The only way the test program wouldn't cause an exception was if the call to Worker.shutdown() occurs before the next call to ShardConsumer.getNextTask.

I have some ideas on how to improve shutdown, but I still need to think through them some more.

There are some approaches that would allow you to the shutdown behavior you expect. Most of them would require that processRecords determine when it's time to shutdown, and stop processing records.

[matthewbogner]

Would it make sense to implement a "quiesce" of sorts, so that existing processRecords could complete and perform final checkpointing before the Lease manager destroys itself and erases all records of who owns what ?

[pfifer]

That's pretty much what I'm thinking. There is some difficulty in handling blocked record processors. The other complexity is figuring out how to fit this in while maintaining compatibility with the existing record processors.

[pfifer]

I have a variant of the Kinesis client library that supports quiescing: https://github.com/pfifer/amazon-kinesis-client/tree/determinstic-shutdown

I still haven't done any of the unit tests, but have tested the shutdown behavior with my test application. I still need track down, and verify the edge cases especially around handling end of shard scenarios.

[matthewbogner]

Thanks for the update. I'm still interested in having the solution, when you feel it is mature and well tested.

[pfifer]

I'm hoping to have it complete, and ready for merging later this week. I don't think it will make the 1.7.0 release, but should be ready shortly after that.

[pfifer]

Graceful shutdown is now merged, and released. It make take a bit for 1.7.1 to become available on Central though.

[shushantarora]

still failing in 1.7.2
When called worker.shutdown() , it throws exception in recordprocessor shutdown and shutdown reason is Zoombie.

com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException: Can't update checkpoint - instance doesn't hold the lease for this shard
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibLeaseCoordinator.setCheckpoint(KinesisClientLibLeaseCoordinator.java:173)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.RecordProcessorCheckpointer.advancePosition(RecordProcessorCheckpointer.java:216)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.RecordProcessorCheckpointer.checkpoint(RecordProcessorCheckpointer.java:77)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.ProcessTask.call(ProcessTask.java:176)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.MetricsCollectingTaskDecorator.call(MetricsCollectingTaskDecorator.java:49)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.MetricsCollectingTaskDecorator.call(MetricsCollectingTaskDecorator.java:24)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

00:27:00,893 INFO pool-4-thread-1 Worker:shouldShutdown:634 - All record processors have been shutdown successfully.
00:27:00,893 INFO pool-4-thread-1 Worker:finalShutdown:610 - Starting worker's final shutdown.
00:27:00,893 INFO pool-4-thread-1 CWPublisherRunnable:shutdown:165 - Shutting down CWPublication thread.
00:27:00,926 INFO cw-metrics-publisher CWPublisherRunnable:run:96 - CWPublication thread finished.
00:27:00,927 INFO pool-4-thread-1 Worker:run:368 - Worker loop is complete. Exiting from worker.

[pfifer]

To use graceful shutdown you must implement a new interface on your record processor: https://github.com/awslabs/amazon-kinesis-client/blob/master/src/main/java/com/amazonaws/services/kinesis/clientlibrary/interfaces/v2/IShutdownNotificationAware.java

You still can't checkpoint in shutdown when the reason is ZOMBIE.

[akumariiit]

Hi @pfifer ...i am trying the gracefulshutdown, but it is just waiting for record processors to finish, not sure where it is blocked, processor do not take that much time to complete one batch.

shutdown notification, and 9 record processor to complete final shutdown
INFO [2018-09-27 04:15:20,605] com.amazonaws.services.kinesis.clientlibrary.lib.worker.GracefulShutdownCoordinator$GracefulShutdownCallable: Waiting for 9 record process to complete shutdown notification, and 9 record processor to complete final shutdown
INFO [2018-09-27 04:15:21,609] com.amazonaws.services.kinesis.clientlibrary.lib.worker.GracefulShutdownCoordinator$GracefulShutdownCallable: Waiting for 9 record process to complete shutdown notification, and 9 record processor to complete final shutdown
INFO [2018-09-27 04:15:22,612] com.amazonaws.services.kinesis.clientlibrary.lib.worker.GracefulShutdownCoordinator$GracefulShutdownCallable: Waiting for 9 record process to complete shutdown notification, and 9 record processor to complete final shutdown
INFO [2018-09-27 04:15:23,614] com.amazonaws.services.kinesis.clientlibrary.lib.worker.GracefulShutdownCoordinator$GracefulShutdownCallable: Waiting for 9 record process to complete shutdown notification, and 9 record processor to complete final shutdown
INFO [2018-09-27 04:15:24,618] com.amazonaws.services.kinesis.clientlibrary.lib.worker.GracefulShutdownCoordinator$GracefulShutdownCallable: Waiting for 9 record process to complete shutdown notification, and 9 record processor to complete final shutdown
INFO [2018-09-27 04:15:25,622] com.amazonaws.services.kinesis.clientlibrary.lib.worker.GracefulShutdownCoordinator$GracefulShutdownCallable: Waiting for 9 record process to complete shutdown notification, and 9 record processor to complete final shutdown
INFO [2018-09-27 04:15:26,623] com.amazonaws.services.kinesis.clientlibrary.lib.worker.GracefulShutdownCoordinator$GracefulShutdownCallable: Waiting for 9 record process to complete shutdown notification, and 9 record processor to complete final shutdown

[stsatlantis]

@akumariiit How did you solve your issue with the graceful shutdown? it doesn't seem to terminate, like ever.

[kaisermario]

@akumariiit @stsatlantis We have the same issue. Any update on this? I don't know which process is blocking this shutdown procedure.
@pfifer