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

Shutdown requested from plumbing in multi lang record processor.

Browse source
This commit is contained in:
Ikram ulhaq 2017-06-05 13:10:10 -07:00
parent 7be1059b9a
commit b5946cf19b
2 changed files with 3 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -148,16 +148,9 @@ public class MultiLangRecordProcessor implements IRecordProcessor, IShutdownNoti
LOG.info("Record processor was not initialized so no need to initiate a final checkpoint."); LOG.info("Record processor was not initialized so no need to initiate a final checkpoint.");
return; return;
} }
try {
LOG.info("Requesting a checkpoint on shutdown notification."); LOG.info("Requesting a checkpoint on shutdown notification.");
// ProcessRecordsInput emptyInput = new ProcessRecordsInput(); if (!protocol.shutdownRequested(checkpointer)) {
LOG.error("Child process failed to shutdown");
checkpointer.checkpoint();
} catch (InvalidStateException e) {
LOG.error("Checkpoint triggered during shutdown encountered InvalidStateException: " + e, e);
} catch (ShutdownException e) {
LOG.error("Checkpoint triggered during shutdown encountered ShutdownException: " + e, e);
} }
} }

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

@ -172,7 +172,6 @@ public class StreamingRecordProcessorTest {
recordProcessor.initialize(new InitializationInput().withShardId(shardId)); recordProcessor.initialize(new InitializationInput().withShardId(shardId));
recordProcessor.processRecords(new ProcessRecordsInput().withRecords(testRecords).withCheckpointer(unimplementedCheckpointer)); recordProcessor.processRecords(new ProcessRecordsInput().withRecords(testRecords).withCheckpointer(unimplementedCheckpointer));
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)); recordProcessor.shutdown(new ShutdownInput().withCheckpointer(unimplementedCheckpointer).withShutdownReason(ShutdownReason.ZOMBIE));
} }