removedOffsets = new HashMap<>();
+ for (TopicPartition topicPartition : removedPartitions) {
+ if (pendingOffsets.containsKey(topicPartition)) {
+ removedOffsets.put(topicPartition, pendingOffsets.remove(topicPartition));
+ }
+ }
+ assignment.removeAll(removedPartitions);
+ return removedOffsets;
+ } finally {
+ unpauseAcknowledgements();
+ }
+ }
+
+ private void pauseAcknowledgments() throws InterruptedException {
+ synchronized (pausedLock) {
+ acknowledgementsPaused = true;
+ while (acknowledgeEntryCount.get() != 0) {
+ pausedLock.wait();
+ }
+ }
+ }
+
+ private void unpauseAcknowledgements() {
+ synchronized (unpausedLock){
+ acknowledgementsPaused = false;
+ unpausedLock.notifyAll();
+ }
+ }
+
+}
diff --git a/connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java b/connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java
new file mode 100644
index 0000000..f493979
--- /dev/null
+++ b/connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java
@@ -0,0 +1,167 @@
+package forklift.connectors;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import forklift.consumer.ForkliftConsumerI;
+import forklift.consumer.KafkaTopicConsumer;
+import forklift.message.KafkaMessage;
+import forklift.producers.ForkliftProducerI;
+import forklift.producers.KafkaForkliftProducer;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import javax.jms.Connection;
+import javax.jms.Message;
+
+/**
+ * Manages a {@link org.apache.kafka.clients.consumer.KafkaConsumer}. A subscription is made whenever a call to getQueue or get Topic
+ * is received. Subscriptions are removed if no consumption is seen for
+ */
+public class KafkaConnector implements ForkliftConnectorI {
+ private static final Logger log = LoggerFactory.getLogger(KafkaConnector.class);
+ private final String kafkaHosts;
+ private final String schemaRegistries;
+ private final String groupId;
+ private KafkaConsumer kafkaConsumer;
+ private KafkaProducer kafkaProducer;
+ private MessageStream messageStream = new MessageStream();
+
+ static ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule())
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ KafkaController controller;
+
+ public KafkaConnector(String kafkaHosts, String schemaRegistries, String groupId) {
+ this.kafkaHosts = kafkaHosts;
+ this.schemaRegistries = schemaRegistries;
+ this.groupId = groupId;
+ }
+
+ @Override
+ public void start() throws ConnectorException {
+
+ Properties producerProperties = new Properties();
+ producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaHosts);
+ producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
+ io.confluent.kafka.serializers.KafkaAvroSerializer.class);
+ producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
+ io.confluent.kafka.serializers.KafkaAvroSerializer.class);
+ //schema.registry.url is a comma separated list of urls
+ producerProperties.put("schema.registry.url", schemaRegistries);
+ this.kafkaProducer = new KafkaProducer(producerProperties);
+ this.controller = createController();
+ this.controller.start();
+ }
+
+ private KafkaController createController() {
+ Properties props = new Properties();
+ props.put("bootstrap.servers", kafkaHosts);
+ props.put("group.id", groupId);
+ props.put("enable.auto.commit", false);
+ props.put("consumer.timeout.ms", "-1");
+ props.put("key.deserializer", io.confluent.kafka.serializers.KafkaAvroDeserializer.class);
+ props.put("value.deserializer", io.confluent.kafka.serializers.KafkaAvroDeserializer.class);
+ props.put("schema.registry.url", schemaRegistries);
+ props.put("specific.avro.reader", false);
+ //props.put("auto.offset.reset", "earliest");
+ this.kafkaConsumer = new KafkaConsumer(props);
+ this.controller = new KafkaController(kafkaConsumer, messageStream);
+ return controller;
+ }
+
+ @Override
+ public void stop() throws ConnectorException {
+ try {
+ this.controller.stop(2000, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ log.error("KafkaConnector interrupted while stopping");
+ }
+ this.kafkaProducer.close();
+ }
+
+ @Override
+ public Connection getConnection() throws ConnectorException {
+ throw new UnsupportedOperationException("Kafka does not support Connections, please work with Queues and Topics");
+ }
+
+ @Override
+ public ForkliftConsumerI getQueue(String name) throws ConnectorException {
+ return this.getTopic(name);
+ }
+
+ @Override
+ public ForkliftConsumerI getTopic(String name) throws ConnectorException {
+ synchronized (this) {
+ if (this.controller == null || !this.controller.isRunning()) {
+ this.controller = createController();
+ this.controller.start();
+ }
+ }
+ return new KafkaTopicConsumer(name, controller, messageStream);
+ }
+
+ @Override
+ public ForkliftProducerI getQueueProducer(String name) {
+ return this.getTopicProducer(name);
+ }
+
+ @Override
+ public ForkliftProducerI getTopicProducer(String name) {
+ return new KafkaForkliftProducer(name, this.kafkaProducer);
+ }
+
+ @Override
+ public ForkliftMessage jmsToForklift(Message m) {
+ try {
+ final ForkliftMessage msg = new ForkliftMessage(m);
+ if (m instanceof KafkaMessage) {
+ KafkaMessage kafkaMessage = (KafkaMessage)m;
+ msg.setProperties(kafkaMessage.getProperties());
+ ConsumerRecord record = kafkaMessage.getConsumerRecord();
+ if (record.value() instanceof GenericRecord) {
+ GenericRecord genericRecord = (GenericRecord)record.value();
+ Object value = genericRecord.get("forkliftMapMsg");
+ value = value == null ? genericRecord.get("forkliftMsg") : value;
+ value = value == null ? genericRecord.get("forkliftJsonMsg") : value;
+ if (value == null) {
+ String jsonValue = genericRecord.toString();
+ value = jsonValue != null && jsonValue.startsWith("{") ? jsonValue : value;
+ }
+ if (value == null) {
+ msg.setFlagged(true);
+ msg.setWarning("Unable to parse message for topic: " + record.topic() + " with value: " + record.value());
+ } else {
+ msg.setMsg(value.toString());
+ }
+ } else {
+ ObjectMapper mapper = new ObjectMapper();
+ try {
+ //inefficient as we map from object to json, then back to the object. This approach works
+ //without any changes to the forklift core libraries however.
+ msg.setMsg(mapper.writeValueAsString(record.value()));
+ } catch (JsonProcessingException e) {
+ msg.setFlagged(true);
+ msg.setWarning("Unable to parse object to json for topic: " +
+ record.topic() +
+ " with value: " +
+ record.value());
+ }
+ }
+ } else {
+ msg.setFlagged(true);
+ msg.setWarning("Unexpected message type: " + m.getClass().getName());
+ }
+
+ return msg;
+ } catch (Exception e) {
+ return null;
+ }
+ }
+}
diff --git a/connectors/kafka/src/main/java/forklift/connectors/KafkaController.java b/connectors/kafka/src/main/java/forklift/connectors/KafkaController.java
new file mode 100644
index 0000000..9a80998
--- /dev/null
+++ b/connectors/kafka/src/main/java/forklift/connectors/KafkaController.java
@@ -0,0 +1,225 @@
+package forklift.connectors;
+
+import forklift.message.KafkaMessage;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+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.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.WakeupException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import javax.jms.JMSException;
+
+/**
+ * Manages the {@link org.apache.kafka.clients.consumer.KafkaConsumer} thread. Polled records are sent to the MessageStream. Commits
+ * are batched and performed on any {@link #acknowledge(org.apache.kafka.clients.consumer.ConsumerRecord) acknowledged records}. Best
+ * performance is usually achieved using one controller instance for all topic subscriptions.
+ *
+ * WARNING: Kafka does not lend itself well to message level commits. For this reason, the controller sends commits
+ * as a batch once every poll cycle. It should be noted that it is possible for a message to be
+ * processed twice should an error occur after the acknowledgement and processing of a message but before the commit takes place.
+ */
+public class KafkaController {
+
+ private static final Logger log = LoggerFactory.getLogger(KafkaController.class);
+ private volatile boolean running = false;
+ private final Set topics = ConcurrentHashMap.newKeySet();
+ private final ExecutorService executor = Executors.newSingleThreadExecutor();
+ private final KafkaConsumer kafkaConsumer;
+ private final MessageStream messageStream;
+ private volatile boolean topicsChanged = false;
+ private Object topicsMonitor = new Object();
+ private AcknowledgedRecordHandler acknowlegmentHandler = new AcknowledgedRecordHandler();
+
+ public KafkaController(KafkaConsumer kafkaConsumer, MessageStream messageStream) {
+ this.kafkaConsumer = kafkaConsumer;
+ this.messageStream = messageStream;
+ }
+
+ /**
+ * Adds a topic which the underlying {@link org.apache.kafka.clients.consumer.KafkaConsumer} will be subscribed to. Adds
+ * the topic to the messageStream.
+ *
+ * @param topic the topic to subscribe to
+ * @return true if the topic was added, false if already added
+ */
+ public boolean addTopic(String topic) {
+ if (!topics.contains(topic)) {
+ messageStream.addTopic(topic);
+ topics.add(topic);
+ topicsChanged = true;
+ synchronized (topicsMonitor) {
+ topicsMonitor.notify();
+ }
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Unsubscribes the underlying {@link org.apache.kafka.clients.consumer.KafkaConsumer} from the passed in topic and removes the
+ * topic from the message stream.
+ *
+ * @param topic the topic to remove
+ * @return true if the topic was removed, false if it wasn't present
+ */
+ public boolean removeTopic(String topic) {
+ boolean removed = topics.remove(topic);
+ if (removed) {
+ messageStream.removeTopic(topic);
+ topicsChanged = true;
+ }
+ return removed;
+ }
+
+ /**
+ * Whether or not this service is running.
+ *
+ * @return true if this service is running, else false
+ */
+ public boolean isRunning() {
+ return running;
+ }
+
+ /**
+ * Acknowledge that processing is beginning on a record. True is returned if this controller is still managing the
+ * partition the record originated from. If the partition is no longer owned by this controller, false is returned
+ * and the message should not be processed.
+ *
+ * @param record the record to acknowledge
+ * @return true if the record should be processed, else false
+ * @throws InterruptedException
+ * @throws JMSException
+ */
+ public boolean acknowledge(ConsumerRecord record) throws InterruptedException, JMSException {
+ log.debug("Acknowledge message with topic {} partition {} offset {}", record.topic(), record.partition(), record.offset());
+ return running && this.acknowlegmentHandler.acknowledgeRecord(record);
+ }
+
+ /**
+ * Starts the controller. Must be called before any other methods.
+ *
+ * WARNING: Avoid starting a service which has already run and been stopped. No attempt is made
+ * to recover a stopped controller.
+ */
+ public void start() {
+ running = true;
+ executor.submit(() -> controlLoop());
+ }
+
+ /**
+ * Stops the controller.
+ *
+ * @param timeout the time to wait for the service to stop
+ * @param timeUnit the TimeUnit of timeout
+ * @throws InterruptedException
+ */
+ public void stop(long timeout, TimeUnit timeUnit) throws InterruptedException {
+ running = false;
+ kafkaConsumer.wakeup();
+ executor.shutdownNow();
+ executor.awaitTermination(timeout, timeUnit);
+ }
+
+ private void controlLoop() {
+ try {
+ while (running) {
+ boolean updatedAssignment = false;
+ if (topics.size() == 0) {
+ //check if the last remaining topic was removed
+ if (kafkaConsumer.assignment().size() > 0) {
+ kafkaConsumer.unsubscribe();
+ }
+ synchronized (topicsMonitor) {
+ //recheck wait condition inside synchronized block
+ if (topics.size() == 0) {
+ //pause the polling thread until a topic comes in
+ topicsMonitor.wait();
+ }
+ }
+ }
+ if (topicsChanged) {
+ topicsChanged = false;
+ kafkaConsumer.subscribe(topics, new RebalanceListener());
+ updatedAssignment = true;
+ }
+ ConsumerRecords records = kafkaConsumer.poll(100);
+ if (updatedAssignment) {
+ this.acknowlegmentHandler.addPartitions(kafkaConsumer.assignment());
+ }
+ if (records.count() > 0) {
+ log.debug("Adding: " + records.count() + " to record stream");
+ messageStream.addRecords(consumerRecordsToKafkaMessages(records));
+ }
+ Map offsetData = this.acknowlegmentHandler.flushAcknowledged();
+ if (offsetData.size() > 0) {
+ String offsetDescription =
+ offsetData.entrySet()
+ .stream()
+ .map(entry -> "topic: " + entry.getKey().topic() + ", " +
+ "partition: " + entry.getKey().partition() + ", " +
+ "offset: " + entry.getValue().offset())
+ .collect(Collectors.joining("|"));
+ log.debug("Commiting offsets {}", offsetDescription);
+ kafkaConsumer.commitSync(offsetData);
+ }
+ }
+ } catch (WakeupException | InterruptedException e) {
+ log.info("Control Loop exiting");
+ } catch (Throwable t) {
+ log.error("Control Loop error, exiting", t);
+ throw t;
+ } finally {
+ running = false;
+ //the kafkaConsumer must be closed in the poll thread
+ log.info("KafkaConsumer closing");
+ kafkaConsumer.close();
+ }
+ }
+
+ private Map> consumerRecordsToKafkaMessages(ConsumerRecords records) {
+ Map> messages = new HashMap<>();
+ records.partitions().forEach(tp -> messages.put(tp, new ArrayList<>()));
+ for (ConsumerRecord record : records) {
+ TopicPartition partition = new TopicPartition(record.topic(), record.partition());
+ messages.get(partition).add(new KafkaMessage(this, record));
+ }
+ return messages;
+ }
+
+ private class RebalanceListener implements ConsumerRebalanceListener {
+ @Override public void onPartitionsRevoked(Collection partitions) {
+ log.debug("Partitions revoked");
+ try {
+ Map
+ removedOffsetData =
+ acknowlegmentHandler.removePartitions(partitions);
+
+ kafkaConsumer.commitSync(removedOffsetData);
+ } catch (InterruptedException e) {
+ log.info("Partition rebalance interrupted", e);
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ @Override
+ public void onPartitionsAssigned(Collection partitions) {
+ log.debug("partitions assigned");
+ acknowlegmentHandler.addPartitions(partitions);
+ }
+
+ }
+}
diff --git a/connectors/kafka/src/main/java/forklift/connectors/MessageStream.java b/connectors/kafka/src/main/java/forklift/connectors/MessageStream.java
new file mode 100644
index 0000000..18279d4
--- /dev/null
+++ b/connectors/kafka/src/main/java/forklift/connectors/MessageStream.java
@@ -0,0 +1,76 @@
+package forklift.connectors;
+
+import forklift.message.KafkaMessage;
+import org.apache.kafka.common.TopicPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A stream of available messages which can be retrieved by topic.
+ */
+public class MessageStream {
+ private static final Logger log = LoggerFactory.getLogger(MessageStream.class);
+ Map> topicQueue = new ConcurrentHashMap<>();
+
+ /**
+ * Adds the passed in records to the stream. After being added, a record is available to be retreived through a
+ * call to {@link #nextRecord(String, long)}.
+ *
+ * Note: All passed in records must belong to a topic added through a call to {@link #addTopic(String)}.
+ *
+ * @throws IllegalStateException if the capacity of a stream has been exceeded.
+ * @param records the records to add
+ */
+ public void addRecords(Map> records) {
+ log.debug("Adding records to stream");
+ for (TopicPartition partition : records.keySet()) {
+ for (KafkaMessage message : records.get(partition)) {
+ topicQueue.get(partition.topic()).add(message);
+ }
+ }
+ }
+
+ /**
+ * Configures this stream to receive messages for a given topic.
+ *
+ * @param topic the topic to add
+ */
+ public void addTopic(String topic) {
+ //The capacity of the blocking queue is never expected to be hit as the Consumer should close the topic if it crashes.
+ //This is mainly in place as a safeguard against a memory leak and blocking the consumption of messages
+ this.topicQueue.put(topic, new LinkedBlockingQueue<>(100000));
+ }
+
+ /**
+ * Removes a topic from the stream. Any queued up messages belonging to the removed topic are discarded.
+ *
+ * @param topic the topic to remove
+ */
+ public void removeTopic(String topic) {
+ this.topicQueue.remove(topic);
+ }
+
+ /**
+ * Blocking method to retrieve the next available record for a topic. Messages are retreived
+ * in FIFO order within their topic.
+ *
+ * @param topic the topic of the {@link javax.jms.Message}
+ * @param timeout how long to wait for a message in milliseconds
+ * @return a message if one is available, else null
+ * @throws InterruptedException
+ */
+ public KafkaMessage nextRecord(String topic, long timeout) throws InterruptedException {
+ KafkaMessage message = null;
+ BlockingQueue queue = topicQueue.get(topic);
+ if (queue != null) {
+ message = queue.poll(timeout, TimeUnit.MILLISECONDS);
+ }
+ return message;
+ }
+}
diff --git a/connectors/kafka/src/main/java/forklift/consumer/KafkaTopicConsumer.java b/connectors/kafka/src/main/java/forklift/consumer/KafkaTopicConsumer.java
new file mode 100644
index 0000000..2a32a5e
--- /dev/null
+++ b/connectors/kafka/src/main/java/forklift/consumer/KafkaTopicConsumer.java
@@ -0,0 +1,70 @@
+package forklift.consumer;
+
+import forklift.connectors.KafkaController;
+import forklift.connectors.MessageStream;
+import javax.jms.JMSException;
+import javax.jms.Message;
+
+/**
+ * Retrieves messages from a kafka topic and adds consumer's topic to the {@link forklift.connectors.KafkaController}.
+ */
+public class KafkaTopicConsumer implements ForkliftConsumerI {
+ private final String topic;
+ private final KafkaController controller;
+ private final MessageStream messageStream;
+ private volatile boolean topicAdded = false;
+
+ public KafkaTopicConsumer(String topic, KafkaController controller, MessageStream messageStream) {
+ this.topic = topic;
+ this.controller = controller;
+ this.messageStream = messageStream;
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * Retrieves the {@link forklift.connectors.MessageStream#nextRecord(String, long) nextRecord} from the messageStream.
+ * If no record is available within the specified timeout, null is returned.
+ *
+ * Note: Because we do not wish to poll for kafka topics until we are actively receiving messages, this method is
+ * also responsible for calling {@link forklift.connectors.KafkaController#addTopic(String) addTopic} on the kafkaController.
+ *
+ * @param timeout the time in milliseconds to wait for a record to become available
+ * @return a message if one is available, else null
+ * @throws JMSException if the process is interrupted or the controller is no longer running
+ */
+ @Override
+ public Message receive(long timeout) throws JMSException {
+ try {
+ if(!topicAdded) {
+ synchronized(this) {
+ if(!topicAdded) {
+ controller.addTopic(topic);
+ topicAdded = true;
+ }
+ }
+
+ }
+ //ensure that our controller is still running
+ if(!controller.isRunning()){
+ throw new JMSException("Connection to Kafka Controller lost");
+ }
+ return messageStream.nextRecord(this.topic, timeout);
+ } catch (InterruptedException e) {
+ throw new JMSException("Kafka message receive interrupted");
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * Removes this consumer's topic from the controller. Future calls to {@link #receive(long) receive} will re-add the
+ * topic to the controller.
+ *
+ * @throws JMSException
+ */
+ @Override
+ public void close() throws JMSException {
+ controller.removeTopic(topic);
+ }
+}
diff --git a/connectors/kafka/src/main/java/forklift/message/KafkaMessage.java b/connectors/kafka/src/main/java/forklift/message/KafkaMessage.java
new file mode 100644
index 0000000..38a8aef
--- /dev/null
+++ b/connectors/kafka/src/main/java/forklift/message/KafkaMessage.java
@@ -0,0 +1,241 @@
+package forklift.message;
+
+import forklift.connectors.KafkaController;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Vector;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+
+/**
+ * A Kafka {@link org.apache.kafka.clients.consumer.ConsumerRecord } wrapped in a JMS {@link javax.jms.Message} functionality.
+ * Please note that many JMS specific properties, such as JMSMessageID and JMSTimestamp, are simply ignored. Properties are
+ * held to allow the storage of metadata, but are not transmitted across the KafkaBroker.
+ */
+public class KafkaMessage implements Message {
+ private final KafkaController controller;
+ private final ConsumerRecord consumerRecord;
+ Map properties = new HashMap<>();
+
+ public KafkaMessage(KafkaController controller, ConsumerRecord consumerRecord) {
+ this.controller = controller;
+ this.consumerRecord = consumerRecord;
+ }
+
+ public ConsumerRecord getConsumerRecord() {
+ return this.consumerRecord;
+ }
+
+ @Override public String getJMSMessageID() throws JMSException {
+ return null;
+ }
+
+ @Override public void setJMSMessageID(String id) throws JMSException {
+
+ }
+
+ @Override public long getJMSTimestamp() throws JMSException {
+ return 0;
+ }
+
+ @Override public void setJMSTimestamp(long timestamp) throws JMSException {
+
+ }
+
+ @Override public byte[] getJMSCorrelationIDAsBytes() throws JMSException {
+ return new byte[0];
+ }
+
+ @Override public void setJMSCorrelationIDAsBytes(byte[] correlationID) throws JMSException {
+
+ }
+
+ @Override public void setJMSCorrelationID(String correlationID) throws JMSException {
+
+ }
+
+ @Override public String getJMSCorrelationID() throws JMSException {
+ return null;
+ }
+
+ @Override public Destination getJMSReplyTo() throws JMSException {
+ return null;
+ }
+
+ @Override public void setJMSReplyTo(Destination replyTo) throws JMSException {
+
+ }
+
+ @Override public Destination getJMSDestination() throws JMSException {
+ return null;
+ }
+
+ @Override public void setJMSDestination(Destination destination) throws JMSException {
+
+ }
+
+ @Override public int getJMSDeliveryMode() throws JMSException {
+ return 0;
+ }
+
+ @Override public void setJMSDeliveryMode(int deliveryMode) throws JMSException {
+
+ }
+
+ @Override public boolean getJMSRedelivered() throws JMSException {
+ return false;
+ }
+
+ @Override public void setJMSRedelivered(boolean redelivered) throws JMSException {
+
+ }
+
+ @Override public String getJMSType() throws JMSException {
+ return null;
+ }
+
+ @Override public void setJMSType(String type) throws JMSException {
+
+ }
+
+ @Override public long getJMSExpiration() throws JMSException {
+ return 0;
+ }
+
+ @Override public void setJMSExpiration(long expiration) throws JMSException {
+
+ }
+
+ @Override public int getJMSPriority() throws JMSException {
+ return 0;
+ }
+
+ @Override public void setJMSPriority(int priority) throws JMSException {
+
+ }
+
+ @Override public void clearProperties() throws JMSException {
+ properties.clear();
+ }
+
+ @Override public boolean propertyExists(String name) throws JMSException {
+ return properties.containsKey(name);
+ }
+
+ @Override public boolean getBooleanProperty(String name) throws JMSException {
+ return (boolean)properties.get(name);
+ }
+
+ @Override public byte getByteProperty(String name) throws JMSException {
+ return (byte)properties.get(name);
+ }
+
+ @Override public short getShortProperty(String name) throws JMSException {
+ return (short)properties.get(name);
+ }
+
+ @Override public int getIntProperty(String name) throws JMSException {
+ return (int)properties.get(name);
+ }
+
+ @Override public long getLongProperty(String name) throws JMSException {
+ return (long)properties.get(name);
+ }
+
+ @Override public float getFloatProperty(String name) throws JMSException {
+ return (float)properties.get(name);
+ }
+
+ @Override public double getDoubleProperty(String name) throws JMSException {
+ return (double)properties.get(name);
+ }
+
+ @Override public String getStringProperty(String name) throws JMSException {
+ return (String)properties.get(name);
+ }
+
+ @Override public Object getObjectProperty(String name) throws JMSException {
+ return properties.get(name);
+ }
+
+ @Override public Enumeration getPropertyNames() throws JMSException {
+ return new Vector(properties.keySet()).elements();
+ }
+
+ @Override public void setBooleanProperty(String name, boolean value) throws JMSException {
+ properties.put(name, value);
+ }
+
+ @Override public void setByteProperty(String name, byte value) throws JMSException {
+ properties.put(name, value);
+ }
+
+ @Override public void setShortProperty(String name, short value) throws JMSException {
+ properties.put(name, value);
+ }
+
+ @Override public void setIntProperty(String name, int value) throws JMSException {
+ properties.put(name, value);
+
+ }
+
+ @Override public void setLongProperty(String name, long value) throws JMSException {
+ properties.put(name, value);
+
+ }
+
+ @Override public void setFloatProperty(String name, float value) throws JMSException {
+ properties.put(name, value);
+
+ }
+
+ @Override public void setDoubleProperty(String name, double value) throws JMSException {
+ properties.put(name, value);
+
+ }
+
+ @Override public void setStringProperty(String name, String value) throws JMSException {
+ properties.put(name, value);
+
+ }
+
+ @Override public void setObjectProperty(String name, Object value) throws JMSException {
+ properties.put(name, value);
+
+ }
+
+ public Map getProperties() {
+ return properties;
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * JMS Messages can typically be acknowledged either synchronously or asynchronously at a message level before processing begins.
+ * This kafka implementation does not allow for that. In order to adapt to the JMS specification, we send the acknowledgment to the
+ * {@link forklift.connectors.KafkaController#acknowledge(org.apache.kafka.clients.consumer.ConsumerRecord)} method, with the
+ * expectation that false will be returned should we no longer hold onto the partition the message came from. In that case, a
+ * {@link javax.jms.JMSException} with an {@link javax.jms.JMSException#getErrorCode() errorCode} of "KAFKA-REBALANCE" will be thrown.
+ * As Kafka can be expected to rebalance somewhat frequently as consumers are added and removed, it is up to the consumer of
+ * this message to appropriate handle the error and prevent processing of this message.
+ *
+ * @throws JMSException
+ */
+ @Override
+ public void acknowledge() throws JMSException {
+ try {
+ if (!controller.acknowledge(consumerRecord)) {
+ throw new JMSException("Unable to acknowledge message, possibly due to kafka partition rebalance", "KAFKA-REBALANCE");
+ }
+ } catch (InterruptedException e) {
+ throw new JMSException("Error acknowledging message");
+ }
+ }
+
+ @Override public void clearBody() throws JMSException {
+
+ }
+}
diff --git a/connectors/kafka/src/main/java/forklift/producers/KafkaForkliftProducer.java b/connectors/kafka/src/main/java/forklift/producers/KafkaForkliftProducer.java
new file mode 100644
index 0000000..b26d2a2
--- /dev/null
+++ b/connectors/kafka/src/main/java/forklift/producers/KafkaForkliftProducer.java
@@ -0,0 +1,141 @@
+package forklift.producers;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import forklift.connectors.ForkliftMessage;
+import forklift.message.Header;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.specific.SpecificRecord;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.errors.SerializationException;
+import java.io.IOException;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Created by afrieze on 2/27/17.
+ */
+public class KafkaForkliftProducer implements ForkliftProducerI {
+
+ private final String topic;
+ private final KafkaProducer kafkaProducer;
+ static ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule())
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ String stringSchema = "{\"type\":\"record\"," +
+ "\"name\":\"ForkliftStringMessage\"," +
+ "\"fields\":[{\"name\":\"forkliftMsg\",\"type\":\"string\"}]}";
+
+ String mapSchema = "{\"type\":\"record\"," +
+ "\"name\":\"ForkliftMapMessage\"," +
+ "\"fields\":[{\"name\":\"forkliftMapMsg\",\"type\":\"string\"}]}";
+
+ String jsonSchema = "{\"type\":\"record\"," +
+ "\"name\":\"ForkliftJsonMessage\"," +
+ "\"fields\":[{\"name\":\"forkliftJsonMsg\",\"type\":\"string\"}]}";
+
+ Schema parsedStringSchema = null;
+ Schema parsedMapSchema = null;
+ Schema parsedJsonSchema = null;
+
+ public KafkaForkliftProducer(String topic, KafkaProducer kafkaProducer) {
+ this.kafkaProducer = kafkaProducer;
+ this.topic = topic;
+ Schema.Parser parser = new Schema.Parser();
+ this.parsedMapSchema = parser.parse(mapSchema);
+ this.parsedStringSchema = parser.parse(stringSchema);
+ this.parsedJsonSchema = parser.parse(jsonSchema);
+ }
+
+ @Override
+ public String send(String message) throws ProducerException {
+ GenericRecord avroRecord = new GenericData.Record(parsedStringSchema);
+ avroRecord.put("forkliftMsg", message);
+ ProducerRecord record = new ProducerRecord<>(topic, null, avroRecord);
+ try {
+ kafkaProducer.send(record);
+ } catch (SerializationException e) {
+ throw new ProducerException("Error creating Kafka Message", e);
+ }
+ return null;
+ }
+
+ @Override
+ public String send(ForkliftMessage message) throws ProducerException {
+ return this.send(message.getMsg());
+ }
+
+ @Override
+ public String send(Object message) throws ProducerException {
+ ProducerRecord record = null;
+ if(message instanceof SpecificRecord){
+ record = new ProducerRecord(topic, null, message);
+ }
+ else{
+ try {
+ GenericRecord avroRecord = new GenericData.Record(parsedJsonSchema);
+ String json = mapper.writeValueAsString(message);
+ avroRecord.put("forkliftJsonMsg", json);
+ record = new ProducerRecord<>(topic, null, avroRecord);
+ }catch(Exception e){
+ throw new ProducerException("Error creating Kafka Message", e);
+ }
+ }
+ try {
+ kafkaProducer.send(record);
+ } catch (SerializationException e) {
+ throw new ProducerException("Error creating Kafka Message", e);
+ }
+ return null;
+ }
+
+ @Override
+ public String send(Map message) throws ProducerException {
+ GenericRecord avroRecord = new GenericData.Record(parsedMapSchema);
+
+ String formattedMap = message.keySet().stream().map(key -> key + "=" + message.get(key)).collect(Collectors.joining("\n"));
+
+ avroRecord.put("forkliftMapMsg", formattedMap);
+ ProducerRecord record = new ProducerRecord<>(topic, null, avroRecord);
+ try {
+ kafkaProducer.send(record);
+ } catch (SerializationException e) {
+ throw new ProducerException("Error creating Kafka Message", e);
+ }
+ return null;
+ }
+
+ @Override
+ public String send(Map headers, Map properties, ForkliftMessage message)
+ throws ProducerException {
+ throw new UnsupportedOperationException("Kafka Producer does not support headers or properties");
+ }
+
+ @Override
+ public Map getHeaders() throws ProducerException {
+ throw new UnsupportedOperationException("Kafka Producer does not support headers");
+ }
+
+ @Override
+ public void setHeaders(Map headers) throws ProducerException {
+ throw new UnsupportedOperationException("Kafka Producer does not support headers");
+ }
+
+ @Override
+ public Map getProperties() throws ProducerException {
+ throw new UnsupportedOperationException("Kafka Producer does not support propertied");
+ }
+
+ @Override
+ public void setProperties(Map properties) throws ProducerException {
+ throw new UnsupportedOperationException("Kafka Producer does not support properties");
+ }
+
+ @Override
+ public void close() throws IOException {
+ //do nothing, the passed in KafkaProducer may be used elsewhere and should be closed by the KafkaController
+ }
+}
diff --git a/connectors/kafka/src/test/java/forklift/connectors/AcknowledgedRecordHandlerTests.java b/connectors/kafka/src/test/java/forklift/connectors/AcknowledgedRecordHandlerTests.java
new file mode 100644
index 0000000..43ab2c5
--- /dev/null
+++ b/connectors/kafka/src/test/java/forklift/connectors/AcknowledgedRecordHandlerTests.java
@@ -0,0 +1,77 @@
+package forklift.connectors;
+
+import static org.junit.Assert.assertEquals;
+import com.sun.scenario.effect.Offset;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.Before;
+import org.junit.Test;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by afrieze on 3/3/17.
+ */
+public class AcknowledgedRecordHandlerTests {
+
+ private AcknowledgedRecordHandler handler;
+
+ @Before
+ public void setup() {
+ this.handler = new AcknowledgedRecordHandler();
+ }
+
+ @Test
+ public void acknowledgeRecordFalseTest() throws InterruptedException{
+ ConsumerRecord record = generateRecord("topic1", 0, "value1", 0);
+ boolean acknowledged = this.handler.acknowledgeRecord(record);
+ assertEquals(false, acknowledged);
+ }
+
+ @Test
+ public void acknowledgeRecordTrueTest() throws InterruptedException{
+ int partition = 0;
+ String topic1 = "topic1";
+ ConsumerRecord record = generateRecord(topic1, partition, "value1", 0);
+ List partitions = Arrays.asList(new TopicPartition(topic1, partition));
+ this.handler.addPartitions(partitions);
+ boolean acknowledged = this.handler.acknowledgeRecord(record);
+ assertEquals(true, acknowledged);
+ }
+
+ @Test
+ public void removePartitionsOffsetTest() throws InterruptedException{
+ int partition = 0;
+ long offset = 123;
+ String topic1 = "topic1";
+ ConsumerRecord record = generateRecord(topic1, partition, "value1", offset);
+ TopicPartition topicPartition = new TopicPartition(topic1, partition);
+ List partitions = Arrays.asList(topicPartition);
+ this.handler.addPartitions(partitions);
+ boolean acknowledged = this.handler.acknowledgeRecord(record);
+ Map offsets = this.handler.removePartitions(partitions);
+ assertEquals(true, offsets.containsKey(topicPartition));
+ assertEquals(offset+1, offsets.get(topicPartition).offset());
+ }
+
+
+ @Test
+ public void removePartitionAcknowledgeTest() throws InterruptedException{
+ int partition = 0;
+ long offset = 123;
+ String topic1 = "topic1";
+ ConsumerRecord record = generateRecord(topic1, partition, "value1", offset);
+ TopicPartition topicPartition = new TopicPartition(topic1, partition);
+ List partitions = Arrays.asList(topicPartition);
+ this.handler.addPartitions(partitions);
+ Map offsets = this.handler.removePartitions(partitions);
+ boolean acknowledged = this.handler.acknowledgeRecord(record);
+ assertEquals(false, acknowledged);
+ }
+
+ private ConsumerRecord generateRecord(String topic, int partition, String value, long offset) {
+ return new ConsumerRecord(topic, partition, offset, null, value);
+ }
+}
diff --git a/connectors/kafka/src/test/java/forklift/connectors/KafkaControllerTests.java b/connectors/kafka/src/test/java/forklift/connectors/KafkaControllerTests.java
new file mode 100644
index 0000000..00e0dc9
--- /dev/null
+++ b/connectors/kafka/src/test/java/forklift/connectors/KafkaControllerTests.java
@@ -0,0 +1,169 @@
+package forklift.connectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+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.common.TopicPartition;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Created by afrieze on 3/3/17.
+ */
+public class KafkaControllerTests {
+
+ @Mock
+ private MessageStream messageStream;
+
+ @Mock
+ private KafkaConsumer kafkaConsumer;
+
+ private KafkaController controller;
+
+ @Captor
+ private ArgumentCaptor> subscribeCaptor;
+
+ @Before
+ public void setup() {
+ MockitoAnnotations.initMocks(this);
+ this.controller = new KafkaController(kafkaConsumer, messageStream);
+ }
+
+ @After
+ public void teardown() throws InterruptedException {
+ this.controller.stop(500, TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ public void addTopicTrueTest() {
+ String topic1 = "topic1";
+ boolean added = this.controller.addTopic(topic1);
+ assertEquals(true, added);
+ }
+
+ @Test
+ public void addTopicFalseTest() {
+ String topic1 = "topic1";
+ boolean added = this.controller.addTopic(topic1);
+ assertEquals(true, added);
+ added = this.controller.addTopic(topic1);
+ assertEquals(false, added);
+ }
+
+ @Test
+ public void removeAddedTopicTest() {
+ String topic1 = "topic1";
+ this.controller.addTopic(topic1);
+ boolean removed = this.controller.removeTopic(topic1);
+ assertEquals(true, removed);
+ }
+
+ @Test
+ public void removeNotAddedTopicTest() {
+ String topic1 = "topic1";
+ boolean removed = this.controller.removeTopic(topic1);
+ assertEquals(false, removed);
+ }
+
+ @Test
+ public void isRunningTest() throws InterruptedException {
+ this.controller.start();
+ assertEquals(true, controller.isRunning());
+ this.controller.stop(10, TimeUnit.MILLISECONDS);
+ assertEquals(false, controller.isRunning());
+ }
+
+ @Test
+ public void firstSubscribeAndPollingTest() throws InterruptedException {
+ String topic1 = "topic1";
+ ConsumerRecords records = mock(ConsumerRecords.class);
+ when(kafkaConsumer.poll(anyLong())).thenReturn(records);
+ this.controller.start();
+ this.controller.addTopic(topic1);
+ verify(kafkaConsumer, timeout(200).times(1)).subscribe(subscribeCaptor.capture(), any());
+ //verify that the control loop is polling repeatedly. Normally there would be a delay but
+ //the kafkaConsumer has been mocked to return immediatly on poll
+ verify(kafkaConsumer, timeout(200).atLeast(5)).poll((anyLong()));
+ assertEquals(1, subscribeCaptor.getValue().size());
+ assertTrue(subscribeCaptor.getValue().contains(topic1));
+ this.controller.stop(10, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Tests that a topic can be added and subscribed, then removed and unsubscribed, then added again and the controller
+ * still functions
+ */
+ @Test
+ public void addRemoveAddTest() throws InterruptedException {
+ String topic1 = "topic1";
+ ConsumerRecords records = mock(ConsumerRecords.class);
+ when(kafkaConsumer.poll(anyLong())).thenReturn(records);
+ this.controller.start();
+ this.controller.addTopic(topic1);
+ verify(kafkaConsumer, timeout(200).times(1)).subscribe(subscribeCaptor.capture(), any());
+ //verify that the control loop is polling repeatedly. Normally there would be a delay but
+ //the kafkaConsumer has been mocked to return immediatly on poll
+ verify(kafkaConsumer, timeout(200).atLeast(5)).poll((anyLong()));
+ assertEquals(1, subscribeCaptor.getValue().size());
+ assertTrue(subscribeCaptor.getValue().contains(topic1));
+ //remove the topic
+ this.controller.removeTopic(topic1);
+ verify(kafkaConsumer, timeout(200).times(1)).subscribe(subscribeCaptor.capture(), any());
+ assertEquals(0, subscribeCaptor.getValue().size());
+ this.controller.stop(10, TimeUnit.MILLISECONDS);
+ //add the topic back
+ this.controller.addTopic(topic1);
+ verify(kafkaConsumer, timeout(200).times(1)).subscribe(subscribeCaptor.capture(), any());
+ verify(kafkaConsumer, timeout(200).atLeast(5)).poll((anyLong()));
+ assertEquals(1, subscribeCaptor.getValue().size());
+ assertTrue(subscribeCaptor.getValue().contains(topic1));
+ }
+
+ /**
+ * Tests that the controller stops running if an error is encountered putting messages on the stream
+ *
+ * @throws InterruptedException
+ */
+ @Test
+ public void shutdownOnMessageStreamError() throws InterruptedException {
+ String topic1 = "topic1";
+ ConsumerRecord record1 = generateRecord(topic1, 1,"value1", 1);
+ Map> recordMap = new HashMap<>();
+ List topicRecords = new ArrayList<>();
+ topicRecords.add(record1);
+ recordMap.put(new TopicPartition(topic1, 1), topicRecords);
+ ConsumerRecords records = new ConsumerRecords(recordMap);
+ when(kafkaConsumer.poll(anyLong())).thenReturn(records);
+ doThrow(new IllegalStateException()).when(messageStream).addRecords(any());
+ this.controller.start();
+ this.controller.addTopic(topic1);
+ verify(kafkaConsumer, timeout(200).times(1)).subscribe(subscribeCaptor.capture(), any());
+ Thread.sleep(50);
+ assertEquals(false, controller.isRunning());
+ }
+
+ private ConsumerRecord generateRecord(String topic, int partition, String value, long offset) {
+ return new ConsumerRecord(topic, partition, offset, null, value);
+ }
+
+}
diff --git a/connectors/kafka/src/test/java/forklift/connectors/MessageStreamTests.java b/connectors/kafka/src/test/java/forklift/connectors/MessageStreamTests.java
new file mode 100644
index 0000000..e866236
--- /dev/null
+++ b/connectors/kafka/src/test/java/forklift/connectors/MessageStreamTests.java
@@ -0,0 +1,108 @@
+package forklift.connectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import forklift.message.KafkaMessage;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.Before;
+import org.junit.Test;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class MessageStreamTests {
+
+ private KafkaController controller;
+ private MessageStream stream;
+
+ @Before
+ public void setup() {
+ stream = new MessageStream();
+ }
+
+ @Test
+ public void addAndGetNextMessageTest() throws InterruptedException {
+ String topic = "topic1";
+ stream.addTopic(topic);
+ ConsumerRecord record = generateRecord(topic, "value1");
+ Map> messages = generateMessages(record);
+ stream.addRecords(messages);
+ KafkaMessage message = stream.nextRecord(topic, 100);
+ assertEquals(record, message.getConsumerRecord());
+ }
+
+ @Test
+ public void nextMessageMultipleTopics() throws InterruptedException{
+
+ String topic1 = "topic1";
+ String topic2 = "topic2";
+ stream.addTopic(topic1);
+ stream.addTopic(topic2);
+ ConsumerRecord record1 = generateRecord(topic1, "value1");
+ ConsumerRecord record2 = generateRecord(topic2, "value1");
+ Map> messages = generateMessages(record1, record2);
+ stream.addRecords(messages);
+ KafkaMessage message1 = stream.nextRecord(topic1, 100);
+ KafkaMessage message2 = stream.nextRecord(topic2, 100);
+ assertEquals(record1, message1.getConsumerRecord());
+ assertEquals(record2, message2.getConsumerRecord());
+ }
+
+ @Test
+ public void nextMessageNullTest() throws InterruptedException {
+ String topic = "topic1";
+ stream.addTopic(topic);
+ KafkaMessage message = stream.nextRecord(topic, 10);
+ assertEquals(null, message);
+ }
+
+ @Test
+ public void nextMessageOrderTest() throws InterruptedException {
+ String topic = "topic1";
+ ConsumerRecord[] records = new ConsumerRecord[100];
+ for (int i = 0; i < 100; i++) {
+ records[i] = generateRecord(topic, "value-" + i);
+ }
+ Map> messages = generateMessages(records);
+ stream.addTopic(topic);
+ stream.addRecords(messages);
+ for (int i = 0; i < 100; i++) {
+ ConsumerRecord record = stream.nextRecord(topic, 10).getConsumerRecord();
+ assertEquals("value-" + i, record.value());
+ }
+ }
+
+ @Test
+ public void removeTopic() throws InterruptedException {
+ String topic = "topic1";
+ stream.addTopic(topic);
+ ConsumerRecord record = generateRecord(topic, "value1");
+ Map> messages = generateMessages(record);
+ stream.addRecords(messages);
+ stream.removeTopic(topic);
+ KafkaMessage message = stream.nextRecord(topic, 100);
+ assertEquals(null, message);
+ }
+
+ private ConsumerRecord generateRecord(String topic, String value) {
+ return new ConsumerRecord(topic, 0, 0, null, value);
+ }
+
+ private Map> generateMessages(ConsumerRecord... records) {
+ Map> messages = new HashMap<>();
+ for (ConsumerRecord record : records) {
+ TopicPartition partition = new TopicPartition(record.topic(), record.partition());
+ if (!messages.containsKey(partition)) {
+ messages.put(partition, new ArrayList<>());
+ }
+ KafkaMessage message = mock(KafkaMessage.class);
+ when(message.getConsumerRecord()).thenReturn(record);
+ messages.get(partition).add(message);
+ }
+ return messages;
+ }
+
+}
diff --git a/connectors/kafka/src/test/java/forklift/connectors/Person.java b/connectors/kafka/src/test/java/forklift/connectors/Person.java
new file mode 100644
index 0000000..21d0592
--- /dev/null
+++ b/connectors/kafka/src/test/java/forklift/connectors/Person.java
@@ -0,0 +1,30 @@
+package forklift.connectors;
+
+/**
+ * Created by afrieze on 3/2/17.
+ */
+public class Person {
+ private String firstName;
+ private String lastName;
+
+ public Person(String firstName, String lastName){
+ this.firstName = firstName;
+ this.lastName = lastName;
+ }
+
+ public String getFirstName() {
+ return firstName;
+ }
+
+ public void setFirstName(String firstName) {
+ this.firstName = firstName;
+ }
+
+ public String getLastName() {
+ return lastName;
+ }
+
+ public void setLastName(String lastName) {
+ this.lastName = lastName;
+ }
+}
diff --git a/connectors/kafka/src/test/java/forklift/consumer/KafkaTopicConsumerTests.java b/connectors/kafka/src/test/java/forklift/consumer/KafkaTopicConsumerTests.java
new file mode 100644
index 0000000..a2cf713
--- /dev/null
+++ b/connectors/kafka/src/test/java/forklift/consumer/KafkaTopicConsumerTests.java
@@ -0,0 +1,66 @@
+package forklift.consumer;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import forklift.connectors.KafkaController;
+import forklift.connectors.MessageStream;
+import forklift.message.KafkaMessage;
+import org.junit.Before;
+import org.junit.Test;
+import javax.jms.JMSException;
+import javax.jms.Message;
+
+public class KafkaTopicConsumerTests {
+
+ private String topic;
+ private KafkaController controller;
+ private MessageStream messageStream;
+ private KafkaTopicConsumer consumer;
+
+ @Before
+ public void setup() {
+ this.topic = "testTopic";
+ this.controller = mock(KafkaController.class);
+ when(controller.isRunning()).thenReturn(true);
+ this.messageStream = mock(MessageStream.class);
+ consumer = new KafkaTopicConsumer(topic, controller, messageStream);
+ }
+
+ @Test
+ public void receiveTimeoutTest() throws InterruptedException, JMSException {
+ long timeout = 100;
+ when(messageStream.nextRecord(this.topic, timeout)).thenReturn(null);
+ Message result = consumer.receive(timeout);
+ assertEquals(null, result);
+ }
+
+ @Test
+ public void receiveMessageTest() throws InterruptedException, JMSException {
+ KafkaMessage message = mock(KafkaMessage.class);
+ long timeout = 100;
+ when(messageStream.nextRecord(this.topic, timeout)).thenReturn(message);
+ Message result = consumer.receive(timeout);
+ assertEquals(message, result);
+ }
+
+ @Test(expected=JMSException.class)
+ public void receiveWithControllerNotRunning() throws JMSException {
+ when(this.controller.isRunning()).thenReturn(false);
+ consumer.receive(100);
+ }
+
+ @Test
+ public void addTopicTest() throws JMSException {
+ consumer.receive(100);
+ verify(controller).addTopic(this.topic);
+ }
+
+ @Test
+ public void closeAndRemoveTopicTest() throws JMSException {
+ consumer.close();
+ verify(controller).removeTopic(this.topic);
+ }
+}
diff --git a/connectors/kafka/src/test/java/forklift/integration/IntegrationTest.java b/connectors/kafka/src/test/java/forklift/integration/IntegrationTest.java
new file mode 100644
index 0000000..b73d556
--- /dev/null
+++ b/connectors/kafka/src/test/java/forklift/integration/IntegrationTest.java
@@ -0,0 +1,76 @@
+package forklift.integration;
+
+import static org.junit.Assert.assertTrue;
+import forklift.connectors.ConnectorException;
+import forklift.connectors.KafkaConnector;
+import forklift.connectors.Person;
+import forklift.producers.ForkliftProducerI;
+import forklift.producers.ProducerException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Integration tests to test actually sending messages to a kafka instance. Please modify the setup method
+ * to match your own kafka and confluent schema registry locations.
+ */
+public class IntegrationTest {
+
+ private KafkaConnector connector;
+
+ @Before
+ public void setup() throws ConnectorException {
+ this.connector = new KafkaConnector("localhost:29092", "http://localhost:28081", "app");
+ connector.start();
+ }
+
+ @Ignore
+ @After
+ public void teardown() throws ConnectorException {
+ connector.stop();
+ }
+
+ @Ignore
+ @Test
+ public void producerStringTests() throws ProducerException, ConnectorException {
+ ForkliftProducerI producer = connector.getTopicProducer("forklift-stringTopic");
+ producer.send("Test Message");
+ assertTrue(true);
+ }
+
+ @Ignore
+ @Test
+ public void producerMapTests() throws ProducerException, ConnectorException {
+ Map values = new HashMap();
+ values.put("value1", "1");
+ values.put("value2", "2");
+ ForkliftProducerI producer = connector.getTopicProducer("forklift-mapTopic");
+ producer.send(values);
+ assertTrue(true);
+ }
+
+ @Ignore
+ @Test
+ public void producerPersonTest() throws ProducerException, ConnectorException {
+ Person person = new Person("John", "Doe");
+ ForkliftProducerI producer = connector.getTopicProducer("forklift-pojoTopic");
+ producer.send(person);
+ assertTrue(true);
+ }
+
+ @Ignore
+ @Test
+ public void multipleRecordsAndTopics() throws ProducerException, ConnectorException {
+ ForkliftProducerI pojoProducer = connector.getTopicProducer("forklift-pojoTopic");
+ ForkliftProducerI stringProducer = connector.getTopicProducer("forklift-stringTopic");
+ for (int i = 0; i < 10000; i++) {
+ Person person = new Person("John" + i, "Doe" + i);
+ //pojoProducer.send(person);
+ stringProducer.send("Test Message" + i);
+ }
+ assertTrue(true);
+ }
+}
diff --git a/connectors/kafka/src/test/java/forklift/message/KafkaMessageTests.java b/connectors/kafka/src/test/java/forklift/message/KafkaMessageTests.java
new file mode 100644
index 0000000..9070d26
--- /dev/null
+++ b/connectors/kafka/src/test/java/forklift/message/KafkaMessageTests.java
@@ -0,0 +1,46 @@
+package forklift.message;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import forklift.connectors.KafkaController;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.Before;
+import org.junit.Test;
+import javax.jms.JMSException;
+
+public class KafkaMessageTests {
+
+ private KafkaController controller;
+ private KafkaMessage message;
+ private ConsumerRecord record;
+
+ @Before
+ public void setup() {
+ controller = mock(KafkaController.class);
+ record = new ConsumerRecord("testTopic", 0, 1L, "key", "value");
+ message = new KafkaMessage(controller, record);
+ }
+
+ @Test
+ public void acknowledgeJMSRebalanceException() throws InterruptedException {
+ boolean exception = false;
+ try {
+ when(controller.acknowledge(record)).thenReturn(false);
+ message.acknowledge();
+ } catch (JMSException e) {
+ exception = true;
+ assertEquals("KAFKA-REBALANCE", e.getErrorCode());
+ }
+ assertTrue(exception);
+ }
+
+ @Test
+ public void acknowledgeCallsControllerSuccess() throws JMSException, InterruptedException {
+ when(controller.acknowledge(record)).thenReturn(true);
+ message.acknowledge();
+ verify(controller).acknowledge(record);
+ }
+}
diff --git a/core/build.sbt b/core/build.sbt
index 3413a93..72b25bb 100644
--- a/core/build.sbt
+++ b/core/build.sbt
@@ -2,7 +2,7 @@ organization := "com.github.dcshock"
name := "forklift"
-version := "0.22"
+version := "0.23"
// target and Xlint cause sbt dist to fail
javacOptions ++= Seq("-source", "1.8")//, "-target", "1.8", "-Xlint")
diff --git a/core/src/main/java/forklift/consumer/ConsumerDeploymentEvents.java b/core/src/main/java/forklift/consumer/ConsumerDeploymentEvents.java
index f998b4b..63799c1 100644
--- a/core/src/main/java/forklift/consumer/ConsumerDeploymentEvents.java
+++ b/core/src/main/java/forklift/consumer/ConsumerDeploymentEvents.java
@@ -7,18 +7,17 @@
import forklift.decorators.Queue;
import forklift.decorators.Topic;
import forklift.deployment.Deployment;
+import forklift.deployment.FileDeployment;
import forklift.deployment.DeploymentEvents;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.annotation.Annotation;
-import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
-import java.util.stream.Collectors;
public class ConsumerDeploymentEvents implements DeploymentEvents {
private static final Logger log = LoggerFactory.getLogger(ConsumerDeploymentEvents.class);
@@ -108,27 +107,17 @@ public synchronized void onUndeploy(Deployment deployment) {
}
// Shutdown each of the services by calling all of their undeployment methods.
- for (ConsumerService s : serviceDeployments.remove(deployment)) {
- try {
- s.onUndeploy();
- } catch (Exception e) {
- log.warn("", e);
- }
+ final List services = serviceDeployments.remove(deployment);
+ if (services != null && !services.isEmpty()) {
+ services.forEach(s -> {
+ try {
+ s.onUndeploy();
+ } catch (Exception e) {
+ log.warn("", e);
+ }
+ });
}
CoreClassLoaders.getInstance().unregister(deployment.getClassLoader());
}
-
- /**
- * We allow jar/zip files.
- * @param deployment
- * @return
- */
- @Override
- public boolean filter(Deployment deployment) {
- log.info("Filtering: " + deployment);
-
- return deployment.getDeployedFile().getName().endsWith(".jar") ||
- deployment.getDeployedFile().getName().endsWith(".zip");
- }
}
diff --git a/core/src/main/java/forklift/consumer/MessageRunnable.java b/core/src/main/java/forklift/consumer/MessageRunnable.java
index 7776103..09acbab 100644
--- a/core/src/main/java/forklift/consumer/MessageRunnable.java
+++ b/core/src/main/java/forklift/consumer/MessageRunnable.java
@@ -61,9 +61,18 @@ public void run() {
try {
msg.getJmsMsg().acknowledge();
} catch (JMSException e) {
- log.error("Error while acking message.", e);
+ //Error code specific to a plugin. Necessary while we are using
+ //the JMS message in order to reduce spamming errors which are actually expected behavior from kafka.
+ if("KAFKA-REBALANCE".equals(e.getErrorCode())){
+ log.warn("TopicPartition no longer available, not processing message.");
+ }
+ else {
+ log.error("Error while acking message.", e);
+ }
close();
return;
+ } catch(Throwable e){
+ log.error("Exception", e);
}
// { Validating }
diff --git a/core/src/main/java/forklift/deployment/ClassDeployment.java b/core/src/main/java/forklift/deployment/ClassDeployment.java
new file mode 100644
index 0000000..388520c
--- /dev/null
+++ b/core/src/main/java/forklift/deployment/ClassDeployment.java
@@ -0,0 +1,103 @@
+package forklift.deployment;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+import forklift.decorators.*;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.stream.Stream;
+
+/**
+ *
+ * A deployment composed of classes. Generally used to specify classes which are part of a deployment at runtime.
+ *
+ * Created by afrieze on 10/28/16.
+ */
+public class ClassDeployment implements Deployment {
+
+ private Set> queues = new HashSet<>();
+ private Set> topics = new HashSet<>();
+ private Set> services = new HashSet<>();
+ private Set> coreServices = new HashSet<>();
+
+ public ClassDeployment(Class ...deploymentClasses){
+ Preconditions.checkNotNull(deploymentClasses);
+ for(Class c : deploymentClasses){
+ if(c.isAnnotationPresent(CoreService.class)){
+ coreServices.add(c);
+ }
+ if(c.isAnnotationPresent(Queue.class)){
+ queues.add(c);
+ }
+ if(c.isAnnotationPresent(Queues.class)){
+ queues.add(c);
+ }
+ if(c.isAnnotationPresent(Service.class)){
+ services.add(c);
+ }
+ if(c.isAnnotationPresent(Topic.class)){
+ topics.add(c);
+ }
+ if(c.isAnnotationPresent(Topics.class)){
+ topics.add(c);
+ }
+ }
+ }
+
+ @Override
+ public Set> getCoreServices() {
+ return coreServices;
+ }
+
+ @Override
+ public Set> getServices() {
+ return services;
+ }
+
+ @Override
+ public Set> getQueues() {
+ return queues;
+ }
+
+ @Override
+ public Set> getTopics() {
+ return topics;
+ }
+
+ @Override
+ public ClassLoader getClassLoader() {
+ return ClassLoader.getSystemClassLoader();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ ClassDeployment that = (ClassDeployment) o;
+
+ if (!queues.equals(that.queues)) return false;
+ if (!topics.equals(that.topics)) return false;
+ if (!services.equals(that.services)) return false;
+ return coreServices.equals(that.coreServices);
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = queues.hashCode();
+ result = 31 * result + topics.hashCode();
+ result = 31 * result + services.hashCode();
+ result = 31 * result + coreServices.hashCode();
+ return result;
+ }
+
+ private boolean identicalSets(Set set1, Set set2){
+ if(set1.size() != set2.size()){
+ return false;
+ }
+ return set1.containsAll(set2);
+ }
+}
diff --git a/core/src/main/java/forklift/deployment/Deployment.java b/core/src/main/java/forklift/deployment/Deployment.java
index f58e51b..97902c4 100644
--- a/core/src/main/java/forklift/deployment/Deployment.java
+++ b/core/src/main/java/forklift/deployment/Deployment.java
@@ -1,176 +1,38 @@
package forklift.deployment;
-import forklift.classloader.ChildFirstClassLoader;
-import forklift.classloader.RunAsClassLoader;
-import forklift.decorators.CoreService;
-import forklift.decorators.Queue;
-import forklift.decorators.Queues;
-import forklift.decorators.Service;
-import forklift.decorators.Topic;
-import forklift.decorators.Topics;
-import org.reflections.Reflections;
-import org.reflections.util.ConfigurationBuilder;
-
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.URI;
-import java.net.URL;
-import java.util.HashSet;
-import java.util.List;
import java.util.Set;
-import java.util.jar.JarEntry;
-import java.util.jar.JarFile;
-import java.util.stream.Collectors;
-
-public class Deployment {
- private Set> queues = new HashSet<>();
- private Set> topics = new HashSet<>();
- private Set> services = new HashSet<>();
- private Set> coreServices = new HashSet<>();
- private ClassLoader cl;
-
- private File deployedFile;
- private Reflections reflections;
-
- public Deployment() {
-
- }
-
- public Deployment(File deployedFile)
- throws IOException {
- if (deployedFile == null)
- throw new IOException("Missing file");
-
- this.deployedFile = deployedFile;
-
- if (!deployedFile.getName().endsWith(".jar") && !deployedFile.getName().endsWith(".zip")) {
- throw new IOException("Unhandled file type");
- }
-
- // Read jars out of the deployed file.
- final JarFile jar = new JarFile(deployedFile);
- final List jarUrls = jar.stream().filter(entry -> {
- return entry.getName().endsWith(".jar") || entry.getName().endsWith(".zip");
- }).map(entry -> {
- try {
- return jarEntryAsUri(jar, entry).toURL();
- } catch (Exception e) {
- return null;
- }
- }).collect(Collectors.toList());
- jarUrls.add(deployedFile.toURI().toURL());
-
- // TODO - we should cleanup temp jars when the deploy is thrown away.
-
- final URL[] urls = jarUrls.toArray(new URL[0]);
-
- // Assign a new classloader to this deployment.
- cl = new ChildFirstClassLoader(urls, getClass().getClassLoader());
-
- // Reflect the deployment to determine if there are any consumers
- // annotated.
- reflections = new Reflections(new ConfigurationBuilder()
- .addClassLoader(cl)
- .setUrls(urls));
-
- RunAsClassLoader.run(cl, () -> {
- coreServices.addAll(reflections.getTypesAnnotatedWith(CoreService.class));
- queues.addAll(reflections.getTypesAnnotatedWith(Queue.class));
- queues.addAll(reflections.getTypesAnnotatedWith(Queues.class));
- services.addAll(reflections.getTypesAnnotatedWith(Service.class));
- topics.addAll(reflections.getTypesAnnotatedWith(Topic.class));
- topics.addAll(reflections.getTypesAnnotatedWith(Topics.class));
- });
-
- if (coreServices.size() > 0 && (queues.size() > 0 || topics.size() > 0 || services.size() > 0))
- throw new IOException("Invalid core service due to queues/topics/services being deployed along side.");
- }
-
- public boolean isJar() {
- return deployedFile.getPath().endsWith(".jar");
- }
-
- public boolean isClass() {
- return deployedFile.getPath().endsWith(".class");
- }
-
- public File getDeployedFile() {
- return deployedFile;
- }
-
- public void setDeployedFile(File deployedFile) {
- this.deployedFile = deployedFile;
- }
-
- public ClassLoader getClassLoader() {
- return cl;
- }
-
- public Set> getCoreServices() {
- return coreServices;
- }
-
- public Set> getServices() {
- return services;
- }
-
- public Set> getQueues() {
- return queues;
- }
-
- public Set> getTopics() {
- return topics;
- }
-
- public Reflections getReflections() {
- return reflections;
- }
-
- @Override
- public boolean equals(Object o) {
- if (((Deployment)o).getDeployedFile().equals(deployedFile))
- return true;
- return false;
- }
-
- @Override
- public String toString() {
- return "Deployment [queues=" + queues + ", topics=" + topics + ", cl="
- + cl + ", deployedFile=" + deployedFile + ", reflections="
- + reflections + "]";
- }
-
- private static URI jarEntryAsUri(JarFile jarFile, JarEntry jarEntry)
- throws IOException {
- if (jarFile == null || jarEntry == null)
- throw new IOException("Invalid jar file or entry");
- InputStream input = null;
- OutputStream output = null;
- try {
- String name = jarEntry.getName().replace('/', '_');
- int i = name.lastIndexOf(".");
- String extension = i > -1 ? name.substring(i) : "";
- File file = File.createTempFile(
- name.substring(0, name.length() - extension.length()) +
- ".", extension);
- file.deleteOnExit();
- input = jarFile.getInputStream(jarEntry);
- output = new FileOutputStream(file);
- int readCount;
- byte[] buffer = new byte[4096];
- while ((readCount = input.read(buffer)) != -1) {
- output.write(buffer, 0, readCount);
- }
- return file.toURI();
- } finally {
- if (input != null)
- input.close();
- if (output != null)
- output.close();
- }
- }
+/**
+ * Defines the methods required for a Forklift Deployment.
+ *
+ * Created by afrieze on 10/28/16.
+ */
+public interface Deployment {
+
+ /**
+ * @return clases in this Deployment annotated with the {@link forklift.decorators.CoreService} annotation
+ */
+ Set> getCoreServices();
+
+ /**
+ * @return clases in this Deployment annotated with the {@link forklift.decorators.Service} annotation
+ */
+ Set> getServices();
+
+ /**
+ * @return clases in this Deployment annotated with the {@link forklift.decorators.Queue} annotation
+ */
+ Set> getQueues();
+
+ /**
+ * @return clases in this Deployment annotated with the {@link forklift.decorators.Topics} annotation
+ */
+ Set> getTopics();
+
+ /**
+ * Returns a {@link ClassLoader} capable of loading the classes encapsulated by this deployment
+ *
+ * @return {@link ClassLoader}
+ */
+ ClassLoader getClassLoader();
}
diff --git a/core/src/main/java/forklift/deployment/DeploymentManager.java b/core/src/main/java/forklift/deployment/DeploymentManager.java
index e4ece32..c1ab231 100644
--- a/core/src/main/java/forklift/deployment/DeploymentManager.java
+++ b/core/src/main/java/forklift/deployment/DeploymentManager.java
@@ -6,22 +6,23 @@
import java.io.IOException;
public class DeploymentManager extends Registrar {
- public synchronized Deployment registerDeployedFile(File f)
+ public synchronized FileDeployment registerDeployedFile(File f)
throws IOException {
- final Deployment d = new Deployment(f);
+ final FileDeployment d = new FileDeployment(f);
register(d);
return d;
}
public synchronized Deployment unregisterDeployedFile(File f) {
- Deployment d = new Deployment();
+ FileDeployment d = new FileDeployment();
d.setDeployedFile(f);
return unregister(d);
}
public synchronized boolean isDeployed(File f) {
- Deployment d = new Deployment();
+ FileDeployment d = new FileDeployment();
d.setDeployedFile(f);
return isRegistered(d);
}
+
}
diff --git a/core/src/main/java/forklift/deployment/FileDeployment.java b/core/src/main/java/forklift/deployment/FileDeployment.java
new file mode 100644
index 0000000..7052786
--- /dev/null
+++ b/core/src/main/java/forklift/deployment/FileDeployment.java
@@ -0,0 +1,181 @@
+package forklift.deployment;
+
+import forklift.classloader.ChildFirstClassLoader;
+import forklift.classloader.RunAsClassLoader;
+import forklift.decorators.CoreService;
+import forklift.decorators.Queue;
+import forklift.decorators.Queues;
+import forklift.decorators.Service;
+import forklift.decorators.Topic;
+import forklift.decorators.Topics;
+import org.reflections.Reflections;
+import org.reflections.util.ConfigurationBuilder;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.URI;
+import java.net.URL;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import java.util.stream.Collectors;
+
+public class FileDeployment implements Deployment{
+ private Set> queues = new HashSet<>();
+ private Set> topics = new HashSet<>();
+ private Set> services = new HashSet<>();
+ private Set> coreServices = new HashSet<>();
+ private ClassLoader cl;
+
+ private File deployedFile;
+ private Reflections reflections;
+
+ public FileDeployment() {
+
+ }
+
+
+ public FileDeployment(File deployedFile)
+ throws IOException {
+ if (deployedFile == null)
+ throw new IOException("Missing file");
+
+ this.deployedFile = deployedFile;
+
+ if (!deployedFile.getName().endsWith(".jar") && !deployedFile.getName().endsWith(".zip")) {
+ throw new IOException("Unhandled file type");
+ }
+
+ // Read jars out of the deployed file.
+ final JarFile jar = new JarFile(deployedFile);
+ final List jarUrls = jar.stream().filter(entry -> {
+ return entry.getName().endsWith(".jar") || entry.getName().endsWith(".zip");
+ }).map(entry -> {
+ try {
+ return jarEntryAsUri(jar, entry).toURL();
+ } catch (Exception e) {
+ return null;
+ }
+ }).collect(Collectors.toList());
+ jarUrls.add(deployedFile.toURI().toURL());
+
+ // TODO - we should cleanup temp jars when the deploy is thrown away.
+
+ final URL[] urls = jarUrls.toArray(new URL[0]);
+
+ // Assign a new classloader to this deployment.
+ cl = new ChildFirstClassLoader(urls, getClass().getClassLoader());
+
+ // Reflect the deployment to determine if there are any consumers
+ // annotated.
+ reflections = new Reflections(new ConfigurationBuilder()
+ .addClassLoader(cl)
+ .setUrls(urls));
+
+ RunAsClassLoader.run(cl, () -> {
+ coreServices.addAll(reflections.getTypesAnnotatedWith(CoreService.class));
+ queues.addAll(reflections.getTypesAnnotatedWith(Queue.class));
+ queues.addAll(reflections.getTypesAnnotatedWith(Queues.class));
+ services.addAll(reflections.getTypesAnnotatedWith(Service.class));
+ topics.addAll(reflections.getTypesAnnotatedWith(Topic.class));
+ topics.addAll(reflections.getTypesAnnotatedWith(Topics.class));
+ });
+
+ if (coreServices.size() > 0 && (queues.size() > 0 || topics.size() > 0 || services.size() > 0))
+ throw new IOException("Invalid core service due to queues/topics/services being deployed along side.");
+ }
+
+ public boolean isJar() {
+ return deployedFile.getPath().endsWith(".jar");
+ }
+
+ public boolean isClass() {
+ return deployedFile.getPath().endsWith(".class");
+ }
+
+ public File getDeployedFile() {
+ return deployedFile;
+ }
+
+ public void setDeployedFile(File deployedFile) {
+ this.deployedFile = deployedFile;
+ }
+
+ public ClassLoader getClassLoader() {
+ return cl;
+ }
+
+ @Override
+ public Set> getCoreServices() {
+ return coreServices;
+ }
+
+ @Override
+ public Set> getServices() {
+ return services;
+ }
+
+ @Override
+ public Set> getQueues() {
+ return queues;
+ }
+
+ @Override
+ public Set> getTopics() {
+ return topics;
+ }
+
+ public Reflections getReflections() {
+ return reflections;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (((FileDeployment)o).getDeployedFile().equals(deployedFile))
+ return true;
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return "FileDeployment [queues=" + queues + ", topics=" + topics + ", cl="
+ + cl + ", deployedFile=" + deployedFile + ", reflections="
+ + reflections + "]";
+ }
+
+ private static URI jarEntryAsUri(JarFile jarFile, JarEntry jarEntry)
+ throws IOException {
+ if (jarFile == null || jarEntry == null)
+ throw new IOException("Invalid jar file or entry");
+
+ InputStream input = null;
+ OutputStream output = null;
+ try {
+ String name = jarEntry.getName().replace('/', '_');
+ int i = name.lastIndexOf(".");
+ String extension = i > -1 ? name.substring(i) : "";
+ File file = File.createTempFile(
+ name.substring(0, name.length() - extension.length()) +
+ ".", extension);
+ file.deleteOnExit();
+ input = jarFile.getInputStream(jarEntry);
+ output = new FileOutputStream(file);
+ int readCount;
+ byte[] buffer = new byte[4096];
+ while ((readCount = input.read(buffer)) != -1) {
+ output.write(buffer, 0, readCount);
+ }
+ return file.toURI();
+ } finally {
+ if (input != null)
+ input.close();
+ if (output != null)
+ output.close();
+ }
+ }
+}
diff --git a/core/src/test/java/forklift/consumer/ConsumerDeploymentEventsTest.java b/core/src/test/java/forklift/consumer/ConsumerDeploymentEventsTest.java
index 3f13f01..196a8cc 100644
--- a/core/src/test/java/forklift/consumer/ConsumerDeploymentEventsTest.java
+++ b/core/src/test/java/forklift/consumer/ConsumerDeploymentEventsTest.java
@@ -1,7 +1,7 @@
package forklift.consumer;
import forklift.ForkliftTest;
-import forklift.deployment.Deployment;
+import forklift.deployment.FileDeployment;
import forklift.deployment.DeploymentWatch;
import com.google.common.io.Files;
@@ -27,10 +27,10 @@ public void consumerDeployEvent()
Files.copy(ForkliftTest.testJar(), deployFile);
watch.run();
- Mockito.verify(events).onDeploy(Mockito.any(Deployment.class));
+ Mockito.verify(events).onDeploy(Mockito.any(FileDeployment.class));
deployFile.delete();
watch.run();
- Mockito.verify(events).onUndeploy(Mockito.any(Deployment.class));
+ Mockito.verify(events).onUndeploy(Mockito.any(FileDeployment.class));
}
}
diff --git a/core/src/test/java/forklift/deployment/ClassDeploymentTest.java b/core/src/test/java/forklift/deployment/ClassDeploymentTest.java
new file mode 100644
index 0000000..697fd00
--- /dev/null
+++ b/core/src/test/java/forklift/deployment/ClassDeploymentTest.java
@@ -0,0 +1,93 @@
+package forklift.deployment;
+
+import forklift.ForkliftTest;
+
+import forklift.deployment.deploymentClasses.TestCoreService1;
+import forklift.deployment.deploymentClasses.TestQueue1;
+import forklift.deployment.deploymentClasses.TestQueue2;
+import forklift.deployment.deploymentClasses.TestService1;
+import forklift.deployment.deploymentClasses.TestTopic1;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.jar.Attributes;
+import java.util.jar.JarOutputStream;
+import java.util.jar.Manifest;
+
+import static org.junit.Assert.*;
+
+public class ClassDeploymentTest {
+
+ @Test(expected = NullPointerException.class)
+ public void testNullDeployment() throws IOException {
+ new ClassDeployment(null);
+ }
+
+ @Test
+ public void testPartRetrieval() throws IOException {
+ File f = ForkliftTest.testMultiTQJar();
+ ClassDeployment d = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ assertNotNull(d);
+ assertEquals(2, d.getQueues().size());
+ assertEquals(1, d.getTopics().size());
+ assertEquals(1, d.getCoreServices().size());
+ assertEquals(1, d.getServices().size());
+ }
+
+ @Test
+ public void hashCodeEqualsTest(){
+ ClassDeployment d1 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ ClassDeployment d2 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ assertEquals(d1.hashCode(), d2.hashCode());
+ }
+
+ @Test
+ public void hashCodeNotEqualsTest(){
+ ClassDeployment d1 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ ClassDeployment d2 = new ClassDeployment(TestCoreService1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ assertNotEquals(d1.hashCode(), d2.hashCode());
+ }
+
+ @Test
+ public void equalsReflexiveTest(){
+ ClassDeployment d1 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ assertTrue(d1.equals(d1));
+ }
+ @Test
+ public void equalsSymmetricTrueTest(){
+ ClassDeployment d1 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ ClassDeployment d2 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ assertTrue(d1.equals(d2));
+ assertTrue(d2.equals(d1));
+ }
+ @Test
+ public void equalsSymmetricFalseTest1(){
+ ClassDeployment d1 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ ClassDeployment d2 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestService1.class, TestTopic1.class);
+ assertFalse(d1.equals(d2));
+ assertFalse(d2.equals(d1));
+ }
+ @Test
+ public void equalsSymmetricFalseTest2(){
+ ClassDeployment d1 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ ClassDeployment d2 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestTopic1.class, TestTopic1.class);
+ assertFalse(d1.equals(d2));
+ assertFalse(d2.equals(d1));
+ }
+ @Test
+ public void equalsTransitiveTest(){
+ ClassDeployment d1 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ ClassDeployment d2 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ ClassDeployment d3 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ assertTrue(d1.equals(d2));
+ assertTrue(d2.equals(d3));
+ assertTrue(d1.equals(d3));
+ }
+ @Test
+ public void equalsNullReferenceTest(){
+ ClassDeployment d1 = new ClassDeployment(TestCoreService1.class, TestQueue1.class, TestQueue2.class, TestService1.class, TestTopic1.class);
+ assertFalse(d1.equals(null));
+ }
+}
diff --git a/core/src/test/java/forklift/deployment/DeploymentManagerTest.java b/core/src/test/java/forklift/deployment/DeploymentManagerTest.java
index 54d1f05..6f0b715 100644
--- a/core/src/test/java/forklift/deployment/DeploymentManagerTest.java
+++ b/core/src/test/java/forklift/deployment/DeploymentManagerTest.java
@@ -18,7 +18,7 @@ public void deployJar()
DeploymentManager deploymentManager = new DeploymentManager();
final File jar = ForkliftTest.testJar();
- Deployment deployment = deploymentManager.registerDeployedFile(jar);
+ FileDeployment deployment = deploymentManager.registerDeployedFile(jar);
assertTrue(deploymentManager.isDeployed(jar));
assertEquals(1, deployment.getQueues().size());
diff --git a/core/src/test/java/forklift/deployment/DeploymentWatchTest.java b/core/src/test/java/forklift/deployment/DeploymentWatchTest.java
index c7f5cfb..ffb7d82 100644
--- a/core/src/test/java/forklift/deployment/DeploymentWatchTest.java
+++ b/core/src/test/java/forklift/deployment/DeploymentWatchTest.java
@@ -24,18 +24,21 @@ public void watch()
final File file = File.createTempFile("test", ".jar", dir);
Files.copy(ForkliftTest.testJar(), file);
+ FileDeployment fileDeployment = new FileDeployment();
+ fileDeployment.setDeployedFile(file);
+
final AtomicBoolean deploy = new AtomicBoolean(true);
DeploymentWatch watch = new DeploymentWatch(dir, new DeploymentEvents() {
@Override
public void onDeploy(Deployment deployment) {
assertTrue(deploy.get());
- assertEquals(file, deployment.getDeployedFile());
+ assertEquals(fileDeployment, deployment);
}
@Override
public void onUndeploy(Deployment deployment) {
assertFalse(deploy.get());
- assertEquals(file, deployment.getDeployedFile());
+ assertEquals(fileDeployment, deployment);
}
});
@@ -59,19 +62,21 @@ public void loadProperties()
final String props = "db=prod\ndeploy=prod\nusername=mysql\npassword=mysql\n";
fos.write(props.getBytes());
fos.close();
+ FileDeployment fileDeployment = new FileDeployment();
+ fileDeployment.setDeployedFile(file);
final AtomicBoolean deploy = new AtomicBoolean(true);
DeploymentWatch watch = new DeploymentWatch(dir, new DeploymentEvents() {
@Override
public void onDeploy(Deployment deployment) {
assertTrue(deploy.get());
- assertEquals(file, deployment.getDeployedFile());
+ assertEquals(fileDeployment, deployment);
}
@Override
public void onUndeploy(Deployment deployment) {
assertFalse(deploy.get());
- assertEquals(file, deployment.getDeployedFile());
+ assertEquals(fileDeployment, deployment);
}
});
@@ -95,19 +100,22 @@ public void propsWithComment()
final String props = "db=prod\n#Db Creds\n\ndb.username=mysql\ndb.password=mysql\n";
fos.write(props.getBytes());
fos.close();
+ FileDeployment fileDeployment = new FileDeployment();
+ fileDeployment.setDeployedFile(file);
+
final AtomicBoolean deploy = new AtomicBoolean(true);
DeploymentWatch watch = new DeploymentWatch(dir, new DeploymentEvents() {
@Override
public void onDeploy(Deployment deployment) {
assertTrue(deploy.get());
- assertEquals(file, deployment.getDeployedFile());
+ assertEquals(fileDeployment, deployment);
}
@Override
public void onUndeploy(Deployment deployment) {
assertFalse(deploy.get());
- assertEquals(file, deployment.getDeployedFile());
+ assertEquals(fileDeployment, deployment);
}
});
@@ -132,19 +140,21 @@ public void badProps()
final String props = "db:prod\n\\u00sx=blah\nusername mysql\npassword=mysql\n";
fos.write(props.getBytes());
fos.close();
+ FileDeployment fileDeployment = new FileDeployment();
+ fileDeployment.setDeployedFile(file);
final AtomicBoolean deploy = new AtomicBoolean(true);
DeploymentWatch watch = new DeploymentWatch(dir, new DeploymentEvents() {
@Override
public void onDeploy(Deployment deployment) {
assertTrue(deploy.get());
- assertEquals(file, deployment.getDeployedFile());
+ assertEquals(fileDeployment, deployment);
}
@Override
public void onUndeploy(Deployment deployment) {
assertFalse(deploy.get());
- assertEquals(file, deployment.getDeployedFile());
+ assertEquals(fileDeployment, deployment);
}
});
diff --git a/core/src/test/java/forklift/deployment/DeploymentTest.java b/core/src/test/java/forklift/deployment/FileDeploymentTest.java
similarity index 91%
rename from core/src/test/java/forklift/deployment/DeploymentTest.java
rename to core/src/test/java/forklift/deployment/FileDeploymentTest.java
index 332b35f..828c1cb 100644
--- a/core/src/test/java/forklift/deployment/DeploymentTest.java
+++ b/core/src/test/java/forklift/deployment/FileDeploymentTest.java
@@ -15,10 +15,10 @@
import java.util.jar.JarOutputStream;
import java.util.jar.Manifest;
-public class DeploymentTest {
+public class FileDeploymentTest {
@Test(expected = IOException.class)
public void testNullDeployment() throws IOException {
- new Deployment(null);
+ new FileDeployment(null);
}
// Any kind of bad Jar file should only throw IOExcpetion otherwise we may bring down the system
@@ -27,7 +27,7 @@ public void testEmptyDeployment() throws IOException {
File f = File.createTempFile("test", ".txt");
try {
- new Deployment(f);
+ new FileDeployment(f);
} finally {
// Don't leave test files around
f.delete();
@@ -44,7 +44,7 @@ public void testManifestOnlyDeploy() throws IOException {
target.close();
try {
- Deployment d = new Deployment(f);
+ FileDeployment d = new FileDeployment(f);
assertEquals(f.getName(), d.getDeployedFile().getName());
assertTrue(d.isJar());
assertEquals(0, d.getQueues().size());
@@ -62,7 +62,7 @@ public void testManifestOnlyDeploy() throws IOException {
@Test
public void testDeployJar() throws IOException {
File f = ForkliftTest.testMultiTQJar();
- Deployment d = new Deployment(f);
+ FileDeployment d = new FileDeployment(f);
assertNotNull(d);
assertTrue(d.isJar());
assertEquals(2, d.getQueues().size());
@@ -77,7 +77,7 @@ public void testDeployJar() throws IOException {
@Test
public void testDeployJarJar() throws IOException {
File f = ForkliftTest.testJarJar();
- Deployment d = new Deployment(f);
+ FileDeployment d = new FileDeployment(f);
assertNotNull(d);
assertTrue(d.isJar());
assertEquals(4, d.getQueues().size());
diff --git a/core/src/test/java/forklift/deployment/deploymentClasses/TestCoreService1.java b/core/src/test/java/forklift/deployment/deploymentClasses/TestCoreService1.java
new file mode 100644
index 0000000..4e4733a
--- /dev/null
+++ b/core/src/test/java/forklift/deployment/deploymentClasses/TestCoreService1.java
@@ -0,0 +1,10 @@
+package forklift.deployment.deploymentClasses;
+
+import forklift.decorators.CoreService;
+
+/**
+ * Created by afrieze on 10/31/16.
+ */
+@CoreService
+public class TestCoreService1 {
+}
diff --git a/core/src/test/java/forklift/deployment/deploymentClasses/TestQueue1.java b/core/src/test/java/forklift/deployment/deploymentClasses/TestQueue1.java
new file mode 100644
index 0000000..c800c7d
--- /dev/null
+++ b/core/src/test/java/forklift/deployment/deploymentClasses/TestQueue1.java
@@ -0,0 +1,10 @@
+package forklift.deployment.deploymentClasses;
+
+import forklift.decorators.Queue;
+
+/**
+ * Created by afrieze on 10/31/16.
+ */
+@Queue("test")
+public class TestQueue1 {
+}
diff --git a/core/src/test/java/forklift/deployment/deploymentClasses/TestQueue2.java b/core/src/test/java/forklift/deployment/deploymentClasses/TestQueue2.java
new file mode 100644
index 0000000..ac5411f
--- /dev/null
+++ b/core/src/test/java/forklift/deployment/deploymentClasses/TestQueue2.java
@@ -0,0 +1,10 @@
+package forklift.deployment.deploymentClasses;
+
+import forklift.decorators.Queue;
+
+/**
+ * Created by afrieze on 10/31/16.
+ */
+@Queue("test")
+public class TestQueue2 {
+}
diff --git a/core/src/test/java/forklift/deployment/deploymentClasses/TestService1.java b/core/src/test/java/forklift/deployment/deploymentClasses/TestService1.java
new file mode 100644
index 0000000..4272468
--- /dev/null
+++ b/core/src/test/java/forklift/deployment/deploymentClasses/TestService1.java
@@ -0,0 +1,10 @@
+package forklift.deployment.deploymentClasses;
+
+import forklift.decorators.Service;
+
+/**
+ * Created by afrieze on 10/31/16.
+ */
+@Service
+public class TestService1 {
+}
diff --git a/core/src/test/java/forklift/deployment/deploymentClasses/TestTopic1.java b/core/src/test/java/forklift/deployment/deploymentClasses/TestTopic1.java
new file mode 100644
index 0000000..1d87089
--- /dev/null
+++ b/core/src/test/java/forklift/deployment/deploymentClasses/TestTopic1.java
@@ -0,0 +1,10 @@
+package forklift.deployment.deploymentClasses;
+
+import forklift.decorators.Topic;
+
+/**
+ * Created by afrieze on 10/31/16.
+ */
+@Topic("testTopic")
+public class TestTopic1 {
+}
diff --git a/project/Build.scala b/project/Build.scala
index e241e10..a2673fd 100644
--- a/project/Build.scala
+++ b/project/Build.scala
@@ -32,9 +32,14 @@ object ForkliftBuild extends Build {
base = file("connectors/activemq")
).dependsOn(core)
+ lazy val kafka = Project(
+ id = "kafka",
+ base = file("connectors/kafka")
+ ).dependsOn(core)
+
lazy val server = Project(
id = "server",
base = file("server")
- ).dependsOn(core, activemq, replay, retry)
+ ).dependsOn(core, activemq, replay, retry, kafka)
}
diff --git a/server/.gitignore b/server/.gitignore
new file mode 100644
index 0000000..ae3c172
--- /dev/null
+++ b/server/.gitignore
@@ -0,0 +1 @@
+/bin/
diff --git a/server/README.md b/server/README.md
new file mode 100644
index 0000000..823bbe4
--- /dev/null
+++ b/server/README.md
@@ -0,0 +1,85 @@
+# README #
+
+## Overview ##
+The server project allows developers to embed a forklift server within their applications.
+
+### Embedding with ActiveMQ ###
+```java
+ public void startApplication() throws InterruptedException {
+ ForkliftOpts options = new ForkliftOpts();
+ String ip = "localhost";
+ //Note the broker url format: consule.{activemq-service-name}
+ options.setBrokerUrl("consul.activemq-broker");
+ options.setConsulHost(ip);
+ server = new ForkliftServer(options);
+ try {
+ server.startServer(10000, TimeUnit.SECONDS);
+ // Forklift closes Unirest...we need to reopen it with a call to Options.refresh()
+ Options.refresh();
+ } catch (InterruptedException e) {
+ }
+ if (server.getServerState() == ServerState.RUNNING) {
+ //Register any Consumers you wish to run in the embedded forklift
+ server.registerDeployment(TestStringConsumer.class, TestMapConsumer.class, TestPersonConsumer.class);
+ } else {
+ try {
+ server.stopServer(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ }
+ System.exit(-1);
+ }
+
+
+ public void stopApplication() {
+ if(server != null){
+ try {
+ server.stopServer(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.error("Forklift shutdown interrupted", e);
+ }
+ }
+ }
+ }
+```
+
+### Embedding with Kafka and Confluent's Schema-Registry ###
+```java
+
+ public void startApplication() throws InterruptedException {
+ ForkliftOpts options = new ForkliftOpts();
+ String ip = "localhost";
+ //Note the broker url format: consule.{kafka-service-name}.{schema-registry-service-name}
+ options.setBrokerUrl("consul.kafka.schema-registry");
+ //This will be used as kafka's group name
+ options.setApplicationName("MyApplication");
+ options.setConsulHost(ip);
+ server = new ForkliftServer(options);
+ try {
+ server.startServer(10000, TimeUnit.SECONDS);
+ // Forklift closes Unirest...we need to reopen it with a call to Options.refresh()
+ Options.refresh();
+ } catch (InterruptedException e) {
+ }
+ if (server.getServerState() == ServerState.RUNNING) {
+ //Register any Consumers you wish to run in the embedded forklift
+ server.registerDeployment(TestStringConsumer.class, TestMapConsumer.class, TestPersonConsumer.class);
+ } else {
+ try {
+ server.stopServer(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ }
+ System.exit(-1);
+ }
+
+
+ public void stopApplication() {
+ if(server != null){
+ try {
+ server.stopServer(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.error("Forklift shutdown interrupted", e);
+ }
+ }
+ }
+ }
+```
diff --git a/server/build.sbt b/server/build.sbt
index 42e72fe..0f33f04 100644
--- a/server/build.sbt
+++ b/server/build.sbt
@@ -2,7 +2,7 @@ organization := "com.github.dcshock"
name := "forklift-server"
-version := "0.32"
+version := "0.34"
enablePlugins(JavaAppPackaging)
@@ -15,8 +15,9 @@ initialize := {
}
libraryDependencies ++= Seq(
- "com.github.dcshock" % "forklift" % "0.22",
+ "com.github.dcshock" % "forklift" % "0.23",
"com.github.dcshock" % "forklift-activemq" % "0.10",
+ "com.github.dcshock" % "forklift-kafka" % "0.1",
"org.apache.activemq" % "activemq-broker" % "5.14.0",
"com.github.dcshock" % "forklift-replay" % "0.14",
"com.github.dcshock" % "forklift-retry" % "0.11",
diff --git a/server/src/main/java/forklift/ForkliftMain.java b/server/src/main/java/forklift/ForkliftMain.java
new file mode 100644
index 0000000..57e1aa9
--- /dev/null
+++ b/server/src/main/java/forklift/ForkliftMain.java
@@ -0,0 +1,62 @@
+package forklift;
+
+import org.kohsuke.args4j.CmdLineException;
+import org.kohsuke.args4j.CmdLineParser;
+
+import java.io.File;
+import java.util.concurrent.TimeUnit;
+
+import static org.kohsuke.args4j.ExampleMode.ALL;
+
+/**
+ * Created by afrieze on 11/4/16.
+ */
+public class ForkliftMain {
+
+ /**
+ * Launch a Forklift server instance.
+ */
+ public static void main(String[] args) throws Throwable {
+ final ForkliftOpts opts = new ForkliftOpts();
+ final CmdLineParser argParse = new CmdLineParser(opts);
+ try {
+ argParse.parseArgument(args);
+ } catch (CmdLineException e) {
+ // if there's a problem in the command line,
+ // you'll get this exception. this will report
+ // an error message.
+ System.err.println(e.getMessage());
+ System.err.println("java SampleMain [options...] arguments...");
+ // print the list of available options
+ argParse.printUsage(System.err);
+ System.err.println();
+ // print option sample. This is useful some time
+ System.err.println(" Example: java SampleMain" + argParse.printExample(ALL));
+
+ return;
+ }
+
+ File f = new File(opts.getConsumerDir());
+ if (!f.exists() || !f.isDirectory()) {
+ System.err.println();
+ System.err.println(" -monitor1 is not a valid directory.");
+ System.err.println();
+ argParse.printUsage(System.err);
+ System.err.println();
+ return;
+ }
+ ForkliftServer server = new ForkliftServer(opts);
+ server.startServer(20, TimeUnit.SECONDS);
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ try {
+ server.stopServer(10, TimeUnit.SECONDS);
+ } catch(InterruptedException e){
+ e.printStackTrace(System.out);
+ }
+ }
+ });
+ }
+
+}
diff --git a/server/src/main/java/forklift/ForkliftOpts.java b/server/src/main/java/forklift/ForkliftOpts.java
index 2d976ca..61d4e77 100644
--- a/server/src/main/java/forklift/ForkliftOpts.java
+++ b/server/src/main/java/forklift/ForkliftOpts.java
@@ -3,49 +3,52 @@
import org.kohsuke.args4j.Option;
public class ForkliftOpts {
- @Option(name="-monitor1", required=true, usage="consumer deployment directory")
+ @Option(name = "-monitor1", required = true, usage = "consumer deployment directory")
private String consumerDir;
- @Option(name="-monitor2", usage="consumer deployment directory")
+ @Option(name = "-monitor2", usage = "consumer deployment directory")
private String propsDir;
- @Option(name="-url", required=true, usage="broker connection url")
+ @Option(name = "-url", required = false, usage = "specify the application name for this instance")
+ private String applicationName;
+
+ @Option(name = "-url", required = true, usage = "broker connection url")
private String brokerUrl;
- @Option(name="-retryDir", usage="directory for persisted retry messages")
+ @Option(name = "-retryDir", usage = "directory for persisted retry messages")
private String retryDir;
- @Option(name="-retryESHost", usage="elastic search host name for retry storage")
+ @Option(name = "-retryESHost", usage = "elastic search host name for retry storage")
private String retryESHost;
- @Option(name="-retryESPort", usage="elastic search port number for retry storage")
+ @Option(name = "-retryESPort", usage = "elastic search port number for retry storage")
private int retryESPort = 9200;
- @Option(name="-retryESSsl", usage="connect to elastic search via ssl (https://)")
+ @Option(name = "-retryESSsl", usage = "connect to elastic search via ssl (https://)")
private boolean retryESSsl;
- @Option(name="-runRetries", usage="run retries on this instance")
+ @Option(name = "-runRetries", usage = "run retries on this instance")
private boolean runRetries;
- @Option(name="-replayDir", usage="replay log directory")
+ @Option(name = "-replayDir", usage = "replay log directory")
private String replayDir;
- @Option(name="-replayESHost", usage="elastic search host name for replay storage")
+ @Option(name = "-replayESHost", usage = "elastic search host name for replay storage")
private String replayESHost;
- @Option(name="-replayESPort", usage="elastic search port number for replay storage")
+ @Option(name = "-replayESPort", usage = "elastic search port number for replay storage")
private int replayESPort = 9200;
- @Option(name="-replayESSsl", usage="connect to elastic search via ssl (https://)")
+ @Option(name = "-replayESSsl", usage = "connect to elastic search via ssl (https://)")
private boolean replayESSsl;
- @Option(name="-replayESServer", usage="start an embedded elastic search server")
+ @Option(name = "-replayESServer", usage = "start an embedded elastic search server")
private boolean replayESServer;
- @Option(name="-consulHost", usage="consul host name")
+ @Option(name = "-consulHost", usage = "consul host name")
private String consulHost = "localhost";
- @Option(name="-replayESCluster", usage="name of the elastic search cluster to use for replay logs.")
+ @Option(name = "-replayESCluster", usage = "name of the elastic search cluster to use for replay logs.")
private String replayESCluster = "elasticsearch";
public String getConsumerDir() {
@@ -167,4 +170,12 @@ public String getReplayESCluster() {
public void setReplayESCluster(String replayESCluster) {
this.replayESCluster = replayESCluster;
}
+
+ public String getApplicationName() {
+ return applicationName;
+ }
+
+ public void setApplicationName(String applicationName) {
+ this.applicationName = applicationName;
+ }
}
diff --git a/server/src/main/java/forklift/ForkliftServer.java b/server/src/main/java/forklift/ForkliftServer.java
index f36fe95..1024b80 100644
--- a/server/src/main/java/forklift/ForkliftServer.java
+++ b/server/src/main/java/forklift/ForkliftServer.java
@@ -1,12 +1,20 @@
package forklift;
-import static org.kohsuke.args4j.ExampleMode.ALL;
-
+import com.google.common.base.Preconditions;
import consul.Consul;
import forklift.connectors.ActiveMQConnector;
import forklift.connectors.ForkliftConnectorI;
+import forklift.connectors.KafkaConnector;
import forklift.consumer.ConsumerDeploymentEvents;
import forklift.consumer.LifeCycleMonitors;
+import forklift.decorators.CoreService;
+import forklift.decorators.Queue;
+import forklift.decorators.Service;
+import forklift.decorators.Topic;
+import forklift.decorators.Topics;
+import forklift.deployment.ClassDeployment;
+import forklift.deployment.Deployment;
+import forklift.deployment.DeploymentManager;
import forklift.deployment.DeploymentWatch;
import forklift.replay.ReplayES;
import forklift.replay.ReplayLogger;
@@ -14,19 +22,36 @@
import forklift.retry.RetryHandler;
import forklift.stats.StatsCollector;
import org.apache.activemq.broker.BrokerService;
-import org.kohsuke.args4j.CmdLineException;
-import org.kohsuke.args4j.CmdLineParser;
-
+import org.apache.http.annotation.ThreadSafe;
import java.io.File;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.io.FileNotFoundException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
/**
- * Start Forklift as a server.
+ * Start Forklift as a server. A running forklift server is responsible for
+ *
+ * 1. Watching a directory for new deployments
+ * 2. Watching a directory for new properties
+ * 3. Accepting new deployments at runtime via the {@link #registerDeployment(Class[])} method
+ * 4. Optionally running a Broker
+ * 5. Connections to the Broker
+ * 6. Starting and managing the Deployment lifecycles
+ * 7. Retry Strategy
+ * 8. Replay Strategy
+ *
+ *
* @author zdavep
*/
+@ThreadSafe
public final class ForkliftServer {
- // Lock Waits
- private static final AtomicBoolean running = new AtomicBoolean(false);
+
+ private ExecutorService executor = Executors.newSingleThreadExecutor();
+
+ private volatile ServerState state = ServerState.LATENT;
// Logging
private static org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(ForkliftServer.class);
@@ -35,174 +60,295 @@ public final class ForkliftServer {
private static int SLEEP_INTERVAL = 10000; // 10 seconds
private static BrokerService broker = null;
+ private final ForkliftOpts opts;
+ private DeploymentWatch consumerWatch;
+ private Forklift forklift;
+ private DeploymentWatch propsWatch;
+
+ public ForkliftServer(ForkliftOpts options) {
+ this.opts = options;
+ }
+
+ private ReplayES replayES;
+ private ConsumerDeploymentEvents deploymentEvents;
+ private DeploymentManager classDeployments = new DeploymentManager();
+ private CountDownLatch runningLatch = new CountDownLatch(1);
/**
- * Launch a Forklift server instance.
+ * Attempts to start the forklift server. This call is blocking and will not return until either the server starts successfully or the waitTime reaches 0.
+ * A response of false does not mean the server will stop its attempt to startup. This method may only be called once.
+ *
+ * @param waitTime the maximum time to wait
+ * @param timeUnit the time unit of the waitTime
+ * @return the {@link ServerState state} of the server at the time this method returns
+ * @throws InterruptedException if the current thread is interrupted while waiting for the server to start
+ * @throws IllegalStateException if this method has already been called
*/
- public static void main(String[] args) throws Throwable {
- final ForkliftOpts opts = new ForkliftOpts();
- final CmdLineParser argParse = new CmdLineParser(opts);
- try {
- argParse.parseArgument(args);
- } catch( CmdLineException e ) {
- // if there's a problem in the command line,
- // you'll get this exception. this will report
- // an error message.
- System.err.println(e.getMessage());
- System.err.println("java SampleMain [options...] arguments...");
- // print the list of available options
- argParse.printUsage(System.err);
- System.err.println();
-
- // print option sample. This is useful some time
- System.err.println(" Example: java SampleMain" + argParse.printExample(ALL));
-
- return;
+ public ServerState startServer(long waitTime, TimeUnit timeUnit) throws InterruptedException {
+ synchronized (this) {
+ Preconditions.checkState(state == ServerState.LATENT);
+ state = ServerState.STARTING;
+ log.info("Forklift server starting");
}
-
- File f = new File(opts.getConsumerDir());
- if (!f.exists() || !f.isDirectory()) {
- System.err.println();
- System.err.println(" -monitor1 is not a valid directory.");
- System.err.println();
- argParse.printUsage(System.err);
- System.err.println();
- return;
+ executor.execute(this::launch);
+ try {
+ this.runningLatch.await(waitTime, timeUnit);
+ return state;
+ } catch (InterruptedException e) {
+ log.error("Launch Interrupted", e);
+ throw e;
}
+ }
+ /**
+ * Stops the ForkliftServer. This call is blocking and will not return until either the server is shutdown or the waitTime reaches 0.
+ *
+ * @param waitTime the maximum time to wait
+ * @param timeUnit the time unit of the waitTime
+ * @return the {@link ServerState state} of the server at the time this method returns
+ * @throws InterruptedException if the current thread is interrupted before the waitTime has ellapsed
+ */
+ public ServerState stopServer(long waitTime, TimeUnit timeUnit) throws InterruptedException {
+ shutdown();
+ executor.shutdownNow();
+ executor.awaitTermination(waitTime, timeUnit);
+ return state;
+ }
- String brokerUrl = opts.getBrokerUrl();
- if (brokerUrl.startsWith("consul.") && brokerUrl.length() > "consul.".length()) {
- log.info("Building failover url using consul");
-
- final Consul c = new Consul("http://" + opts.getConsulHost(), 8500);
-
- // Build the connection string.
- final String serviceName = brokerUrl.split("\\.")[1];
-
- brokerUrl = "failover:(" +
- c.catalog().service(serviceName).getProviders().stream()
- .filter(srvc -> !srvc.isCritical())
- .map(srvc -> "tcp://" + srvc.getAddress() + ":" + srvc.getPort())
- .reduce("", (a, b) -> a + "," + b) +
- ")";
-
- c.shutdown();
+ /**
+ * @return the current {@link ServerState state} of the server
+ */
+ public ServerState getServerState() {
+ return state;
+ }
- brokerUrl = brokerUrl.replaceAll("failover:\\(,", "failover:(");
+ /**
+ * Registers a collection of classes as new deployment. Classes are scanned for the {@link Queue}, {@link Topic},
+ * {@link Topics} {@link Service}, {@link CoreService} annotations.
+ *
+ * @param deploymentClasses the classes which make up the deployment
+ */
+ public synchronized void registerDeployment(Class... deploymentClasses) {
+ Preconditions.checkState(state == ServerState.RUNNING);
+ Deployment deployment = new ClassDeployment(deploymentClasses);
+ if (!classDeployments.isRegistered(deployment)) {
+ classDeployments.register(deployment);
+ deploymentEvents.onDeploy(deployment);
+ }
+ }
- log.info("url {}", brokerUrl);
- if (brokerUrl.equals("failover:()")) {
- log.error("No brokers found");
- System.exit(-1);
+ /**
+ * Launch a Forklift server instance.
+ */
+ private void launch() {
+ if (setupBrokerAndForklift()) {
+ deploymentEvents = new ConsumerDeploymentEvents(forklift);
+ consumerWatch = setupConsumerWatch(deploymentEvents);
+ propsWatch = setupPropertyWatch(deploymentEvents);
+ this.replayES = setupESReplayHandling(forklift);
+ RetryES retryES = setupESRetryHandling(forklift);
+ if (setupLifeCycleMonitors(replayES, retryES, forklift)) {
+ try {
+ runEventLoop(propsWatch, consumerWatch);
+ } catch (InterruptedException ignored) {
+ }
}
- } else if (brokerUrl.startsWith("embed")) {
- brokerUrl = "tcp://0.0.0.0:61616";
- broker = new BrokerService();
+ }
+ if (state != ServerState.STOPPED) {
+ state = ServerState.ERROR;
+ log.info("Forklift server Error state");
+ }
+ }
- // configure the broker
- broker.addConnector(brokerUrl);
- broker.addConnector("stomp://0.0.0.0:61613");
+ private boolean setupBrokerAndForklift() {
+ try {
+ forklift = new Forklift();
+ final ForkliftConnectorI connector = startAndConnectToBroker();
+ forklift.start(connector);
+ } catch (Exception e) {
+ log.error("Unable to startup broker and forklift", e);
+ }
+ return forklift.isRunning();
+ }
- broker.start();
+ private void runEventLoop(DeploymentWatch propsWatch, DeploymentWatch consumerWatch) throws InterruptedException {
+ state = ServerState.RUNNING;
+ log.info("Forklift server Running");
+ runningLatch.countDown();
+ while (state == ServerState.RUNNING) {
+ log.debug("Scanning for new deployments...");
+ try {
+ if (propsWatch != null)
+ propsWatch.run();
+ } catch (Throwable e) {
+ log.error("", e);
+ }
+ try {
+ if (consumerWatch != null)
+ consumerWatch.run();
+ } catch (Throwable e) {
+ log.error("", e);
+ }
+ synchronized (this) {
+ this.wait(SLEEP_INTERVAL);
+ }
}
+ }
- // Start a forklift server w/ specified connector.
- final Forklift forklift = new Forklift();
- final ConsumerDeploymentEvents deploymentEvents = new ConsumerDeploymentEvents(forklift);
+ private void shutdown() {
+ synchronized (this) {
+ if (state != ServerState.RUNNING || state != ServerState.ERROR) {
+ return;
+ }
+ state = ServerState.STOPPING;
+ log.info("Forklift server Stopping");
- final DeploymentWatch deploymentWatch;
- if (opts.getConsumerDir() != null)
- deploymentWatch = new DeploymentWatch(new java.io.File(opts.getConsumerDir()), deploymentEvents);
- else
- deploymentWatch = null;
+ if (replayES != null) {
+ replayES.shutdown();
+ }
+ if (consumerWatch != null) {
+ consumerWatch.shutdown();
+ }
+ if (propsWatch != null) {
+ propsWatch.shutdown();
+ }
+ classDeployments.getAll().forEach(deploy -> deploymentEvents.onUndeploy(deploy));
+ if (consumerWatch != null) {
+ consumerWatch.shutdown();
+ }
+ forklift.shutdown();
- final DeploymentWatch propsWatch;
- if (opts.getPropsDir() != null)
- propsWatch = new DeploymentWatch(new java.io.File(opts.getPropsDir()), deploymentEvents);
- else
- propsWatch = null;
+ if (broker != null) {
+ try {
+ broker.stop();
+ } catch (Exception ignored) {
+ }
+ }
+ state = ServerState.STOPPED;
+ log.info("Forklift server Stopped");
- final ForkliftConnectorI connector = new ActiveMQConnector(brokerUrl);
- forklift.start(connector);
- if (!forklift.isRunning()) {
- throw new RuntimeException("Unable to start Forklift");
+ this.notifyAll();
}
+ }
- log.info("Registering LifeCycleMonitors");
-
+ private boolean setupLifeCycleMonitors(ReplayES replayES, RetryES retryES, Forklift forklift) {
LifeCycleMonitors.register(StatsCollector.class);
+ boolean setup = true;
+ // Setup retry handling.
+ if (retryES != null) {
+ LifeCycleMonitors.register(retryES);
+ }
+ if (opts.getRetryDir() != null) {
+ LifeCycleMonitors.register(new RetryHandler(forklift.getConnector(), new File(opts.getRetryDir())));
+ }
+ // Always add replay last so that other plugins can update props.
+ if (replayES != null)
+ LifeCycleMonitors.register(replayES);
+ if (opts.getReplayDir() != null) {
+ try {
+ LifeCycleMonitors.register(new ReplayLogger(new File(opts.getReplayDir())));
+ } catch (FileNotFoundException e) {
+ log.error("Unable to find file for Replay Logger", e);
+ setup = false;
+ }
+ }
+ return setup;
+ }
+ private ReplayES setupESReplayHandling(Forklift forklift) {
// Create the replay ES first if it's needed just in case we are utilizing the startup of the embedded es engine.
final ReplayES replayES;
if (opts.getReplayESHost() == null)
replayES = null;
else
- replayES = new ReplayES(!opts.isReplayESServer(), opts.getReplayESHost(), opts.getReplayESPort(), opts.getReplayESCluster(), connector);
+ replayES =
+ new ReplayES(!opts.isReplayESServer(), opts.getReplayESHost(), opts.getReplayESPort(),
+ opts.getReplayESCluster(), forklift.getConnector());
+ return replayES;
+ }
- // Setup retry handling.
- if (opts.getRetryDir() != null)
- LifeCycleMonitors.register(new RetryHandler(forklift.getConnector(), new File(opts.getRetryDir())));
+ private RetryES setupESRetryHandling(Forklift forklift) {
+ RetryES retryES = null;
if (opts.getRetryESHost() != null)
- LifeCycleMonitors.register(new RetryES(forklift.getConnector(), opts.isRetryESSsl(), opts.getRetryESHost(), opts.getRetryESPort(), opts.isRunRetries()));
-
- // Always add replay last so that other plugins can update props.
- if (replayES != null)
- LifeCycleMonitors.register(replayES);
- if (opts.getReplayDir() != null)
- LifeCycleMonitors.register(new ReplayLogger(new File(opts.getReplayDir())));
-
- log.info("Connected to broker on " + brokerUrl);
- log.info("Scanning for Forklift consumers at " + opts.getConsumerDir());
- log.info("Scanning for Forklift consumers at " + opts.getPropsDir());
-
- Runtime.getRuntime().addShutdownHook(new Thread() {
- @Override
- public void run() {
- replayES.shutdown();
-
- // End the deployment watcher.
- running.set(false);
- synchronized (running) {
- running.notify();
- }
-
- if (deploymentWatch != null)
- deploymentWatch.shutdown();
-
- if (propsWatch != null)
- propsWatch.shutdown();
-
- forklift.shutdown();
+ retryES =
+ new RetryES(forklift.getConnector(), opts.isRetryESSsl(), opts.getRetryESHost(), opts.getRetryESPort(),
+ opts.isRunRetries());
+ return retryES;
+ }
- if (broker != null)
- try {
- broker.stop();
- } catch (Exception ignored) { }
- }
- });
+ private DeploymentWatch setupConsumerWatch(ConsumerDeploymentEvents deploymentEvents) {
+ final DeploymentWatch deploymentWatch;
+ if (opts.getConsumerDir() != null) {
+ deploymentWatch = new DeploymentWatch(new java.io.File(opts.getConsumerDir()), deploymentEvents);
+ log.info("Scanning for Forklift consumers at " + opts.getConsumerDir());
+ } else {
+ deploymentWatch = null;
+ }
+ return deploymentWatch;
+ }
- running.set(true);
- while (running.get()) {
- log.debug("Scanning for new deployments...");
+ private DeploymentWatch setupPropertyWatch(ConsumerDeploymentEvents deploymentEvents) {
+ final DeploymentWatch propsWatch;
+ if (opts.getPropsDir() != null) {
+ propsWatch = new DeploymentWatch(new java.io.File(opts.getPropsDir()), deploymentEvents);
+ log.info("Scanning for Properties at " + opts.getPropsDir());
+ } else {
+ propsWatch = null;
+ }
+ return propsWatch;
+ }
- try {
- if (propsWatch != null)
- propsWatch.run();
- } catch (Throwable e) {
- log.error("", e);
- }
+ private ForkliftConnectorI startAndConnectToBroker() throws Exception {
+ String brokerUrl = opts.getBrokerUrl();
+ ForkliftConnectorI connector = null;
+ if (brokerUrl.startsWith("consul.") && brokerUrl.length() > "consul.".length()) {
- try {
- if (deploymentWatch != null)
- deploymentWatch.run();
- } catch (Throwable e) {
- log.error("", e);
- }
+ final Consul c = new Consul("http://" + opts.getConsulHost(), 8500);
+ // Build the connection string.
+ final String serviceName = brokerUrl.split("\\.")[1];
- synchronized (running) {
- running.wait(SLEEP_INTERVAL);
+ if ("kafka".equalsIgnoreCase(serviceName)) {
+ String schemaRegistry = brokerUrl.split("\\.")[2];
+ brokerUrl = c.catalog().service(serviceName).getProviders().stream()
+ .filter(srvc -> !srvc.isCritical())
+ .map(srvc -> srvc.getAddress() + ":" + srvc.getPort())
+ .collect(Collectors.joining(","));
+ String schemaRegistries = c.catalog().service(schemaRegistry).getProviders().stream()
+ .filter(srvc -> !srvc.isCritical())
+ .map(srvc -> "http://" + srvc.getAddress() + ":" + srvc.getPort())
+ .collect(Collectors.joining(","));
+
+ String applicationName = this.opts.getApplicationName() == null?"forklift":this.opts.getApplicationName();
+ connector = new KafkaConnector(brokerUrl,schemaRegistries, applicationName);
+
+ } else {
+
+ log.info("Building failover url using consul");
+ brokerUrl = "failover:(" +
+ c.catalog().service(serviceName).getProviders().stream()
+ .filter(srvc -> !srvc.isCritical())
+ .map(srvc -> "tcp://" + srvc.getAddress() + ":" + srvc.getPort())
+ .reduce("", (a, b) -> a + "," + b) +
+ ")";
+
+ brokerUrl = brokerUrl.replaceAll("failover:\\(,", "failover:(");
+
+ log.info("url {}", brokerUrl);
+ if (brokerUrl.equals("failover:()")) {
+ log.error("No brokers found");
+ System.exit(-1);
+ }
+ connector = new ActiveMQConnector(brokerUrl);
}
+ } else if (brokerUrl.startsWith("embed")) {
+ brokerUrl = "tcp://0.0.0.0:61616";
+ broker = new BrokerService();
+ broker.addConnector(brokerUrl);
+ broker.addConnector("stomp://0.0.0.0:61613");
+ broker.start();
+ connector = new ActiveMQConnector(brokerUrl);
}
+ log.info("Connected to broker on " + brokerUrl);
+ return connector;
}
}
diff --git a/server/src/main/java/forklift/ServerState.java b/server/src/main/java/forklift/ServerState.java
new file mode 100644
index 0000000..abd0134
--- /dev/null
+++ b/server/src/main/java/forklift/ServerState.java
@@ -0,0 +1,15 @@
+package forklift;
+
+/**
+ * The various states which a server may be in
+ *
+ * Created by afrieze on 11/4/16.
+ */
+public enum ServerState {
+ LATENT,
+ STARTING,
+ ERROR,
+ RUNNING,
+ STOPPING,
+ STOPPED
+}
diff --git a/server/src/test/java/forklift/integration/kafka/Consumer/Person.java b/server/src/test/java/forklift/integration/kafka/Consumer/Person.java
new file mode 100644
index 0000000..a290305
--- /dev/null
+++ b/server/src/test/java/forklift/integration/kafka/Consumer/Person.java
@@ -0,0 +1,34 @@
+package forklift.integration.kafka.Consumer;
+
+/**
+ * Created by afrieze on 3/2/17.
+ */
+public class Person {
+ private String firstName;
+ private String lastName;
+
+ public Person(){
+
+ }
+
+ public Person(String firstName, String lastName){
+ this.firstName = firstName;
+ this.lastName = lastName;
+ }
+
+ public String getFirstName() {
+ return firstName;
+ }
+
+ public void setFirstName(String firstName) {
+ this.firstName = firstName;
+ }
+
+ public String getLastName() {
+ return lastName;
+ }
+
+ public void setLastName(String lastName) {
+ this.lastName = lastName;
+ }
+}
diff --git a/server/src/test/java/forklift/integration/kafka/Consumer/TestMapConsumer.java b/server/src/test/java/forklift/integration/kafka/Consumer/TestMapConsumer.java
new file mode 100644
index 0000000..b1d8efe
--- /dev/null
+++ b/server/src/test/java/forklift/integration/kafka/Consumer/TestMapConsumer.java
@@ -0,0 +1,24 @@
+package forklift.integration.kafka.Consumer;
+
+import forklift.decorators.Message;
+import forklift.decorators.OnMessage;
+import forklift.decorators.Topic;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Created by afrieze on 3/2/17.
+ */
+@Topic("forklift-mapTopic")
+public class TestMapConsumer {
+ private static org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(TestStringConsumer.class);
+
+ @Message
+ private Map message;
+
+ @OnMessage
+ public void consume() {
+ String mapValues = message.keySet().stream().map(x->x + "=" + message.get(x)).collect(Collectors.joining(","));
+ log.info("Consumer for queue forklift-objectTopic consumed map with values: " + mapValues);
+ }
+}
diff --git a/server/src/test/java/forklift/integration/kafka/Consumer/TestPersonConsumer.java b/server/src/test/java/forklift/integration/kafka/Consumer/TestPersonConsumer.java
new file mode 100644
index 0000000..57346cc
--- /dev/null
+++ b/server/src/test/java/forklift/integration/kafka/Consumer/TestPersonConsumer.java
@@ -0,0 +1,21 @@
+package forklift.integration.kafka.Consumer;
+
+import forklift.decorators.Message;
+import forklift.decorators.OnMessage;
+import forklift.decorators.Topic;
+
+/**
+ * Created by afrieze on 3/2/17.
+ */
+@Topic("forklift-pojoTopic")
+public class TestPersonConsumer {
+ private static org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(TestStringConsumer.class);
+
+ @Message
+ private Person person;
+
+ @OnMessage
+ public void consume() {
+ log.info("Consumer for queue forklift-personTopic consumed value: " + person.getFirstName());
+ }
+}
diff --git a/server/src/test/java/forklift/integration/kafka/Consumer/TestStringConsumer.java b/server/src/test/java/forklift/integration/kafka/Consumer/TestStringConsumer.java
new file mode 100644
index 0000000..3ee5a03
--- /dev/null
+++ b/server/src/test/java/forklift/integration/kafka/Consumer/TestStringConsumer.java
@@ -0,0 +1,23 @@
+package forklift.integration.kafka.Consumer;
+
+import forklift.decorators.Message;
+import forklift.decorators.MultiThreaded;
+import forklift.decorators.OnMessage;
+import forklift.decorators.Topic;
+
+/**
+ * Created by afrieze on 3/1/17.
+ */
+@MultiThreaded(10)
+@Topic("forklift-stringTopic")
+public class TestStringConsumer {
+ private static org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(TestStringConsumer.class);
+
+ @Message
+ private String value;
+
+ @OnMessage
+ public void consume() {
+ log.info("Consumer for queue forklift-stringTopic consumed value: " + value);
+ }
+}
diff --git a/server/src/test/java/forklift/integration/kafka/ServerIntegrationTests.java b/server/src/test/java/forklift/integration/kafka/ServerIntegrationTests.java
new file mode 100644
index 0000000..eb29d80
--- /dev/null
+++ b/server/src/test/java/forklift/integration/kafka/ServerIntegrationTests.java
@@ -0,0 +1,51 @@
+package forklift.integration.kafka;
+
+import static org.junit.Assert.assertTrue;
+import com.mashape.unirest.http.options.Options;
+import forklift.ForkliftOpts;
+import forklift.ForkliftServer;
+import forklift.ServerState;
+import forklift.integration.kafka.Consumer.TestMapConsumer;
+import forklift.integration.kafka.Consumer.TestPersonConsumer;
+import forklift.integration.kafka.Consumer.TestStringConsumer;
+import org.junit.Ignore;
+import org.junit.Test;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Kafka integration test. Tests the embeddable server with a live kafka and confluent schema registry instance. These
+ * tests pair well with the integration tests in the Kafka connector project.
+ *
+ * Note: You will need to adjust the broker url to match your own environment.
+ */
+public class ServerIntegrationTests {
+
+ @Ignore
+ @Test
+ public void testConsumers() throws InterruptedException {
+ ForkliftOpts options = new ForkliftOpts();
+ String ip = "localhost";
+ options.setBrokerUrl("consul.kafka.schema-registry");
+ options.setApplicationName("app10");
+ options.setConsulHost(ip);
+ ForkliftServer server = new ForkliftServer(options);
+ try {
+ server.startServer(10000, TimeUnit.SECONDS);
+ // Forklift closes Unirest...we need to reopen it with a call to Options.refresh()
+ Options.refresh();
+ } catch (InterruptedException e) {
+ }
+ if (server.getServerState() == ServerState.RUNNING) {
+ server.registerDeployment(TestStringConsumer.class, TestMapConsumer.class, TestPersonConsumer.class);
+ } else {
+ try {
+ server.stopServer(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ }
+ System.exit(-1);
+ }
+ Thread.sleep(130000);
+ server.stopServer(5, TimeUnit.SECONDS);
+ assertTrue(true);
+ }
+}