Kinesis: What is the best/safe way to shutdown a worker?

Question

I am using the AWS Kinesis Client Library.

I need a way to shutdown Kinesis Worker thread during deployments so, that I stop at a checkpoint and not in the middle of processRecords().

I see a shutdown boolean present in Worker.java but it is made private.

The reason I need is that checkpointing and idempotency is critical to me and I don't want to kill the process in the middle of a batch.

[EDIT]

Thanks to @CaptainMurphy, I noticed that Worker.java exposes shutdown() method which safely shuts down the worker and the LeaseCoordinator. What it doesn't do is call shutdown() task in the IRecordProcessor. It abruptly terminates the IRecordProcessor without worrying about the state.

I do understand that Idempotency between checkpoints is not guaranteed by the KCL and the developer should make the design fault tolerant but I feel that the IRecordProcessor should be properly shutdown before LeaseCoordinator stops irrespective of that.

Answer 1

Since version 1.7.1 (see note below) application can request a graceful shutdown, and record processors that implement the IShutdownNotificationAware will be given a chance to checkpoint before being shutdown.

• make sure record processor implements IShutdownNotificationAware interface in addition to one of IRecordProcessor interfaces. Call checkpoint in shutdownRequested(IRecordProcessorCheckpointer checkpointer) method. Pay attention - shutdown method of IRecordProcessor should call checkpoint only if shutdown reason is TERMINATE

• on application shutdown initiate worker shutdown process

  Future<Void> shutdown = worker.requestShutdown();
  shutdown.get(); // wait for shutdown complete
  

PS: Kinesis Client before version 1.7.4 contains race condition which prevents correct shutdown. So use version 1.7.4 or later.

Answer 2

The Record Processor shutdown method does indeed get called when you call shutdown on the Worker. You can trace it back from the ShutdownTask class, which is created by the ShardConsumer class, which is closed by the Worker.

So you can checkpoint at the last received record by listening for the shutdown call, passing the checkpointer the last sequence received function at the iteration point of your last processed value. E.g. in your overridden processRecords():

for(Record currRecord : records)
{
    someProcessSingleRecordMethod(currRecord)
    if(shutdown) 
    { 
        checkpointer.checkpoint(currRecord.getSequenceNumber()); 
        return; 
    } 
}

Where your shutdown method sets the shutdown flag to true.

Note that it's still best practice for Kinesis applications to be designed in an "at least once" fashion in the case of an ungraceful shutdown, such as instance termination. Receive and process "just once" may not be a good use case for Kinesis.