diff --git a/src/main/java/kafka/VolthaKafkaConsumer.java b/src/main/java/kafka/VolthaKafkaConsumer.java
new file mode 100644
index 0000000..a9f0603
--- /dev/null
+++ b/src/main/java/kafka/VolthaKafkaConsumer.java
@@ -0,0 +1,121 @@
+/*
+ * Copyright 2018- Cisco
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka;
+import org.apache.kafka.clients.consumer.*;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.KafkaException;
+import java.util.Collections;
+import java.util.Properties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.Marker;
+import org.slf4j.MarkerFactory;
+
+import javax.xml.ws.http.HTTPException;
+
+import java.util.concurrent.TimeUnit;
+
+import ves.*;
+import config.Config;
+
+public class VolthaKafkaConsumer {
+
+    private final Logger logger = LoggerFactory.getLogger("VolthaKafkaConsumer");
+    private final String dataMarkerText = "DATA";
+    private final Marker dataMarker = MarkerFactory.getMarker(dataMarkerText);
+
+    private KafkaConsumer<Long, String> consumer;
+
+    public VolthaKafkaConsumer() {
+      logger.debug("VolthaKafkaConsumer constructor called");
+      initVesAgent();
+      consumer = createConsumer();
+    }
+
+    private void initVesAgent() {
+      VesAgent.initVes();
+    }
+
+    private KafkaConsumer<Long, String> createConsumer() {
+      logger.debug("Creating Kafka Consumer");
+
+      String kafkaAddress = Config.getKafkaAddress() + ":" + Config.getKafkaPort();
+      final Properties props = new Properties();
+      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
+                                  kafkaAddress);
+      props.put(ConsumerConfig.GROUP_ID_CONFIG,
+                                  "KafkaExampleConsumer");
+      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+              LongDeserializer.class.getName());
+      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+              StringDeserializer.class.getName());
+      props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
+              false);
+
+      // Create the consumer using props.
+      final KafkaConsumer<Long, String> consumer =
+                                  new KafkaConsumer<>(props);
+
+      // Subscribe to the topic.
+      consumer.subscribe(Collections.singletonList(Config.getKafkaTopic()));
+      return consumer;
+  }
+
+  public void runConsumer() throws InterruptedException {
+
+      logger.debug("Starting Consumer");
+
+      while (true) {
+	  ConsumerRecords<Long, String> consumerRecords;
+ 	  try {
+	  	consumerRecords = consumer.poll(100000);
+	  } catch (KafkaException e) {
+		logger.debug("Error with Kafka connection. Retrying in 15 seconds.");
+		this.consumer = createConsumer();
+		TimeUnit.SECONDS.sleep(15);
+		continue;
+	  }
+          logger.info("{} Records retrieved from poll.", consumerRecords.count());
+
+          boolean commit = true;
+          try {
+            consumerRecords.forEach(record -> {
+              logger.info(dataMarker, "Consumer Record:({}, {}, {}, {})\n",
+                      record.key(), record.value(),
+                      record.partition(), record.offset());
+              logger.info("Attempting to send data to VES");
+              boolean success = VesAgent.sendToVES(record.value());
+              if (!success) {
+                throw new HTTPException(0);
+              } else {
+                logger.info("Sent Ves Message");
+              }
+            });
+          } catch (HTTPException e) {
+            logger.info("Ves message failed. Going back to polling.");
+            commit = false;
+          }
+          if (commit) {
+            consumer.commitAsync();
+          }
+      }
+      //consumer.close();
+      //logger.debug("DONE");
+    }
+
+}