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

Merge d4e2a4b91f into e121691ac2

Browse source
This commit is contained in:
Muhammad Ikramul Haq 2017-06-05 17:26:47 +00:00 committed by GitHub
commit 3a74920f4c
6 changed files with 87 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
src/main/java/com/amazonaws/services/kinesis/multilang/MessageWriter.java
View file

@ -26,13 +26,16 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import com.amazonaws.services.kinesis.multilang.messages.CheckpointMessage;
import com.amazonaws.services.kinesis.multilang.messages.InitializeMessage;
import com.amazonaws.services.kinesis.multilang.messages.Message;
import com.amazonaws.services.kinesis.multilang.messages.ProcessRecordsMessage;
import com.amazonaws.services.kinesis.multilang.messages.ShutdownMessage;
import com.amazonaws.services.kinesis.multilang.messages.ShutdownRequestedMessage;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import com.fasterxml.jackson.databind.ObjectMapper;
/**
@ -145,6 +148,13 @@ class MessageWriter {
return writeMessage(new ShutdownMessage(reason));
}
/**
* Writes a {@link ShutdownRequestedMessage} to the subprocess.
*/
Future<Boolean> writeShutdownRequestedMessage() {
return writeMessage(new ShutdownRequestedMessage());
}
/**
* Writes a {@link CheckpointMessage} to the subprocess.
*

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

@ -16,6 +16,7 @@ package com.amazonaws.services.kinesis.multilang;
import java.io.IOException;
import java.io.PrintStream;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@ -140,11 +141,26 @@ 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(5000, TimeUnit.MILLISECONDS);
LOG.info("Process shutdown is complete.");
} catch (InterruptedException | ExecutionException | TimeoutException e) {
LOG.error("Encountered an error during shutdown.", e);
}
}
});
Future<Integer> future = executorService.submit(daemon);
try {
System.exit(future.get());

13
src/main/java/com/amazonaws/services/kinesis/multilang/MultiLangProtocol.java
View file

@ -27,6 +27,7 @@ import com.amazonaws.services.kinesis.multilang.messages.InitializeMessage;
import com.amazonaws.services.kinesis.multilang.messages.Message;
import com.amazonaws.services.kinesis.multilang.messages.ProcessRecordsMessage;
import com.amazonaws.services.kinesis.multilang.messages.ShutdownMessage;
import com.amazonaws.services.kinesis.multilang.messages.ShutdownRequestedMessage;
import com.amazonaws.services.kinesis.multilang.messages.StatusMessage;
import lombok.extern.apachecommons.CommonsLog;
@ -99,6 +100,18 @@ class MultiLangProtocol {
return waitForStatusMessage(ShutdownMessage.ACTION, checkpointer, writeFuture);
}
/**
* Writes a {@link ShutdownRequestedMessage} to the child process's STDIN and waits for the child process to respond with a
* {@link StatusMessage} on its STDOUT.
*
* @param checkpointer A checkpointer.
* @return Whether or not this operation succeeded.
*/
boolean shutdownRequested(IRecordProcessorCheckpointer checkpointer) {
Future<Boolean> writeFuture = messageWriter.writeShutdownRequestedMessage();
return waitForStatusMessage(ShutdownRequestedMessage.ACTION, checkpointer, writeFuture);
}
/**
* Waits for a {@link StatusMessage} for a particular action. If a {@link CheckpointMessage} is received, then this
* method will attempt to checkpoint with the provided {@link IRecordProcessorCheckpointer}. This method returns

28
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,27 @@ 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.");
// ProcessRecordsInput emptyInput = new ProcessRecordsInput();
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);
}
}
/**
* Used to tell whether the processor has been shutdown already.
*/

17
src/main/java/com/amazonaws/services/kinesis/multilang/messages/ShutdownRequestedMessage.java Normal file
View file

@ -0,0 +1,17 @@
package com.amazonaws.services.kinesis.multilang.messages;
/**
* A message to indicate to the client's process that shutdown is requested.
*/
public class ShutdownRequestedMessage extends Message {
/**
* The name used for the action field in {@link Message}.
*/
public static final String ACTION = "shutdownrequested";
/**
* Convenience constructor.
*/
public ShutdownRequestedMessage() {
}
}

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));
}