diff --git a/src/main/java/config/Config.java b/src/main/java/config/Config.java
new file mode 100644
index 0000000..abf5d73
--- /dev/null
+++ b/src/main/java/config/Config.java
@@ -0,0 +1,68 @@
+/*
+ * 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 config;
+
+import java.util.Properties;
+import java.io.FileInputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class Config {
+
+  private static Properties properties;
+
+  private final static Logger logger = LoggerFactory.getLogger("VolthaKafkaConsumer");
+
+  public static void loadProperties(String file) {
+    // create application properties with default
+    try {
+      properties = new Properties();
+
+      // now load properties
+      // from last invocation
+      FileInputStream in = new FileInputStream(file);
+      properties.load(in);
+      in.close();
+    } catch (Exception e) {
+      logger.error(e.getMessage());
+    }
+  }
+
+  public static String get(String key) {
+    return (String)properties.get(key);
+  }
+
+  public static String getVesAddress() {
+    return get("onap_ves_address");
+  }
+
+  public static String getVesPort() {
+    return get("onap_ves_port");
+  }
+
+  public static String getKafkaAddress() {
+    return get("kafka_address");
+  }
+
+  public static String getKafkaPort() {
+    return get("kafka_port");
+  }
+
+  public static String getKafkaTopic() {
+    return get("kafka_topic");
+  }
+}
diff --git a/src/main/java/controller/Application.java b/src/main/java/controller/Application.java
new file mode 100644
index 0000000..6ae05b8
--- /dev/null
+++ b/src/main/java/controller/Application.java
@@ -0,0 +1,63 @@
+/*
+ * 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 controller;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+import kafka.VolthaKafkaConsumer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.InterruptedException;
+
+import config.Config;
+
+
+@SpringBootApplication
+@RestController
+public class Application {
+
+    @RequestMapping("/")
+    public String home() {
+        return "Hello Docker World";
+    }
+
+    public static void main(String[] args) {
+        Config.loadProperties("/opt/ves-agent/config.properties");
+	      KafkaThread kafka = new KafkaThread();
+        kafka.start();
+        SpringApplication.run(Application.class, args);
+    }
+
+}
+class KafkaThread extends Thread {
+
+    private final static Logger logger = LoggerFactory.getLogger("KafkaThread");
+
+    public void run() {
+      logger.debug("Start Kafka Consumer Thread");
+       try {
+         VolthaKafkaConsumer consumer = new VolthaKafkaConsumer();
+       	 consumer.runConsumer();
+       } catch (InterruptedException e) {
+          logger.error(e.getMessage());
+       }
+
+    }
+ }
diff --git a/src/main/java/filter/MarkerFilter.java b/src/main/java/filter/MarkerFilter.java
new file mode 100644
index 0000000..8c20563
--- /dev/null
+++ b/src/main/java/filter/MarkerFilter.java
@@ -0,0 +1,78 @@
+/*
+ * 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 filter;
+
+import java.util.Arrays;
+
+import org.slf4j.Marker;
+
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.filter.AbstractMatcherFilter;
+import ch.qos.logback.core.spi.FilterReply;
+
+import org.slf4j.Marker;
+import org.slf4j.MarkerFactory;
+
+public class MarkerFilter extends AbstractMatcherFilter<ILoggingEvent> {
+
+	Marker markerToMatch;
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see ch.qos.logback.core.filter.Filter#decide(java.lang.Object)
+	 */
+	@Override
+	public FilterReply decide(ILoggingEvent event) {
+		if (!isStarted()) {
+			return FilterReply.NEUTRAL;
+		}
+		Marker marker = event.getMarker();
+		if (marker == null) {
+			return onMismatch;
+		}
+
+		if (markerToMatch.contains(marker)) {
+			return onMatch;
+		} else {
+			return onMismatch;
+		}
+	}
+
+	/**
+	 * The marker to match in the event.
+	 * 
+	 * @param markerToMatch
+	 */
+	public void setMarker(String markerStr) {
+		if (markerStr != null) {
+			Marker marker = MarkerFactory.getMarker(markerStr);
+			this.markerToMatch = marker;
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see ch.qos.logback.core.filter.Filter#start()
+	 */
+	@Override
+	public void start() {
+		if (this.markerToMatch != null) {
+			super.start();
+		}
+	}
+}
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");
+    }
+
+}
diff --git a/src/main/java/ves/VesAgent.java b/src/main/java/ves/VesAgent.java
new file mode 100644
index 0000000..b0465f7
--- /dev/null
+++ b/src/main/java/ves/VesAgent.java
@@ -0,0 +1,82 @@
+/*
+ * 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 ves;
+
+import evel_javalibrary.att.com.*;
+import evel_javalibrary.att.com.AgentMain.EVEL_ERR_CODES;
+import evel_javalibrary.att.com.EvelFault.EVEL_SEVERITIES;
+import evel_javalibrary.att.com.EvelFault.EVEL_SOURCE_TYPES;
+import evel_javalibrary.att.com.EvelFault.EVEL_VF_STATUSES;
+import evel_javalibrary.att.com.EvelHeader.PRIORITIES;
+import evel_javalibrary.att.com.EvelMobileFlow.MOBILE_GTP_PER_FLOW_METRICS;
+import evel_javalibrary.att.com.EvelScalingMeasurement.MEASUREMENT_CPU_USE;
+import evel_javalibrary.att.com.EvelScalingMeasurement.MEASUREMENT_VNIC_PERFORMANCE;
+import evel_javalibrary.att.com.EvelStateChange.EVEL_ENTITY_STATE;
+import evel_javalibrary.att.com.EvelThresholdCross.EVEL_ALERT_TYPE;
+import evel_javalibrary.att.com.EvelThresholdCross.EVEL_EVENT_ACTION;
+import java.net.HttpURLConnection;
+
+import org.apache.log4j.Level;
+import config.Config;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class VesAgent {
+
+  private static final Logger logger = LoggerFactory.getLogger("VesAgent");
+
+  public static void initVes() {
+    logger.info("Initializing VES Agent");
+    try {
+        AgentMain.evel_initialize("http://"+Config.getVesAddress(),
+                Integer.parseInt(Config.getVesPort()),
+                //  "http://1.2.3.4", 8080,
+                //"/vendor_event_listener","/example_vnf",
+                null,null,
+                "will",
+                "pill",
+                null, null, null,
+                //"/home/gokul/newwk/demo/vnfs/VES5.0/evel/sslcerts2/my-keystore.jks", "changeit", "changeit",
+                Level.TRACE);
+    } catch( Exception e ) {
+        e.printStackTrace();
+    }
+  }
+
+  public static boolean sendToVES(String json) {
+
+    EvelFault flt  = new EvelFault("Fault_VOLTHA_failed", "tbd_event_key_unique_to_source",
+            "NIC error", "Hardware failed",
+            EvelHeader.PRIORITIES.EVEL_PRIORITY_HIGH,
+            EVEL_SEVERITIES.EVEL_SEVERITY_MAJOR,
+            EVEL_SOURCE_TYPES.EVEL_SOURCE_CARD,
+            EVEL_VF_STATUSES.EVEL_VF_STATUS_ACTIVE);
+    flt.evel_fault_addl_info_add("voltha", json);
+    //flt.evel_fault_addl_info_add("nicsw", "fail");
+    flt.evel_fault_category_set("Communication");
+    logger.info("Sending fault event");
+    int code = AgentMain.evel_post_event_immediate(flt);
+    logger.info("Fault event http code received: " + code);
+    if(code == 0 || code >= HttpURLConnection.HTTP_BAD_REQUEST )
+    {
+      return false;
+    } else {
+      return true;
+    }
+  }
+
+}
diff --git a/src/main/resources/log4j.properties b/src/main/resources/log4j.properties
new file mode 100644
index 0000000..8c67d07
--- /dev/null
+++ b/src/main/resources/log4j.properties
@@ -0,0 +1,11 @@
+logDir=/opt/ves-agent/
+
+log4j.rootLogger=TRACE, stdout
+#log4j.rootLogger=DEBUG, stdout
+
+log4j.appender.stdout=org.apache.log4j.DailyRollingFileAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%d{MM/dd/yyyy hh:mm:ss a}|%-5p|%-30c{1}| %m%n
+log4j.appender.stdout.DatePattern='.'yyyy-MM-dd
+log4j.appender.stdout.File=${logDir}/vesagent.log
+log4j.appender.stdout.append=true
diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml
new file mode 100644
index 0000000..89b8950
--- /dev/null
+++ b/src/main/resources/logback.xml
@@ -0,0 +1,68 @@
+<!--
+ 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.
+-->
+<configuration>
+
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+
+    <logger name="org.apache.kafka" level="INFO"/>
+    <logger name="org.apache.kafka.common.metrics" level="INFO"/>
+    
+    <appender name="FILE_MAIN" class="ch.qos.logback.core.rolling.RollingFileAppender">
+      <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+        <fileNamePattern>/opt/ves-agent/log-%d{yyyy-MM-dd}.txt</fileNamePattern>
+        <maxHistory>30</maxHistory>
+        <totalSizeCap>5GB</totalSizeCap>
+      </rollingPolicy>
+      <append>true</append>
+      <encoder>
+         <pattern>%-4relative [%thread] %-5level %logger{35} - %msg%n</pattern>
+      </encoder>
+   </appender>
+
+   <appender name="FILE_DATA" class="ch.qos.logback.core.rolling.RollingFileAppender">
+      <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+        <fileNamePattern>/opt/ves-agent/data-%d{yyyy-MM-dd}.txt</fileNamePattern>
+        <maxHistory>30</maxHistory> 
+        <totalSizeCap>1GB</totalSizeCap>
+      </rollingPolicy>
+      <append>true</append>
+      <filter class="filter.MarkerFilter">
+        <marker>DATA</marker>
+        <onMismatch>DENY</onMismatch>
+      </filter>
+      <filter class="filter.MarkerFilter">
+	<marker>DATA</marker>
+	<onMatch>ACCEPT</onMatch>
+      </filter>
+      <encoder>
+         <pattern>%-4relative [%thread] %-5level %logger{35} - %msg%n</pattern>
+      </encoder>
+   </appender>
+
+   <logger name="org.springframework" level="WARNING"/>
+
+   <root level="debug">
+      <appender-ref ref="FILE_MAIN" />
+      <appender-ref ref="FILE_DATA" />
+      <appender-ref ref="STDOUT" />
+   </root>
+
+</configuration>