Update MultiLangDaemon to Support New Features (#118)

Updated the MultiLangDaemon to use the v2 record processor interfaces, and added features to messages passed to MultiLangDaemon clients.

These changes will require updates to the various MultiLangDaemon clients. The changes for the Python version are complete, and other versions will be updated later.

Author: pfifer

Diff for: pom.xml

@@ -63,6 +63,13 @@
       <version>2.6</version>
     </dependency>
 
+    <dependency>
+      <groupId>org.projectlombok</groupId>
+      <artifactId>lombok</artifactId>
+      <version>1.16.10</version>
+      <scope>provided</scope>
+    </dependency>
+
     <!-- Test -->
     <dependency>
       <groupId>junit</groupId>

Diff for: src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java

@@ -120,7 +120,7 @@ public class KinesisClientLibConfiguration {
     /**
      * User agent set when Amazon Kinesis Client Library makes AWS requests.
      */
-    public static final String KINESIS_CLIENT_LIB_USER_AGENT = "amazon-kinesis-client-library-java-1.7.1-SNAPSHOT";
+    public static final String KINESIS_CLIENT_LIB_USER_AGENT = "amazon-kinesis-client-library-java-1.7.2";
 
     /**
      * KCL will validate client provided sequence numbers with a call to Amazon Kinesis before checkpointing for calls

Diff for: src/main/java/com/amazonaws/services/kinesis/multilang/MessageWriter.java

@@ -18,7 +18,6 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.OutputStreamWriter;
-import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
@@ -27,7 +26,8 @@
 import org.apache.commons.logging.LogFactory;
 
 import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;
-import com.amazonaws.services.kinesis.model.Record;
+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;
@@ -119,19 +119,21 @@ private Future<Boolean> writeMessage(Message message) {
     /**
      * Writes an {@link InitializeMessage} to the subprocess.
      * 
-     * @param shardIdToWrite The shard id.
+     * @param initializationInput
+     *            contains information about the shard being initialized
      */
-    Future<Boolean> writeInitializeMessage(String shardIdToWrite) {
-        return writeMessage(new InitializeMessage(shardIdToWrite));
+    Future<Boolean> writeInitializeMessage(InitializationInput initializationInput) {
+        return writeMessage(new InitializeMessage(initializationInput));
     }
 
     /**
      * Writes a {@link ProcessRecordsMessage} message to the subprocess.
      * 
-     * @param records The records to be processed.
+     * @param processRecordsInput
+     *            the records, and associated metadata to be processed.
      */
-    Future<Boolean> writeProcessRecordsMessage(List<Record> records) {
-        return writeMessage(new ProcessRecordsMessage(records));
+    Future<Boolean> writeProcessRecordsMessage(ProcessRecordsInput processRecordsInput) {
+        return writeMessage(new ProcessRecordsMessage(processRecordsInput));
     }
 
     /**
@@ -146,11 +148,16 @@ Future<Boolean> writeShutdownMessage(ShutdownReason reason) {
     /**
      * Writes a {@link CheckpointMessage} to the subprocess.
      * 
-     * @param sequenceNumber The sequence number that was checkpointed.
-     * @param throwable The exception that was thrown by a checkpoint attempt. Null if one didn't occur.
+     * @param sequenceNumber
+     *            The sequence number that was checkpointed.
+     * @param subSequenceNumber
+     *            the sub sequence number to checkpoint at.
+     * @param throwable
+     *            The exception that was thrown by a checkpoint attempt. Null if one didn't occur.
      */
-    Future<Boolean> writeCheckpointMessageWithError(String sequenceNumber, Throwable throwable) {
-        return writeMessage(new CheckpointMessage(sequenceNumber, throwable));
+    Future<Boolean> writeCheckpointMessageWithError(String sequenceNumber, Long subSequenceNumber,
+            Throwable throwable) {
+        return writeMessage(new CheckpointMessage(sequenceNumber, subSequenceNumber, throwable));
     }
 
     /**

Diff for: src/main/java/com/amazonaws/services/kinesis/multilang/MultiLangDaemon.java

@@ -24,6 +24,7 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory;
 import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
 import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker;
 
@@ -71,7 +72,13 @@ public class MultiLangDaemon implements Callable<Integer> {
     public MultiLangDaemon(KinesisClientLibConfiguration configuration,
             MultiLangRecordProcessorFactory recordProcessorFactory,
             ExecutorService workerThreadPool) {
-        this(new Worker(recordProcessorFactory, configuration, workerThreadPool));
+        this(buildWorker(recordProcessorFactory, configuration, workerThreadPool));
+    }
+
+    private static Worker buildWorker(IRecordProcessorFactory recordProcessorFactory,
+            KinesisClientLibConfiguration configuration, ExecutorService workerThreadPool) {
+        return new Worker.Builder().recordProcessorFactory(recordProcessorFactory).config(configuration)
+                .execService(workerThreadPool).build();
     }
 
     /**

Diff for: src/main/java/com/amazonaws/services/kinesis/multilang/MultiLangDaemonConfig.java

@@ -14,17 +14,24 @@
  */
 package com.amazonaws.services.kinesis.multilang;
 
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Properties;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import com.amazonaws.services.kinesis.clientlibrary.config.KinesisClientLibConfigurator;
 import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 /**
  * This class captures the configuration needed to run the MultiLangDaemon.
@@ -131,10 +138,29 @@ private void prepare(String processingLanguage) {
 
     private static Properties loadProperties(ClassLoader classLoader, String propertiesFileName) throws IOException {
         Properties properties = new Properties();
-        try (InputStream propertiesStream = classLoader.getResourceAsStream(propertiesFileName)) {
-            properties.load(propertiesStream);
+        InputStream propertyStream = null;
+        try {
+            propertyStream = classLoader.getResourceAsStream(propertiesFileName);
+            if (propertyStream == null) {
+                File propertyFile = new File(propertiesFileName);
+                if (propertyFile.exists()) {
+                    propertyStream = new FileInputStream(propertyFile);
+                }
+            }
+
+            if (propertyStream == null) {
+                throw new FileNotFoundException(
+                        "Unable to find property file in classpath, or file system: '" + propertiesFileName + "'");
+            }
+
+            properties.load(propertyStream);
             return properties;
+        } finally {
+            if (propertyStream != null) {
+                propertyStream.close();
+            }
         }
+
     }
 
     private static boolean validateProperties(Properties properties) {
@@ -147,13 +173,16 @@ private static int getMaxActiveThreads(Properties properties) {
 
     private static ExecutorService buildExecutorService(Properties properties) {
         int maxActiveThreads = getMaxActiveThreads(properties);
+        ThreadFactoryBuilder builder = new ThreadFactoryBuilder().setNameFormat("multi-lang-daemon-%04d");
         LOG.debug(String.format("Value for %s property is %d", PROP_MAX_ACTIVE_THREADS, maxActiveThreads));
         if (maxActiveThreads <= 0) {
             LOG.info("Using a cached thread pool.");
-            return Executors.newCachedThreadPool();
+            return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
+                    builder.build());
         } else {
             LOG.info(String.format("Using a fixed thread pool with %d max active threads.", maxActiveThreads));
-            return Executors.newFixedThreadPool(maxActiveThreads);
+            return new ThreadPoolExecutor(maxActiveThreads, maxActiveThreads, 0L, TimeUnit.MILLISECONDS,
+                    new LinkedBlockingQueue<Runnable>(), builder.build());
         }
     }