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

Clean shutdown when multilang deamen is terminated by giving one last chance to checkpoint.

Browse source
This commit is contained in:
Ikram ulhaq 2017-05-31 23:42:13 -07:00
parent 8d339bdb88
commit 19f3326823
3 changed files with 42 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
src/main/java/com/amazonaws/services/kinesis/multilang/MultiLangDaemon.java
View file

@ -140,11 +140,27 @@ public class MultiLangDaemon implements Callable<Integer> {
ExecutorService executorService = config.getExecutorService();
// Daemon
MultiLangDaemon daemon = new MultiLangDaemon(
final MultiLangDaemon daemon = new MultiLangDaemon(
config.getKinesisClientLibConfiguration(),
config.getRecordProcessorFactory(),
executorService);
Runtime.getRuntime().addShutdownHook(new Thread()
{
@Override
public void run()
{
LOG.info("Process terminanted, will initiate shutdown.");
try {
Future<Void> fut = daemon.worker.requestShutdown();
fut.get();
LOG.info("Process shutdown is complete.");
} catch (InterruptedException | ExecutionException e) {
LOG.error("Encountered an error during shutdown.", e);
}
}
});
Future<Integer> future = executorService.submit(daemon);
try {
System.exit(future.get());

25
src/main/java/com/amazonaws/services/kinesis/multilang/MultiLangRecordProcessor.java
View file

@ -20,6 +20,10 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@ -29,13 +33,14 @@ import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownInput;
import com.fasterxml.jackson.databind.ObjectMapper;
/**
* A record processor that manages creating a child process that implements the multi language protocol and connecting
* that child process's input and outputs to a {@link MultiLangProtocol} object and calling the appropriate methods on
* that object when its corresponding {@link #initialize}, {@link #processRecords}, and {@link #shutdown} methods are
* called.
*/
public class MultiLangRecordProcessor implements IRecordProcessor {
public class MultiLangRecordProcessor implements IRecordProcessor, IShutdownNotificationAware {
private static final Log LOG = LogFactory.getLog(MultiLangRecordProcessor.class);
private static final int EXIT_VALUE = 1;
@ -136,6 +141,24 @@ public class MultiLangRecordProcessor implements IRecordProcessor {
}
}
@Override
public void shutdownRequested(IRecordProcessorCheckpointer checkpointer) {
LOG.info("Shutdown is requested.");
if (!initialized) {
LOG.info("Record processor was not initialized so no need to initiate a final checkpoint.");
return;
}
try {
LOG.info("Requesting a checkpoint on shutdown notification.");
checkpointer.checkpoint();
} catch (InvalidStateException e) {
LOG.error("Checkpoint triggered during shutdown encountered InvalidStateException: " + e.toString());
} catch (ShutdownException e) {
LOG.error("Checkpoint triggered during shutdown encountered ShutdownException: " + e.toString());
}
}
/**
* Used to tell whether the processor has been shutdown already.
*/

1
src/test/java/com/amazonaws/services/kinesis/multilang/StreamingRecordProcessorTest.java
View file

@ -172,6 +172,7 @@ public class StreamingRecordProcessorTest {
recordProcessor.initialize(new InitializationInput().withShardId(shardId));
recordProcessor.processRecords(new ProcessRecordsInput().withRecords(testRecords).withCheckpointer(unimplementedCheckpointer));
recordProcessor.processRecords(new ProcessRecordsInput().withRecords(testRecords).withCheckpointer(unimplementedCheckpointer));
recordProcessor.shutdownRequested(Mockito.mock(IRecordProcessorCheckpointer.class));
recordProcessor.shutdown(new ShutdownInput().withCheckpointer(unimplementedCheckpointer).withShutdownReason(ShutdownReason.ZOMBIE));
}