partitionsFor(String topic);
}
diff --git a/kafka/kafka-common/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaExactlyOnceOutputOperator.java b/kafka/kafka-common/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaExactlyOnceOutputOperator.java
new file mode 100644
index 0000000000..ec99931b0e
--- /dev/null
+++ b/kafka/kafka-common/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaExactlyOnceOutputOperator.java
@@ -0,0 +1,415 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.apex.malhar.kafka;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.apex.malhar.lib.wal.FSWindowDataManager;
+import org.apache.apex.malhar.lib.wal.WindowDataManager;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator;
+
+import static org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
+
+/**
+ * Kafka output operator with exactly once processing semantics.
+ *
+ *
+ *
+ * Requirements
+ *
In the Kafka message, only Value will be available for users
+ * Users need to provide Value deserializers for Kafka message as it is used during recovery
+ * Value type should have well defined Equals & HashCodes,
+ * as during messages are stored in HashMaps for comparison.
+ *
+ * Recovery handling
+ *
Offsets of the Kafka partitions are stored in the WindowDataManager at the endWindow
+ * During recovery,
+ *
+ * - Partially written Streaming Window before the crash is constructed. ( Explained below )
+ * - Tuples from the completed Streaming Window's are skipped
+ * - Tuples coming for the partially written Streaming Window are skipped.
+ * (No assumption is made on the order and the uniqueness of the tuples)
+ *
+ *
+ *
+ *
+ *
+ * Partial Window Construction
+ *
Operator uses the Key in the Kafka message, which is not available for use by the operator users.
+ * Key is used to uniquely identify the message written by the particular instance of this operator.
+ * This allows multiple writers to same Kafka partitions. Format of the key is "APPLICATTION_ID#OPERATOR_ID".
+ * During recovery Kafka partitions are read between the latest offset and the last written offsets.
+ * All the tuples written by the particular instance is kept in the Map
+ *
+ *
+ *
+ * Limitations
+ *
Key in the Kafka message is reserved for Operator's use
+ * During recovery, operator needs to read tuples between 2 offsets,
+ * if there are lot of data to be read, Operator may
+ * appear to be blocked to the Stram and can kill the operator.
+ *
+ *
+ * @displayName Kafka Single Port Exactly Once Output(0.9.0)
+ * @category Messaging
+ * @tags output operator
+ * @since 3.5.0
+ */
+@org.apache.hadoop.classification.InterfaceStability.Evolving
+public abstract class AbstractKafkaExactlyOnceOutputOperator extends AbstractKafkaOutputOperator
+ implements Operator.CheckpointNotificationListener
+{
+ private transient String key;
+ private transient String appName;
+ private transient Integer operatorId;
+ private transient Long windowId;
+ private transient Map partialWindowTuples = new HashMap<>();
+ private transient AbstractKafkaConsumer consumer;
+
+ private WindowDataManager windowDataManager = new FSWindowDataManager();
+ private final int KAFKA_CONNECT_ATTEMPT = 10;
+ private final String KEY_SEPARATOR = "#";
+
+ public static final String KEY_DESERIALIZER = "org.apache.kafka.common.serialization.StringDeserializer";
+ public static final String KEY_SERIALIZER = "org.apache.kafka.common.serialization.StringSerializer";
+
+ public final transient DefaultInputPort inputPort = new DefaultInputPort()
+ {
+ @Override
+ public void process(T tuple)
+ {
+ sendTuple(tuple);
+ }
+ };
+
+ // Creates the consumer object and it wraps KafkaConsumer.
+ public abstract AbstractKafkaConsumer createConsumer(Properties prop);
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KEY_SERIALIZER);
+
+ if (getProperties().getProperty(VALUE_DESERIALIZER_CLASS_CONFIG) == null) {
+ throw new IllegalArgumentException(
+ "Value deserializer needs to be set for the operator, as it is used during recovery.");
+ }
+
+ super.setup(context);
+
+ this.operatorId = context.getId();
+ this.windowDataManager.setup(context);
+ this.appName = context.getValue(Context.DAGContext.APPLICATION_NAME);
+ this.key = appName + KEY_SEPARATOR + (new Integer(operatorId));
+
+ this.consumer = KafkaConsumerInit();
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ this.windowId = windowId;
+
+ if (windowId == windowDataManager.getLargestCompletedWindow()) {
+ rebuildPartialWindow();
+ }
+ }
+
+ @Override
+ public void checkpointed(long windowId)
+ {
+ }
+
+ @Override
+ public void committed(long windowId)
+ {
+ try {
+ windowDataManager.committed(windowId);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void beforeCheckpoint(long windowId)
+ {
+ }
+
+ @Override
+ public void teardown()
+ {
+ consumer.close();
+ super.teardown();
+ }
+
+ @Override
+ public void endWindow()
+ {
+ if (windowId <= windowDataManager.getLargestCompletedWindow()) {
+ return;
+ }
+
+ if (!partialWindowTuples.isEmpty()) {
+ throw new RuntimeException("Violates Exactly once. Not all the tuples received after operator reset.");
+ }
+
+ // Every tuples should be written before the offsets are stored in the window data manager.
+ getProducer().flush();
+
+ try {
+ this.windowDataManager.save(getPartitionsAndOffsets(true), windowId);
+ } catch (IOException | InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public WindowDataManager getWindowDataManager()
+ {
+ return windowDataManager;
+ }
+
+ public void setWindowDataManager(WindowDataManager windowDataManager)
+ {
+ this.windowDataManager = windowDataManager;
+ }
+
+ private boolean doesKeyBelongsToThisInstance(int operatorId, String key)
+ {
+ String[] split = key.split(KEY_SEPARATOR);
+
+ if (split.length != 2) {
+ return false;
+ }
+
+ if ((Integer.parseInt(split[1]) == operatorId) && (split[0].equals(appName))) {
+ return true;
+ }
+
+ return false;
+ }
+
+ private boolean alreadyInKafka(T message)
+ {
+ if (windowId <= windowDataManager.getLargestCompletedWindow()) {
+ return true;
+ }
+
+ if (partialWindowTuples.containsKey(message)) {
+
+ Integer val = partialWindowTuples.get(message);
+
+ if (val == 0) {
+ return false;
+ } else if (val == 1) {
+ partialWindowTuples.remove(message);
+ } else {
+ partialWindowTuples.put(message, val - 1);
+ }
+ return true;
+ }
+ return false;
+ }
+
+ private Map getPartitionsAndOffsets(boolean latest) throws ExecutionException, InterruptedException
+ {
+ List partitionInfoList = consumer.partitionsFor(getTopic());
+ List topicPartitionList = new ArrayList<>();
+
+ for (PartitionInfo partitionInfo : partitionInfoList) {
+ topicPartitionList.add(new TopicPartition(getTopic(), partitionInfo.partition()));
+ }
+
+ Map parttionsAndOffset = new HashMap<>();
+ consumer.assignPartitions(topicPartitionList);
+
+ for (PartitionInfo partitionInfo : partitionInfoList) {
+ try {
+ TopicPartition topicPartition = new TopicPartition(getTopic(), partitionInfo.partition());
+ if (latest) {
+ consumer.seekToEnd(topicPartition);
+ } else {
+ consumer.seekToBeginning(topicPartition);
+ }
+ parttionsAndOffset.put(partitionInfo.partition(), consumer.positionPartition(topicPartition));
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ return parttionsAndOffset;
+ }
+
+ private void rebuildPartialWindow()
+ {
+ logger.info("Rebuild the partial window after " + windowDataManager.getLargestCompletedWindow());
+
+ Map storedOffsets;
+ Map currentOffsets;
+
+ try {
+ storedOffsets = (Map)this.windowDataManager.retrieve(windowId);
+ currentOffsets = getPartitionsAndOffsets(true);
+ } catch (IOException | ExecutionException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+
+ if (currentOffsets == null) {
+ logger.info("No tuples found while building partial window " + windowDataManager.getLargestCompletedWindow());
+ return;
+ }
+
+ if (storedOffsets == null) {
+
+ logger.info("Stored offset not available, seeking to the beginning of the Kafka Partition.");
+
+ try {
+ storedOffsets = getPartitionsAndOffsets(false);
+ } catch (ExecutionException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ List topicPartitions = new ArrayList<>();
+
+ for (Map.Entry entry : currentOffsets.entrySet()) {
+ topicPartitions.add(new TopicPartition(getTopic(), entry.getKey()));
+ }
+
+ consumer.assignPartitions(topicPartitions);
+
+ for (Map.Entry entry : currentOffsets.entrySet()) {
+ Long storedOffset = 0L;
+ Integer currentPartition = entry.getKey();
+ Long currentOffset = entry.getValue();
+
+ if (storedOffsets.containsKey(currentPartition)) {
+ storedOffset = storedOffsets.get(currentPartition);
+ }
+
+ if (storedOffset >= currentOffset) {
+ continue;
+ }
+
+ try {
+ consumer.seekToOffset(new TopicPartition(getTopic(), currentPartition), storedOffset);
+ } catch (Exception ex) {
+ logger.info("Rebuilding of the partial window is not complete, exactly once recovery is not possible.");
+ throw new RuntimeException(ex);
+ }
+
+ int kafkaAttempt = 0;
+
+ while (true) {
+
+ ConsumerRecords consumerRecords = consumer.pollRecords(100);
+
+ if (consumerRecords.count() == 0) {
+ if (kafkaAttempt++ == KAFKA_CONNECT_ATTEMPT) {
+ break;
+ }
+ } else {
+ kafkaAttempt = 0;
+ }
+
+ boolean crossedBoundary = false;
+
+ for (ConsumerRecord consumerRecord : consumerRecords) {
+
+ if (consumerRecord.offset() >= currentOffset) {
+ crossedBoundary = true;
+ break;
+ }
+
+ if (!doesKeyBelongsToThisInstance(operatorId, consumerRecord.key())) {
+ continue;
+ }
+
+ T value = consumerRecord.value();
+
+ if (partialWindowTuples.containsKey(value)) {
+ Integer count = partialWindowTuples.get(value);
+ partialWindowTuples.put(value, count + 1);
+ } else {
+ partialWindowTuples.put(value, 1);
+ }
+
+ }
+
+ if (crossedBoundary) {
+ break;
+ }
+ }
+ }
+ }
+
+ private AbstractKafkaConsumer KafkaConsumerInit()
+ {
+ Properties props = new Properties();
+
+ props.put(BOOTSTRAP_SERVERS_CONFIG, getProperties().get(BOOTSTRAP_SERVERS_CONFIG));
+ props.put(KEY_DESERIALIZER_CLASS_CONFIG, KEY_DESERIALIZER);
+ props.put(VALUE_DESERIALIZER_CLASS_CONFIG, getProperties().get(VALUE_DESERIALIZER_CLASS_CONFIG));
+
+ return createConsumer(props);
+ }
+
+ protected void sendTuple(T tuple)
+ {
+ if (alreadyInKafka(tuple)) {
+ return;
+ }
+
+ getProducer().send(new ProducerRecord<>(getTopic(), key, tuple), new Callback()
+ {
+ public void onCompletion(RecordMetadata metadata, Exception e)
+ {
+ if (e != null) {
+ logger.info("Wrting to Kafka failed with an exception {}" + e.getMessage());
+ throw new RuntimeException(e);
+ }
+ }
+ });
+ }
+
+ private static final Logger logger = LoggerFactory.getLogger(AbstractKafkaExactlyOnceOutputOperator.class);
+}
+
diff --git a/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/AbstractEmbeddedKafka.java b/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/AbstractEmbeddedKafka.java
new file mode 100644
index 0000000000..cd690625d7
--- /dev/null
+++ b/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/AbstractEmbeddedKafka.java
@@ -0,0 +1,256 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.apex.malhar.kafka;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+
+import org.I0Itec.zkclient.ZkClient;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
+import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.apache.zookeeper.server.ZooKeeperServer;
+
+import kafka.server.KafkaServer;
+import kafka.utils.TestUtils;
+import kafka.utils.ZKStringSerializer$;
+import kafka.utils.ZkUtils;
+
+public abstract class AbstractEmbeddedKafka
+{
+ private static final String[] KAFKA_PATH = new String[]{"/tmp/kafka-test1","/tmp/kafka-test2"};
+
+ private ZkClient[] zkClient = new ZkClient[2];
+ private ZkUtils[] zkUtils = new ZkUtils[2];
+ private String BROKERHOST = "localhost";
+ private ZooKeeperServer[] zkServer = new ZooKeeperServer[2];
+ private KafkaServer[] kafkaServer = new KafkaServer[2];
+ public static int[] TEST_ZOOKEEPER_PORT;
+ public static int[] TEST_KAFKA_BROKER_PORT;
+ public static String baseDir = "target";
+ private int clusterId = 0;
+ public static final String TEST_TOPIC = "testtopic";
+ public static int testCounter = 0;
+ public static final String END_TUPLE = "END_TUPLE";
+
+ private static final String zkBaseDir = "zookeeper-server-data";
+ private static final String kafkaBaseDir = "kafka-server-data";
+ private static final String[] zkdir = new String[]{"zookeeper-server-data1", "zookeeper-server-data2"};
+ private static final String[] zklogdir = new String[]{"zookeeper-log-data1", "zookeeper-log-data2"};
+ private static ServerCnxnFactory[] zkFactory = new ServerCnxnFactory[2];
+ static final org.slf4j.Logger logger = LoggerFactory.getLogger(AbstractEmbeddedKafka.class);
+
+ // get available ports
+ private void getAvailablePorts()
+ {
+ ServerSocket[] listeners = new ServerSocket[4];
+ int[] p = new int[4];
+
+ try {
+ for (int i = 0; i < 4; i++) {
+ listeners[i] = new ServerSocket(0);
+ p[i] = listeners[i].getLocalPort();
+ }
+
+ for (int i = 0; i < 4; i++) {
+ listeners[i].close();
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ TEST_ZOOKEEPER_PORT = new int[]{p[0], p[1]};
+ TEST_KAFKA_BROKER_PORT = new int[]{p[2], p[3]};
+ }
+
+ public void startZookeeper(int clusterId)
+ {
+ try {
+ int numConnections = 100;
+ int tickTime = 2000;
+ File snapshotDir;
+ File logDir;
+ try {
+ snapshotDir = java.nio.file.Files.createTempDirectory(zkdir[clusterId]).toFile();
+ logDir = java.nio.file.Files.createTempDirectory(zklogdir[clusterId]).toFile();
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to start Kafka", e);
+ }
+
+ snapshotDir.deleteOnExit();
+ logDir.deleteOnExit();
+ zkServer[clusterId] = new ZooKeeperServer(snapshotDir, logDir, tickTime);
+ zkFactory[clusterId] = new NIOServerCnxnFactory();
+ zkFactory[clusterId].configure(new InetSocketAddress(TEST_ZOOKEEPER_PORT[clusterId]), numConnections);
+
+ zkFactory[clusterId].startup(zkServer[clusterId]); // start the zookeeper server.
+ Thread.sleep(2000);
+ //kserver.startup();
+ } catch (Exception ex) {
+ logger.error(ex.getLocalizedMessage());
+ }
+ }
+
+ public void stopZookeeper(int clusterId)
+ {
+ zkServer[clusterId].shutdown();
+ zkFactory[clusterId].closeAll();
+ zkFactory[clusterId].shutdown();
+ }
+
+ public String getBroker()
+ {
+ return getBroker(clusterId);
+ }
+
+ public String getBroker(int clusterId)
+ {
+ return BROKERHOST + ":" + TEST_KAFKA_BROKER_PORT[clusterId];
+ }
+
+ public void start() throws IOException
+ {
+ getAvailablePorts();
+ FileUtils.deleteDirectory(new File(KAFKA_PATH[0]));
+ FileUtils.deleteDirectory(new File(KAFKA_PATH[1]));
+ // Setup Zookeeper
+ startZookeeper(0);
+ startZookeeper(1);
+ // Setup brokers
+ cleanupDir();
+ startBroker(0);
+ startBroker(1);
+ }
+
+ public abstract KafkaServer createKafkaServer(Properties prop);
+
+ public void startBroker(int clusterId)
+ {
+ String zkConnect = BROKERHOST + ":" + TEST_ZOOKEEPER_PORT[clusterId];
+ zkClient[clusterId] = new ZkClient(zkConnect, 30000, 30000, ZKStringSerializer$.MODULE$);
+ zkUtils[clusterId] = ZkUtils.apply(zkClient[clusterId], false);
+
+ Properties props = new Properties();
+ props.setProperty("zookeeper.connect", zkConnect);
+ props.setProperty("broker.id", "" + clusterId);
+ props.setProperty("log.dirs", KAFKA_PATH[clusterId]);
+ props.setProperty("listeners", "PLAINTEXT://" + BROKERHOST + ":" + TEST_KAFKA_BROKER_PORT[clusterId]);
+ kafkaServer[clusterId] = createKafkaServer(props);
+
+ }
+
+ public void stopBroker(int clusterId)
+ {
+ kafkaServer[clusterId].shutdown();
+ zkClient[clusterId].close();
+ }
+
+ public void stop() throws IOException
+ {
+ stopBroker(0);
+ stopBroker(1);
+ stopZookeeper(0);
+ stopZookeeper(1);
+ cleanupDir();
+ }
+
+ private void cleanupDir() throws IOException
+ {
+ FileUtils.deleteDirectory(new File(KAFKA_PATH[0]));
+ FileUtils.deleteDirectory(new File(KAFKA_PATH[1]));
+ }
+
+ public abstract void createTopic(String topic, ZkUtils zkUtils, int noOfPartitions);
+
+ public void createTopic(String topic)
+ {
+ createTopic(topic, clusterId, 1);
+ }
+
+ public void createTopic(String topic, int clusterId, int numOfPartitions)
+ {
+ createTopic(topic, zkUtils[clusterId], numOfPartitions);
+ List servers = new ArrayList();
+ servers.add(kafkaServer[clusterId]);
+ TestUtils.waitUntilMetadataIsPropagated(scala.collection.JavaConversions.asScalaBuffer(servers), topic, 0, 30000);
+ }
+
+ public void publish(String topic, List messages)
+ {
+ Properties producerProps = new Properties();
+ producerProps.setProperty("bootstrap.servers", BROKERHOST + ":" + TEST_KAFKA_BROKER_PORT[clusterId]);
+ producerProps.setProperty("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
+ producerProps.setProperty("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
+
+ try (KafkaProducer producer = new KafkaProducer<>(producerProps)) {
+ for (String message : messages) {
+ ProducerRecord data = new ProducerRecord<>(topic, message.getBytes(StandardCharsets.UTF_8));
+ producer.send(data);
+ }
+ }
+
+ List servers = new ArrayList();
+ servers.add(kafkaServer[clusterId]);
+ TestUtils.waitUntilMetadataIsPropagated(scala.collection.JavaConversions.asScalaBuffer(servers), topic, 0, 30000);
+ }
+
+ public List consume(String topic, int timeout)
+ {
+ return consume(topic, timeout, true);
+ }
+
+ public List consume(String topic, int timeout, boolean earliest)
+ {
+ Properties consumerProps = new Properties();
+ consumerProps.setProperty("bootstrap.servers", BROKERHOST + ":" + TEST_KAFKA_BROKER_PORT[clusterId]);
+ consumerProps.setProperty("group.id", "group0");
+ consumerProps.setProperty("client.id", "consumer0");
+ consumerProps.setProperty("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
+ consumerProps.setProperty("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+ // to make sure the consumer starts from the beginning of the topic
+ consumerProps.put("auto.offset.reset", earliest ? "earliest" : "latest");
+ KafkaConsumer consumer = new KafkaConsumer<>(consumerProps);
+ consumer.subscribe(Arrays.asList(topic));
+
+ List messages = new ArrayList<>();
+
+ ConsumerRecords records = consumer.poll(timeout);
+ for (ConsumerRecord record : records) {
+ messages.add(new String(record.value()));
+ }
+
+ consumer.close();
+
+ return messages;
+ }
+}
diff --git a/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperatorTest.java b/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperatorTest.java
new file mode 100644
index 0000000000..2c3aedda6a
--- /dev/null
+++ b/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperatorTest.java
@@ -0,0 +1,399 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.apex.malhar.kafka;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestWatcher;
+import org.junit.runners.Parameterized;
+import org.slf4j.LoggerFactory;
+
+import org.apache.apex.malhar.lib.wal.FSWindowDataManager;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.LocalMode;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.stram.StramLocalCluster;
+
+/**
+ * A bunch of test to verify the input operator will be automatically partitioned
+ * per kafka partition This test is launching its
+ * own Kafka cluster.
+ */
+public abstract class AbstractKafkaInputOperatorTest
+{
+
+ private int totalBrokers = 0;
+
+ private String partition = null;
+
+ public String testName = "";
+ protected boolean hasMultiPartition = false;
+ protected boolean hasMultiCluster = false;
+
+ public static String APPLICATION_PATH = AbstractEmbeddedKafka.baseDir + File.separator + StramLocalCluster.class.getName() + File.separator;
+
+ public class KafkaTestInfo extends TestWatcher
+ {
+ public org.junit.runner.Description desc;
+
+ public String getDir()
+ {
+ String methodName = desc.getMethodName();
+ String className = desc.getClassName();
+ return "target/" + className + "/" + methodName + "/" + testName;
+ }
+
+ @Override
+ protected void starting(org.junit.runner.Description description)
+ {
+ this.desc = description;
+ }
+ }
+
+ public static AbstractEmbeddedKafka kafkaServer;
+
+ @Rule
+ public final KafkaTestInfo testInfo = new KafkaTestInfo();
+
+ @Parameterized.Parameters(name = "multi-cluster: {0}, multi-partition: {1}, partition: {2}")
+ public static Collection testScenario()
+ {
+ return Arrays.asList(new Object[][]{
+ {true, false, "one_to_one"},// multi cluster with single partition
+ {true, false, "one_to_many"},
+ {true, true, "one_to_one"},// multi cluster with multi partitions
+ {true, true, "one_to_many"},
+ {false, true, "one_to_one"}, // single cluster with multi partitions
+ {false, true, "one_to_many"},
+ {false, false, "one_to_one"}, // single cluster with single partitions
+ {false, false, "one_to_many"}
+ });
+ }
+
+ @Before
+ public void before()
+ {
+ testName = AbstractEmbeddedKafka.TEST_TOPIC + AbstractEmbeddedKafka.testCounter++;
+ logger.info("before() test case: {}", testName);
+ tupleCollection.clear();
+ //reset count for next new test case
+ k = 0;
+
+ int noOfPartitions = hasMultiPartition ? 2 : 1;
+ kafkaServer.createTopic(testName, 0, noOfPartitions);
+ if (hasMultiCluster) {
+ kafkaServer.createTopic(testName, 1, noOfPartitions);
+ }
+ }
+
+ public AbstractKafkaInputOperatorTest(boolean hasMultiCluster, boolean hasMultiPartition, String partition)
+ {
+ // This class want to initialize several kafka brokers for multiple partitions
+ this.hasMultiCluster = hasMultiCluster;
+ this.hasMultiPartition = hasMultiPartition;
+ int cluster = 1 + (hasMultiCluster ? 1 : 0);
+ totalBrokers = (1 + (hasMultiPartition ? 1 : 0)) * cluster;
+ this.partition = partition;
+ }
+
+ private static final org.slf4j.Logger logger = LoggerFactory.getLogger(AbstractKafkaInputOperatorTest.class);
+ private static List tupleCollection = new LinkedList<>();
+
+ /**
+ * whether countDown latch count all tuples or just END_TUPLE
+ */
+ private static final boolean countDownAll = false;
+ private static final int scale = 2;
+ private static final int totalCount = 10 * scale;
+ private static final int failureTrigger = 3 * scale;
+ private static final int tuplesPerWindow = 5 * scale;
+ private static final int waitTime = 60000 + 300 * scale;
+
+ //This latch was used to count the END_TUPLE, but the order of tuple can't be guaranteed,
+ //so, count valid tuple instead.
+ private static CountDownLatch latch;
+ private static boolean hasFailure = false;
+ private static int k = 0;
+ private static Thread monitorThread;
+
+ /**
+ * Test Operator to collect tuples from KafkaSingleInputStringOperator.
+ *
+ * @param
+ */
+ public static class CollectorModule extends BaseOperator
+ {
+ public final transient DefaultInputPort inputPort = new DefaultInputPort()
+ {
+ @Override
+ public void process(byte[] bt)
+ {
+ processTuple(bt);
+ }
+ };
+
+ long currentWindowId;
+
+ long operatorId;
+
+ boolean isIdempotentTest = false;
+
+ transient List windowTupleCollector = Lists.newArrayList();
+ private transient Map> tupleCollectedInWindow = new HashMap<>();
+ private int endTuples = 0;
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ super.setup(context);
+ operatorId = context.getId();
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ currentWindowId = windowId;
+ windowTupleCollector.clear();
+ endTuples = 0;
+ }
+
+ public void processTuple(byte[] bt)
+ {
+ String tuple = new String(bt);
+ if (hasFailure && k++ == failureTrigger) {
+ //you can only kill yourself once
+ hasFailure = false;
+ throw new RuntimeException();
+ }
+ if (tuple.startsWith(AbstractEmbeddedKafka.END_TUPLE)) {
+ endTuples++;
+ }
+
+ windowTupleCollector.add(tuple);
+ }
+
+ @Override
+ public void endWindow()
+ {
+ super.endWindow();
+ if (isIdempotentTest) {
+ String key = operatorId + "," + currentWindowId;
+ List msgsInWin = tupleCollectedInWindow.get(key);
+ if (msgsInWin != null) {
+ Assert.assertEquals(
+ "replay messages should be exactly same as previous window", msgsInWin, windowTupleCollector);
+ } else {
+ List newList = Lists.newArrayList();
+ newList.addAll(windowTupleCollector);
+ tupleCollectedInWindow.put(key, newList);
+ }
+ }
+
+ //discard the tuples of this window if except happened
+ int tupleSize = windowTupleCollector.size();
+ tupleCollection.addAll(windowTupleCollector);
+
+ int countDownTupleSize = countDownAll ? tupleSize : endTuples;
+
+ if (latch != null) {
+ Assert.assertTrue(
+ "received END_TUPLES more than expected.", latch.getCount() >= countDownTupleSize);
+ // If it is a failure test, then clear the tupleCollection and don't decrement the count of latch.
+ // Because tupleCollection and latch are static global values
+ if (hasFailure && latch.getCount() != countDownTupleSize) {
+ tupleCollection.clear();
+ return;
+ }
+ while (countDownTupleSize > 0) {
+ latch.countDown();
+ --countDownTupleSize;
+ }
+ if (latch.getCount() == 0) {
+ /**
+ * The time before countDown() and the shutdown() of the application
+ * will cause fatal error:
+ * "Catastrophic Error: Invalid State - the operator blocked forever!"
+ * as the activeQueues could be cleared but alive haven't changed yet.
+ * throw the ShutdownException to let the engine shutdown;
+ */
+ try {
+ throw new ShutdownException();
+ //lc.shutdown();
+ } finally {
+ /**
+ * interrupt the engine thread, let it wake from sleep and handle
+ * the shutdown at this time, all payload should be handled. so it
+ * should be ok to interrupt
+ */
+ monitorThread.interrupt();
+ }
+ }
+ }
+ }
+
+ }
+
+ /**
+ * Test AbstractKafkaSinglePortInputOperator (i.e. an input adapter for Kafka, aka consumer). This module receives
+ * data from an outside test generator through Kafka message bus and feed that data into Malhar streaming platform.
+ *
+ * [Generate message and send that to Kafka message bus] ==> [Receive that message through Kafka input adapter(i.e.
+ * consumer) and send using emitTuples() interface on output port]
+ *
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testInputOperator() throws Exception
+ {
+ hasFailure = false;
+ testInputOperator(false, false);
+ }
+
+ @Test
+ public void testInputOperatorWithFailure() throws Exception
+ {
+ hasFailure = true;
+ testInputOperator(true, false);
+ }
+
+ @Test
+ public void testIdempotentInputOperatorWithFailure() throws Exception
+ {
+ hasFailure = true;
+ testInputOperator(true, true);
+ }
+
+ public abstract AbstractKafkaInputOperator createKafkaInputOperator(DAG dag, DefaultInputPort inputPort);
+
+ public void testInputOperator(boolean hasFailure, boolean idempotent) throws Exception
+ {
+ // each broker should get a END_TUPLE message
+ latch = new CountDownLatch(countDownAll ? totalCount + totalBrokers : totalBrokers);
+ logger.info(
+ "Test Case: name: {}; totalBrokers: {}; hasFailure: {}; hasMultiCluster: {};" +
+ " hasMultiPartition: {}, partition: {}",
+ testName, totalBrokers, hasFailure, hasMultiCluster, hasMultiPartition, partition);
+
+ // Start producer
+ KafkaTestProducer p = new KafkaTestProducer(testName, hasMultiPartition, hasMultiCluster, kafkaServer);
+ p.setSendCount(totalCount);
+ Thread t = new Thread(p);
+ t.start();
+
+ int expectedReceiveCount = totalCount + totalBrokers;
+
+ // Create DAG for testing.
+ LocalMode lma = LocalMode.newInstance();
+ DAG dag = lma.getDAG();
+
+ // Create Test tuple collector
+ CollectorModule collector = dag.addOperator("TestMessageCollector", CollectorModule.class);
+ collector.isIdempotentTest = idempotent;
+
+ // Create KafkaSinglePortStringInputOperator
+ AbstractKafkaInputOperator node = createKafkaInputOperator(dag, collector.inputPort);
+ node.setInitialPartitionCount(1);
+ // set topic
+ node.setTopics(testName);
+ node.setInitialOffset(AbstractKafkaInputOperator.InitialOffset.EARLIEST.name());
+ node.setClusters(getClusterConfig());
+ node.setStrategy(partition);
+ if (idempotent) {
+ node.setWindowDataManager(new FSWindowDataManager());
+ }
+
+ if (hasFailure) {
+ setupHasFailureTest(node, dag);
+ }
+
+ // Create local cluster
+ LocalMode.Controller lc = lma.getController();
+ lc.setHeartbeatMonitoringEnabled(false);
+
+ //let the Controller to run the inside another thread. It is almost same as call Controller.runAsync(),
+ //but Controller.runAsync() don't expose the thread which run it,
+ //so we don't know when the thread will be terminated.
+ //create this thread and then call join() to make sure the Controller shutdown completely.
+ monitorThread = new Thread((StramLocalCluster)lc, "master");
+ monitorThread.start();
+
+ boolean notTimeout = true;
+ try {
+ // Wait 60s for consumer finish consuming all the messages
+ notTimeout = latch.await(waitTime, TimeUnit.MILLISECONDS);
+ lc.shutdown();
+
+ //wait until control thread finished.
+ monitorThread.join();
+ } catch (Exception e) {
+ logger.warn(e.getMessage());
+ }
+
+ t.join();
+
+ if (!notTimeout || expectedReceiveCount != tupleCollection.size()) {
+ logger.info("Number of received/expected tuples: {}/{}, testName: {}, tuples: \n{}", tupleCollection.size(),
+ expectedReceiveCount, testName, tupleCollection);
+ }
+ Assert.assertTrue("TIMEOUT. testName: " + this.testName + "; Collected data: "
+ + tupleCollection, notTimeout);
+
+ // Check results
+ Assert.assertTrue("testName: " + testName + "; Collected tuple size: " + tupleCollection.size()
+ + "; Expected tuple size: " + expectedReceiveCount + "; data: \n" + tupleCollection,
+ expectedReceiveCount == tupleCollection.size());
+
+ logger.info("End of test case: {}", testName);
+ }
+
+ private void setupHasFailureTest(AbstractKafkaInputOperator operator, DAG dag)
+ {
+ operator.setHoldingBufferSize(5000);
+ dag.setAttribute(Context.DAGContext.CHECKPOINT_WINDOW_COUNT, 1);
+ //dag.setAttribute(Context.OperatorContext.STORAGE_AGENT, new FSStorageAgent(
+ // APPLICATION_PATH + "failureck", new Configuration()));
+ operator.setMaxTuplesPerWindow(tuplesPerWindow);
+ }
+
+ private String getClusterConfig()
+ {
+ return kafkaServer.getBroker(0) +
+ (hasMultiCluster ? ";" + kafkaServer.getBroker(1) : "");
+ }
+
+}
diff --git a/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/AbstractKafkaOutputOperatorTest.java b/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/AbstractKafkaOutputOperatorTest.java
new file mode 100644
index 0000000000..3bfeab5132
--- /dev/null
+++ b/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/AbstractKafkaOutputOperatorTest.java
@@ -0,0 +1,427 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.apex.malhar.kafka;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.apex.malhar.lib.wal.FSWindowDataManager;
+import org.apache.commons.io.FileUtils;
+import org.apache.kafka.clients.producer.ProducerConfig;
+
+import com.datatorrent.api.Attribute;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.LocalMode;
+import com.datatorrent.api.Operator;
+import com.datatorrent.common.util.BaseOperator;
+
+import static com.datatorrent.lib.helper.OperatorContextTestHelper.mockOperatorContext;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
+
+public abstract class AbstractKafkaOutputOperatorTest
+{
+ String testName;
+ private static List tupleCollection = new LinkedList<>();
+ public static AbstractEmbeddedKafka kafkaserver;
+ private final String VALUE_DESERIALIZER = "org.apache.apex.malhar.kafka.KafkaHelper";
+ private final String VALUE_SERIALIZER = "org.apache.apex.malhar.kafka.KafkaHelper";
+ protected boolean hasMultiCluster = false;
+
+ public static String APPLICATION_PATH = AbstractEmbeddedKafka.baseDir + File.separator + "MyKafkaApp" + File.separator;
+
+ @Before
+ public void before()
+ {
+ FileUtils.deleteQuietly(new File(APPLICATION_PATH));
+ testName = AbstractEmbeddedKafka.TEST_TOPIC + AbstractEmbeddedKafka.testCounter++;
+ kafkaserver.createTopic(testName, 0, 1);
+ if (hasMultiCluster) {
+ kafkaserver.createTopic(testName, 1, 1);
+ }
+ }
+
+ @After
+ public void after()
+ {
+ FileUtils.deleteQuietly(new File(APPLICATION_PATH));
+ }
+
+ @Test
+ public void testExactlyOnceWithFailure() throws Exception
+ {
+ List toKafka = GenerateList();
+
+ sendDataToKafka(true, toKafka, true, false);
+
+ List fromKafka = ReadFromKafka();
+
+ Assert.assertTrue("With Failure", compare(fromKafka, toKafka));
+ }
+
+ @Test
+ public void testExactlyOnceWithNoFailure() throws Exception
+ {
+ List toKafka = GenerateList();
+
+ sendDataToKafka(true, toKafka, false, false);
+
+ List fromKafka = ReadFromKafka();
+
+ Assert.assertTrue("With No Failure", compare(fromKafka, toKafka));
+ }
+
+ @Test
+ public void testExactlyOnceWithDifferentTuplesAfterRecovery() throws Exception
+ {
+ List toKafka = GenerateList();
+
+ try {
+ sendDataToKafka(true, toKafka, true, true);
+ } catch (RuntimeException ex) {
+
+ boolean expectedException = false;
+ if (ex.getMessage().contains("Violates")) {
+ expectedException = true;
+ }
+
+ Assert.assertTrue("Different tuples after recovery", expectedException);
+ return;
+ }
+
+ Assert.assertTrue("Wrong tuples during replay, should throw exception", false);
+ }
+
+ @Test
+ public void testKafkaOutput() throws Exception
+ {
+ List toKafka = GenerateList();
+
+ sendDataToKafka(false, toKafka, false, false);
+
+ List fromKafka = ReadFromKafka();
+
+ Assert.assertTrue("No failure", compare(fromKafka, toKafka));
+ }
+
+ @Test
+ public void testKafkaOutputWithFailure() throws Exception
+ {
+ List toKafka = GenerateList();
+
+ sendDataToKafka(false, toKafka, true, true);
+
+ List fromKafka = ReadFromKafka();
+
+ Assert.assertTrue("No failure", fromKafka.size() > toKafka.size());
+ }
+
+ private void sendDataToKafka(boolean exactlyOnce, List toKafka, boolean hasFailure,
+ boolean differentTuplesAfterRecovery) throws InterruptedException
+ {
+ Properties props = new Properties();
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, VALUE_SERIALIZER);
+ if (!exactlyOnce) {
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, AbstractKafkaExactlyOnceOutputOperator.KEY_SERIALIZER);
+ }
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, getClusterConfig());
+ props.put(VALUE_DESERIALIZER_CLASS_CONFIG, VALUE_DESERIALIZER);
+
+ Attribute.AttributeMap attributeMap = new Attribute.AttributeMap.DefaultAttributeMap();
+ attributeMap.put(Context.DAGContext.APPLICATION_NAME, "MyKafkaApp");
+ attributeMap.put(DAG.APPLICATION_PATH, APPLICATION_PATH);
+
+ OperatorContext operatorContext = mockOperatorContext(2, attributeMap);
+
+ cleanUp(operatorContext);
+
+ Operator kafkaOutput;
+ DefaultInputPort inputPort;
+
+ if (exactlyOnce) {
+ AbstractKafkaExactlyOnceOutputOperator kafkaOutputTemp =
+ ResetKafkaOutput(testName, props, operatorContext);
+ inputPort = kafkaOutputTemp.inputPort;
+ kafkaOutput = kafkaOutputTemp;
+ } else {
+ KafkaSinglePortOutputOperator kafkaOutputTemp =
+ ResetKafkaSimpleOutput(testName, props, operatorContext);
+ inputPort = kafkaOutputTemp.inputPort;
+ kafkaOutput = kafkaOutputTemp;
+ }
+
+ kafkaOutput.beginWindow(1);
+ inputPort.getSink().put(toKafka.get(0));
+ inputPort.getSink().put(toKafka.get(1));
+ inputPort.getSink().put(toKafka.get(2));
+ kafkaOutput.endWindow();
+ kafkaOutput.beginWindow(2);
+ inputPort.getSink().put(toKafka.get(3));
+ inputPort.getSink().put(toKafka.get(4));
+ inputPort.getSink().put(toKafka.get(5));
+ kafkaOutput.endWindow();
+ kafkaOutput.beginWindow(3);
+ inputPort.getSink().put(toKafka.get(6));
+ inputPort.getSink().put(toKafka.get(7));
+
+ if (hasFailure) {
+ if (exactlyOnce) {
+ AbstractKafkaExactlyOnceOutputOperator kafkaOutputTemp =
+ ResetKafkaOutput(testName, props, operatorContext);
+ inputPort = kafkaOutputTemp.inputPort;
+ kafkaOutput = kafkaOutputTemp;
+ } else {
+ KafkaSinglePortOutputOperator kafkaOutputTemp =
+ ResetKafkaSimpleOutput(testName, props, operatorContext);
+ inputPort = kafkaOutputTemp.inputPort;
+ kafkaOutput = kafkaOutputTemp;
+ }
+
+ kafkaOutput.beginWindow(2);
+ inputPort.getSink().put(toKafka.get(3));
+ inputPort.getSink().put(toKafka.get(4));
+ inputPort.getSink().put(toKafka.get(5));
+ kafkaOutput.endWindow();
+ kafkaOutput.beginWindow(3);
+ inputPort.getSink().put(toKafka.get(6));
+
+ if (!differentTuplesAfterRecovery) {
+ inputPort.getSink().put(toKafka.get(7));
+ }
+ }
+
+ inputPort.getSink().put(toKafka.get(8));
+ inputPort.getSink().put(toKafka.get(9));
+ kafkaOutput.endWindow();
+ kafkaOutput.beginWindow(4);
+ inputPort.getSink().put(toKafka.get(10));
+ inputPort.getSink().put(toKafka.get(11));
+ kafkaOutput.endWindow();
+
+ cleanUp(operatorContext);
+ }
+
+ public abstract AbstractKafkaExactlyOnceOutputOperator createExaactlyOnceOutputOperator();
+
+ public abstract AbstractKafkaInputOperator createKafkaInputOperator(DAG dag, DefaultInputPort inputPort);
+
+ private AbstractKafkaExactlyOnceOutputOperator ResetKafkaOutput(
+ String testName, Properties props, OperatorContext operatorContext)
+ {
+ AbstractKafkaExactlyOnceOutputOperator kafkaOutput = createExaactlyOnceOutputOperator();
+ kafkaOutput.setTopic(testName);
+ kafkaOutput.setProperties(props);
+ kafkaOutput.setup(operatorContext);
+
+ return kafkaOutput;
+ }
+
+ private KafkaSinglePortOutputOperator ResetKafkaSimpleOutput(
+ String testName, Properties props, OperatorContext operatorContext)
+ {
+ KafkaSinglePortOutputOperator kafkaOutput = new KafkaSinglePortOutputOperator<>();
+ kafkaOutput.setTopic(testName);
+ kafkaOutput.setProperties(props);
+ kafkaOutput.setup(operatorContext);
+
+ return kafkaOutput;
+ }
+
+ private void cleanUp(OperatorContext operatorContext)
+ {
+ FSWindowDataManager windowDataManager = new FSWindowDataManager();
+ windowDataManager.setup(operatorContext);
+ try {
+ windowDataManager.committed(windowDataManager.getLargestCompletedWindow());
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ private boolean compare(List fromKafka, List toKafka)
+ {
+ if (fromKafka.size() != toKafka.size()) {
+ return false;
+ }
+
+ for (int i = 0; i < fromKafka.size(); ++i) {
+ if (!fromKafka.get(i).equals(toKafka.get(i))) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ private String getClusterConfig()
+ {
+ return kafkaserver.getBroker(0) +
+ (hasMultiCluster ? ";" + kafkaserver.getBroker(1) : "");
+ }
+
+ private List GenerateList()
+ {
+ List people = new ArrayList<>();
+
+ for (Integer i = 0; i < 12; ++i) {
+ people.add(new Person(i.toString(), i));
+ }
+
+ return people;
+ }
+
+ private List ReadFromKafka()
+ {
+ tupleCollection.clear();
+
+ // Create KafkaSinglePortStringInputOperator
+ Properties props = new Properties();
+ props.put(BOOTSTRAP_SERVERS_CONFIG, getClusterConfig());
+ props.put(KEY_DESERIALIZER_CLASS_CONFIG, AbstractKafkaExactlyOnceOutputOperator.KEY_DESERIALIZER);
+ props.put(VALUE_DESERIALIZER_CLASS_CONFIG, VALUE_DESERIALIZER);
+ props.put(GROUP_ID_CONFIG, "KafkaTest");
+
+ LocalMode lma = LocalMode.newInstance();
+ DAG dag = lma.getDAG();
+
+ // Create Test tuple collector
+ CollectorModule collector1 = dag.addOperator("collector", new CollectorModule());
+
+ // Create KafkaSinglePortStringInputOperator
+ AbstractKafkaInputOperator node = createKafkaInputOperator(dag, collector1.inputPort);
+ node.setConsumerProps(props);
+ node.setInitialPartitionCount(1);
+ // set topic
+ node.setTopics(testName);
+ node.setInitialOffset(AbstractKafkaInputOperator.InitialOffset.EARLIEST.name());
+ node.setClusters(getClusterConfig());
+ node.setStrategy("one_to_one");
+
+ // Create local cluster
+ final LocalMode.Controller lc = lma.getController();
+ lc.setHeartbeatMonitoringEnabled(false);
+
+ lc.run(30000);
+
+ return tupleCollection;
+ }
+
+ public static class CollectorModule extends BaseOperator
+ {
+ public final transient CollectorInputPort inputPort = new CollectorInputPort(this);
+
+ long currentWindowId;
+ long operatorId;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ operatorId = context.getId();
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ currentWindowId = windowId;
+ }
+
+ @Override
+ public void endWindow()
+ {
+ super.endWindow();
+ }
+ }
+
+ public static class CollectorInputPort extends DefaultInputPort
+ {
+ CollectorModule ownerNode;
+
+ CollectorInputPort(CollectorModule node)
+ {
+ this.ownerNode = node;
+ }
+
+ @Override
+ public void process(byte[] bt)
+ {
+ tupleCollection.add(new KafkaHelper().deserialize("r", bt));
+ }
+ }
+
+ public static class Person
+ {
+ public String name;
+ public Integer age;
+
+ public Person(String name, Integer age)
+ {
+ this.name = name;
+ this.age = age;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ Person person = (Person)o;
+
+ if (name != null ? !name.equals(person.name) : person.name != null) {
+ return false;
+ }
+
+ return age != null ? age.equals(person.age) : person.age == null;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ int result = name != null ? name.hashCode() : 0;
+ result = 31 * result + (age != null ? age.hashCode() : 0);
+ return result;
+ }
+
+ @Override
+ public String toString()
+ {
+ return name + age.toString();
+ }
+ }
+}
diff --git a/kafka/kafka09/src/test/java/org/apache/apex/malhar/kafka/KafkaHelper.java b/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/KafkaHelper.java
similarity index 79%
rename from kafka/kafka09/src/test/java/org/apache/apex/malhar/kafka/KafkaHelper.java
rename to kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/KafkaHelper.java
index abf3a5b849..e8d4158fec 100644
--- a/kafka/kafka09/src/test/java/org/apache/apex/malhar/kafka/KafkaHelper.java
+++ b/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/KafkaHelper.java
@@ -25,11 +25,11 @@
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;
-public class KafkaHelper implements Serializer,
- Deserializer
+public class KafkaHelper implements Serializer,
+ Deserializer
{
@Override
- public KafkaOutputOperatorTest.Person deserialize(String s, byte[] bytes)
+ public AbstractKafkaOutputOperatorTest.Person deserialize(String s, byte[] bytes)
{
ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
int nameLength = byteBuffer.getInt();
@@ -37,11 +37,11 @@ public KafkaOutputOperatorTest.Person deserialize(String s, byte[] bytes)
byteBuffer.get(name, 0, nameLength);
- return new KafkaOutputOperatorTest.Person(new String(name), byteBuffer.getInt());
+ return new AbstractKafkaOutputOperatorTest.Person(new String(name), byteBuffer.getInt());
}
@Override
- public byte[] serialize(String s, KafkaOutputOperatorTest.Person person)
+ public byte[] serialize(String s, AbstractKafkaOutputOperatorTest.Person person)
{
byte[] name = person.name.getBytes();
diff --git a/kafka/kafka010/src/test/java/org/apache/apex/malhar/kafka/KafkaTestPartitioner.java b/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/KafkaTestPartitioner.java
similarity index 100%
rename from kafka/kafka010/src/test/java/org/apache/apex/malhar/kafka/KafkaTestPartitioner.java
rename to kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/KafkaTestPartitioner.java
diff --git a/kafka/kafka010/src/test/java/org/apache/apex/malhar/kafka/KafkaTestProducer.java b/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/KafkaTestProducer.java
similarity index 84%
rename from kafka/kafka010/src/test/java/org/apache/apex/malhar/kafka/KafkaTestProducer.java
rename to kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/KafkaTestProducer.java
index 322f070779..ae4fffd6c4 100644
--- a/kafka/kafka010/src/test/java/org/apache/apex/malhar/kafka/KafkaTestProducer.java
+++ b/kafka/kafka-common/src/test/java/org/apache/apex/malhar/kafka/KafkaTestProducer.java
@@ -68,13 +68,18 @@ public void setMessages(List messages)
}
private Properties createProducerConfig(int cid)
+ {
+ return createProducerConfig(cid, "localhost:" + AbstractEmbeddedKafka.TEST_KAFKA_BROKER_PORT[cid]);
+ }
+
+ private Properties createProducerConfig(int cid, String brokerId)
{
Properties props = new Properties();
props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, KafkaTestPartitioner.class.getName());
- String brokerList = "localhost:" + KafkaOperatorTestBase.TEST_KAFKA_BROKER_PORT[cid];
- brokerList += hasPartition ? (",localhost:" + KafkaOperatorTestBase.TEST_KAFKA_BROKER_PORT[cid]) : "";
+ String brokerList = brokerId;
+ brokerList += hasPartition ? (",localhost:" + AbstractEmbeddedKafka.TEST_KAFKA_BROKER_PORT[cid]) : "";
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
props.setProperty(ProducerConfig.METADATA_MAX_AGE_CONFIG, "20000");
props.setProperty(ProducerConfig.ACKS_CONFIG, getAckType());
@@ -104,6 +109,21 @@ public KafkaTestProducer(String topic, boolean hasPartition, boolean hasMultiClu
}
}
+ public KafkaTestProducer(String topic, boolean hasPartition, boolean hasMultiCluster, AbstractEmbeddedKafka server)
+ {
+ // Use random partitioner. Don't need the key type. Just set it to Integer.
+ // The message is of type String.
+ this.topic = topic;
+ this.hasPartition = hasPartition;
+ this.hasMultiCluster = hasMultiCluster;
+ producer = new KafkaProducer<>(createProducerConfig(0, server.getBroker(0)));
+ if (hasMultiCluster) {
+ producer1 = new KafkaProducer<>(createProducerConfig(1, server.getBroker(1)));
+ } else {
+ producer1 = null;
+ }
+ }
+
public KafkaTestProducer(String topic, boolean hasPartition)
{
this(topic, hasPartition, false);
@@ -127,9 +147,9 @@ private void generateMessages()
}
// produce the end tuple to let the test input operator know it's done produce messages
for (int i = 0; i < (hasPartition ? 2 : 1); ++i) {
- sendTasks.add(producer.send(new ProducerRecord<>(topic, "" + i, KafkaOperatorTestBase.END_TUPLE)));
+ sendTasks.add(producer.send(new ProducerRecord<>(topic, "" + i, AbstractEmbeddedKafka.END_TUPLE)));
if (hasMultiCluster) {
- sendTasks.add(producer1.send(new ProducerRecord<>(topic, "" + i, KafkaOperatorTestBase.END_TUPLE)));
+ sendTasks.add(producer1.send(new ProducerRecord<>(topic, "" + i, AbstractEmbeddedKafka.END_TUPLE)));
}
}
}
diff --git a/kafka/kafka010/pom.xml b/kafka/kafka010/pom.xml
index 02be49615f..f38b95895c 100755
--- a/kafka/kafka010/pom.xml
+++ b/kafka/kafka010/pom.xml
@@ -94,5 +94,12 @@
test
test