Clean shutdown when multilang deamen is terminated by giving one last chance to checkpoint. #174
Conversation
… chance to checkpoint.
| } | ||
| try { | ||
| LOG.info("Requesting a checkpoint on shutdown notification."); | ||
| checkpointer.checkpoint(); |
There was a problem hiding this comment.
Whether to checkpoint is determined by the application, so checkpointing here isn't safe. This should use a similar dispatch mechanism that processRecords uses. It might be possible to shortcut the dispatch at this time, and just make whether to checkpoint a configuration flag.
The dispatch code for processRecords for reference.
| config.getKinesisClientLibConfiguration(), | ||
| config.getRecordProcessorFactory(), | ||
| executorService); | ||
|
|
||
| Runtime.getRuntime().addShutdownHook(new Thread() | ||
| { |
There was a problem hiding this comment.
Can you please match the existing style with the curly brace on the same line.
| LOG.info("Process terminanted, will initiate shutdown."); | ||
| try { | ||
| Future<Void> fut = daemon.worker.requestShutdown(); | ||
| fut.get(); |
There was a problem hiding this comment.
In a worst case scenario this could cause the shutdown to become stuck. It would be best to start with some timeout value of how long to wait for the shutdown request to complete. I usually start with something in the range of 5 seconds. In the future it may make sense to make it configurable.
| LOG.info("Requesting a checkpoint on shutdown notification."); | ||
| checkpointer.checkpoint(); | ||
| } catch (InvalidStateException e) { | ||
| LOG.error("Checkpoint triggered during shutdown encountered InvalidStateException: " + e.toString()); |
There was a problem hiding this comment.
Java will automatically call toString during string concatenation. Additionally for exceptions it's common to pass the exception as the second argument to the logger call. This will allow the logger to emit a stack trace as well e.g.
LOG.error("Checkpoint triggered during shutdown encountered InvalidStateException: " + e, e);
| { | ||
| LOG.info("Process terminanted, will initiate shutdown."); | ||
| try { | ||
| Future<Void> fut = daemon.worker.requestShutdown(); |
There was a problem hiding this comment.
You need to ensure that when requestShutdown is invoked. There is no process records invoked. Need a lock may be ?!
There was a problem hiding this comment.
The way dispatch is handled in the worker already guarantees this. The record processors are represented as a state machine, and all request shutdown does is request that on the change event to go to the Shutdown Notification Requested state. This is the diagram for the state changes.
| /** | ||
| * The name used for the action field in {@link Message}. | ||
| */ | ||
| public static final String ACTION = "shutdownrequested"; |
There was a problem hiding this comment.
Minor: Please use camelCase for the action name for consistency.
| * @param checkpointer A checkpointer. | ||
| * @return Whether or not this operation succeeded. | ||
| */ | ||
| boolean shutdownRequested(IRecordProcessorCheckpointer checkpointer) { |
There was a problem hiding this comment.
Could you please add some tests for this method in MultiLangProtocolTest.java
|
Thanks everything looks good. Starting the process to merge the changes now. |
|
One last thing: Please confirm that we can use, modify, copy, and redistribute this contribution. Thanks. |
|
@pfifer - yes confirmed. You can modify, copy and redistribute this contribution. Thanks, |
* Added support for graceful shutdown in MultiLang Clients * PR awslabs#174 * PR awslabs#182 * Updated documentation for `v2.IRecordProcessor#shutdown`, and `KinesisClientLibConfiguration#idleTimeBetweenReadsMillis` * PR awslabs#170 * Updated to version 1.11.151 of the AWS Java SDK * PR awslabs#183
No description provided.