diff --git a/apps/kafka-integration/api/pom.xml b/apps/kafka-integration/api/pom.xml
new file mode 100644
index 0000000..56905f4
--- /dev/null
+++ b/apps/kafka-integration/api/pom.xml
@@ -0,0 +1,58 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2016-present Open Networking Laboratory
+  ~
+  ~ 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+              xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+              xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <artifactId>onos-kafka</artifactId>
+        <groupId>org.onosproject</groupId>
+        <version>1.7.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>onos-app-kafka-api</artifactId>
+    <packaging>bundle</packaging>
+
+    <url>http://onosproject.org</url>
+
+    <description>Kafka Integration Application</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+            <version>3.0.0-beta-2</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+       <plugins>
+          <plugin>
+              <groupId>org.apache.felix</groupId>
+              <artifactId>maven-bundle-plugin</artifactId>
+              <extensions>true</extensions>
+              <configuration>
+                  <instructions>
+                  </instructions>
+              </configuration>
+          </plugin>
+       </plugins>
+    </build>
+</project>
+
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/EventExporterService.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/EventExporterService.java
similarity index 90%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/EventExporterService.java
rename to apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/EventExporterService.java
index 3ca5414..e203697 100644
--- a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/EventExporterService.java
+++ b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/EventExporterService.java
@@ -18,11 +18,13 @@
 import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
 import org.onosproject.kafkaintegration.errors.InvalidApplicationException;
 import org.onosproject.kafkaintegration.errors.InvalidGroupIdException;
-import org.onosproject.kafkaintegration.errors.UnsupportedEventException;
+
+import com.google.common.annotations.Beta;
 
 /**
  * APIs for subscribing to Onos Event Messages.
  */
+@Beta
 public interface EventExporterService {
 
     /**
@@ -44,13 +46,11 @@
      * Allows registered listener to subscribe for a specific event type.
      *
      * @param subscriber Subscription data containing the event type
-     * @throws UnsupportedEventException
      * @throws InvalidGroupIdException
      * @throws InvalidApplicationException
      */
     void subscribe(EventSubscriber subscriber)
-            throws UnsupportedEventException, InvalidGroupIdException,
-            InvalidApplicationException;
+            throws InvalidGroupIdException, InvalidApplicationException;
 
     /**
      * Allows the registered listener to unsubscribe for a specific event.
diff --git a/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/ExportableEventListener.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/ExportableEventListener.java
new file mode 100644
index 0000000..8aae1b7
--- /dev/null
+++ b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/ExportableEventListener.java
@@ -0,0 +1,31 @@
+/**
+ * Copyright 2016 Open Networking Laboratory
+ * 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 org.onosproject.kafkaintegration.api;
+
+import org.onosproject.event.EventListener;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent;
+
+import com.google.common.annotations.Beta;
+
+/**
+ * API for listeners to listen for Events generated by the ONOS event listener.
+ * At present the only listener will be a Kafka Manager or Module whose sole
+ * purpose is to publish the received data to a Kafka message bus.
+ *
+ */
+@Beta
+public interface ExportableEventListener extends EventListener<OnosEvent> {
+
+}
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/dto/EventSubscriber.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/dto/EventSubscriber.java
similarity index 100%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/dto/EventSubscriber.java
rename to apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/dto/EventSubscriber.java
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/dto/EventSubscriberGroupId.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/dto/EventSubscriberGroupId.java
similarity index 100%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/dto/EventSubscriberGroupId.java
rename to apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/dto/EventSubscriberGroupId.java
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/dto/OnosEvent.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/dto/OnosEvent.java
similarity index 100%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/dto/OnosEvent.java
rename to apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/dto/OnosEvent.java
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/dto/package-info.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/dto/package-info.java
similarity index 100%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/dto/package-info.java
rename to apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/dto/package-info.java
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/package-info.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/package-info.java
similarity index 100%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/api/package-info.java
rename to apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/api/package-info.java
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/InvalidApplicationException.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/errors/InvalidApplicationException.java
similarity index 100%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/InvalidApplicationException.java
rename to apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/errors/InvalidApplicationException.java
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/InvalidGroupIdException.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/errors/InvalidGroupIdException.java
similarity index 100%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/InvalidGroupIdException.java
rename to apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/errors/InvalidGroupIdException.java
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/UnsupportedEventException.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/errors/UnsupportedEventException.java
similarity index 100%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/UnsupportedEventException.java
rename to apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/errors/UnsupportedEventException.java
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/package-info.java b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/errors/package-info.java
similarity index 99%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/package-info.java
rename to apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/errors/package-info.java
index 28427d4..35773da 100644
--- a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/package-info.java
+++ b/apps/kafka-integration/api/src/main/java/org/onosproject/kafkaintegration/errors/package-info.java
@@ -14,5 +14,6 @@
 
 /**
  * Application specific Exception classes.
+ *
  */
 package org.onosproject.kafkaintegration.errors;
diff --git a/apps/kafka-integration/app/app.xml b/apps/kafka-integration/app/app.xml
new file mode 100644
index 0000000..5912d4b
--- /dev/null
+++ b/apps/kafka-integration/app/app.xml
@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2016-present Open Networking Laboratory
+  ~
+  ~ 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.
+  -->
+<app name="org.onosproject.kafkaintegration" origin="Calix" version="${project.version}"
+     category="Utility" url="http://onosproject.org" title="Kafka Integration Application"
+     featuresRepo="mvn:${project.groupId}/${project.artifactId}/${project.version}/xml/features"
+     features="${project.artifactId}" apps="org.onosproject.incubator.protobuf">
+    <description>${project.description}</description>
+    <artifact>mvn:${project.groupId}/${project.artifactId}/${project.version}</artifact>
+    <artifact>mvn:${project.groupId}/onos-app-kafka-api/${project.version}</artifact>
+</app>
diff --git a/apps/kafka-integration/app/features.xml b/apps/kafka-integration/app/features.xml
new file mode 100644
index 0000000..1e219b0
--- /dev/null
+++ b/apps/kafka-integration/app/features.xml
@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<!--
+  ~ Copyright 2016-present Open Networking Laboratory
+  ~
+  ~ 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.
+  -->
+<features xmlns="http://karaf.apache.org/xmlns/features/v1.2.0" name="${project.artifactId}-${project.version}">
+    <feature name="${project.artifactId}" version="${project.version}"
+             description="${project.description}">
+        <feature>onos-api</feature>
+        <bundle>mvn:${project.groupId}/onos-app-kafka-api/${project.version}</bundle>
+        <bundle>mvn:${project.groupId}/onos-app-kafka/${project.version}</bundle>
+    </feature>
+</features>
diff --git a/apps/kafka-integration/app/pom.xml b/apps/kafka-integration/app/pom.xml
new file mode 100644
index 0000000..026b410
--- /dev/null
+++ b/apps/kafka-integration/app/pom.xml
@@ -0,0 +1,231 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2016-present Open Networking Laboratory
+  ~
+  ~ 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.onosproject</groupId>
+        <artifactId>onos-kafka</artifactId>
+        <version>1.7.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>onos-app-kafka</artifactId>
+
+    <packaging>bundle</packaging>
+    <description>
+          Kafka Integration Application.
+          This will export ONOS Events to Northbound Kafka Server.
+    </description>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <onos.version>${project.version}</onos.version>
+        <onos.app.name>org.onosproject.kafkaintegration</onos.app.name>
+        <onos.app.title>Kafka Integration Application</onos.app.title>
+        <onos.app.origin>Calix, Inc.</onos.app.origin>
+        <web.context>/onos/kafka</web.context>
+        <api.version>1.0.0</api.version>
+        <api.package>org.onosproject.kafkaintegration.rest</api.package>
+        <api.title>Kafka Integration Application REST API</api.title>
+        <api.description>
+               APIs for subscribing to Events generated by ONOS
+        </api.description>
+        <onos.app.category>Utility</onos.app.category>
+        <onos.app.url>https://wiki.onosproject.org/display/ONOS/Kafka+Integration</onos.app.url>
+        <onos.app.readme>Export ONOS events to a Northbound Kafka server</onos.app.readme>
+        <onos.app.requires>org.onosproject.incubator.protobuf</onos.app.requires>
+    </properties>
+
+       <dependencies>
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-api</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-app-kafka-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-incubator-protobuf</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onlab-osgi</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-rest</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-api</artifactId>
+            <scope>test</scope>
+            <classifier>tests</classifier>
+        </dependency>
+
+        <dependency>
+            <groupId>javax.ws.rs</groupId>
+            <artifactId>javax.ws.rs-api</artifactId>
+            <version>2.0.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+            <version>3.0.0-beta-2</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.codehaus.jackson</groupId>
+            <artifactId>jackson-core-asl</artifactId>
+            <version>1.9.13</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.codehaus.jackson</groupId>
+            <artifactId>jackson-mapper-asl</artifactId>
+            <version>1.9.13</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.glassfish.jersey.containers</groupId>
+            <artifactId>jersey-container-servlet</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-core-serializers</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.felix</groupId>
+            <artifactId>org.apache.felix.scr.annotations</artifactId>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.felix</groupId>
+                <artifactId>maven-bundle-plugin</artifactId>
+                <extensions>true</extensions>
+                <configuration>
+                    <instructions>
+                    <Bundle-SymbolicName>
+                        ${project.groupId}.${project.artifactId}
+                    </Bundle-SymbolicName>
+                    <_wab>src/main/webapp/</_wab>
+                    <Include-Resource>
+                        WEB-INF/classes/apidoc/swagger.json=target/swagger.json,
+                                    {maven-resources}
+                    </Include-Resource>
+                    <Import-Package>
+                        org.slf4j,
+                        org.osgi.framework,
+                        javax.ws.rs,
+                        javax.ws.rs.core,
+                        org.glassfish.jersey.servlet,
+                        com.fasterxml.jackson.databind,
+                        com.fasterxml.jackson.databind.node,
+                        com.fasterxml.jackson.core,
+                        org.onlab.packet.*,
+                        org.onosproject.*,
+                        com.google.common.*
+                    </Import-Package>
+                    <Web-ContextPath>${web.context}</Web-ContextPath>
+                </instructions>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>1.8</source>
+                    <target>1.8</target>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.felix</groupId>
+                <artifactId>maven-scr-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>generate-scr-srcdescriptor</id>
+                        <goals>
+                            <goal>scr</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <supportedProjectTypes>
+                        <supportedProjectType>bundle</supportedProjectType>
+                        <supportedProjectType>war</supportedProjectType>
+                    </supportedProjectTypes>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.onosproject</groupId>
+                <artifactId>onos-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>cfg</id>
+                        <phase>generate-resources</phase>
+                        <goals>
+                            <goal>cfg</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>swagger</id>
+                        <goals>
+                            <goal>swagger</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>app</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>app</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>
diff --git a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/ConversionFactory.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/ConversionFactory.java
new file mode 100644
index 0000000..8f69a20
--- /dev/null
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/ConversionFactory.java
@@ -0,0 +1,68 @@
+/**
+ * Copyright 2016 Open Networking Laboratory
+ * 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 org.onosproject.kafkaintegration.converter;
+
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;;
+
+/**
+ * Returns the appropriate converter object based on the ONOS event type.
+ *
+ */
+public final class ConversionFactory {
+
+    // Store converters for all supported events
+    private Map<Type, EventConverter> converters =
+            new HashMap<Type, EventConverter>() {
+                {
+                    put(DEVICE, new DeviceEventConverter());
+                    put(LINK, new LinkEventConverter());
+                }
+            };
+
+    // Exists to defeat instantiation
+    private ConversionFactory() {
+    }
+
+    private static class SingletonHolder {
+        private static final ConversionFactory INSTANCE =
+                new ConversionFactory();
+    }
+
+    /**
+     * Returns a static reference to the Conversion Factory.
+     *
+     * @return singleton object
+     */
+    public static ConversionFactory getInstance() {
+        return SingletonHolder.INSTANCE;
+    }
+
+    /**
+     * Returns an Event converter object for the specified ONOS event type.
+     *
+     * @param event ONOS event type
+     * @return Event Converter object
+     */
+    public EventConverter getConverter(Type event) {
+        return converters.get(event);
+    }
+
+}
diff --git a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/DeviceEventConverter.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/DeviceEventConverter.java
new file mode 100644
index 0000000..e08d6e1
--- /dev/null
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/DeviceEventConverter.java
@@ -0,0 +1,111 @@
+/**
+ * Copyright 2016 Open Networking Laboratory
+ * 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 org.onosproject.kafkaintegration.converter;
+
+import org.onosproject.event.Event;
+import org.onosproject.grpc.net.Device.DeviceCore;
+import org.onosproject.grpc.net.Device.DeviceType;
+import org.onosproject.grpc.net.DeviceEvent.DeviceEventType;
+import org.onosproject.grpc.net.DeviceEvent.DeviceNotification;
+import org.onosproject.grpc.net.Port.PortCore;
+import org.onosproject.grpc.net.Port.PortType;
+import org.onosproject.net.device.DeviceEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.protobuf.GeneratedMessage;
+
+/**
+ * Converts ONOS Device event message to GPB format.
+ *
+ */
+class DeviceEventConverter implements EventConverter {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public GeneratedMessage convertToProtoMessage(Event<?, ?> event) {
+
+        DeviceEvent deviceEvent = (DeviceEvent) event;
+
+        if (!deviceEventSubtypeSupported(deviceEvent)) {
+            log.error("Unsupported Onos Device Event {}. There is no matching"
+                    + "proto Device Event type", deviceEvent.type().toString());
+            return null;
+        }
+
+        return buildDeviceProtoMessage(deviceEvent);
+    }
+
+    /**
+     * Checks if the ONOS Device Event type is supported.
+     *
+     * @param event ONOS Device event
+     * @return true if there is a match and false otherwise
+     */
+    private boolean deviceEventSubtypeSupported(DeviceEvent event) {
+        DeviceEventType[] deviceEvents = DeviceEventType.values();
+        for (DeviceEventType deviceEventType : deviceEvents) {
+            if (deviceEventType.name().equals(event.type().name())) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    private DeviceNotification buildDeviceProtoMessage(DeviceEvent deviceEvent) {
+        DeviceNotification notification = DeviceNotification.newBuilder()
+                .setDeviceEventType(getProtoType(deviceEvent))
+                .setDevice(DeviceCore.newBuilder()
+                        .setChassisId(deviceEvent.subject().chassisId().id()
+                                .toString())
+                        .setDeviceId(deviceEvent.subject().id().toString())
+                        .setHwVersion(deviceEvent.subject().hwVersion())
+                        .setManufacturer(deviceEvent.subject().manufacturer())
+                        .setSerialNumber(deviceEvent.subject().serialNumber())
+                        .setSwVersion(deviceEvent.subject().swVersion())
+                        .setType(DeviceType.valueOf(deviceEvent.type().name()))
+                        .build())
+                .setPort(PortCore.newBuilder()
+                        .setIsEnabled(deviceEvent.port().isEnabled())
+                        .setPortNumber(deviceEvent.port().number().toString())
+                        .setPortSpeed(deviceEvent.port().portSpeed())
+                        .setType(PortType
+                                .valueOf(deviceEvent.port().type().name()))
+                        .build())
+                .build();
+
+        return notification;
+    }
+
+    /**
+     * Retrieves the protobuf generated device event type.
+     *
+     * @param event ONOS Device Event
+     * @return generated Device Event Type
+     */
+    private DeviceEventType getProtoType(DeviceEvent event) {
+        DeviceEventType protobufEventType = null;
+        DeviceEventType[] deviceEvents = DeviceEventType.values();
+        for (DeviceEventType deviceEventType : deviceEvents) {
+            if (deviceEventType.name().equals(event.type().name())) {
+                protobufEventType = deviceEventType;
+            }
+        }
+
+        return protobufEventType;
+    }
+}
diff --git a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/EventConverter.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/EventConverter.java
new file mode 100644
index 0000000..e15ebdc
--- /dev/null
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/EventConverter.java
@@ -0,0 +1,36 @@
+/**
+ * Copyright 2016 Open Networking Laboratory
+ * 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 org.onosproject.kafkaintegration.converter;
+
+import org.onosproject.event.Event;
+
+import com.google.protobuf.GeneratedMessage;
+
+/**
+ *
+ * APIs for converting between ONOS event objects and GPB data objects.
+ *
+ */
+public interface EventConverter {
+
+    /**
+     * Converts ONOS specific event data to a format that is suitable for export
+     * to Kafka.
+     *
+     * @param event ONOS Event object
+     * @return converted data in GPB format.
+     */
+    public GeneratedMessage convertToProtoMessage(Event<?, ?> event);
+}
diff --git a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java
new file mode 100644
index 0000000..2d65537
--- /dev/null
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java
@@ -0,0 +1,104 @@
+/**
+ * Copyright 2016 Open Networking Laboratory
+ * 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 org.onosproject.kafkaintegration.converter;
+
+import org.onosproject.event.Event;
+import org.onosproject.grpc.net.Link.ConnectPoint;
+import org.onosproject.grpc.net.Link.LinkCore;
+import org.onosproject.grpc.net.Link.LinkState;
+import org.onosproject.grpc.net.Link.LinkType;
+import org.onosproject.grpc.net.LinkEvent.LinkEventType;
+import org.onosproject.grpc.net.LinkEvent.LinkNotification;
+import org.onosproject.net.link.LinkEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.protobuf.GeneratedMessage;
+
+/**
+ * Converts for ONOS Link event message to GPB format.
+ *
+ */
+class LinkEventConverter implements EventConverter {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public GeneratedMessage convertToProtoMessage(Event<?, ?> event) {
+
+        LinkEvent linkEvent = (LinkEvent) event;
+
+        if (!linkEventSubtypeSupported(linkEvent)) {
+            log.error("Unsupported Onos Event {}. There is no matching"
+                    + "proto Event type", linkEvent.type().toString());
+            return null;
+        }
+
+        return buildDeviceProtoMessage(linkEvent);
+    }
+
+    private boolean linkEventSubtypeSupported(LinkEvent event) {
+        LinkType[] kafkaLinkEvents = LinkType.values();
+        for (LinkType linkEventType : kafkaLinkEvents) {
+            if (linkEventType.name().equals(event.type().name())) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    private LinkNotification buildDeviceProtoMessage(LinkEvent linkEvent) {
+        LinkNotification notification = LinkNotification.newBuilder()
+                .setLinkEventType(getProtoType(linkEvent))
+                .setLink(LinkCore.newBuilder()
+                        .setState(LinkState
+                                .valueOf(linkEvent.subject().state().name()))
+                        .setType(LinkType
+                                .valueOf(linkEvent.subject().type().name()))
+                        .setDst(ConnectPoint.newBuilder()
+                                .setDeviceId(linkEvent.subject().dst()
+                                        .deviceId().toString())
+                                .setPortNumber(linkEvent.subject().dst().port()
+                                        .toString()))
+                        .setSrc(ConnectPoint.newBuilder()
+                                .setDeviceId(linkEvent.subject().src()
+                                        .deviceId().toString())
+                                .setPortNumber(linkEvent.subject().src().port()
+                                        .toString())))
+                .build();
+
+        return notification;
+    }
+
+    /**
+     * Returns the specific Kafka Device Event Type for the corresponding ONOS
+     * Device Event Type.
+     *
+     * @param event ONOS Device Event
+     * @return Kafka Device Event Type
+     */
+    private LinkEventType getProtoType(LinkEvent event) {
+        LinkEventType generatedEventType = null;
+        LinkEventType[] kafkaEvents = LinkEventType.values();
+        for (LinkEventType linkEventType : kafkaEvents) {
+            if (linkEventType.name().equals(event.type().name())) {
+                generatedEventType = linkEventType;
+            }
+        }
+
+        return generatedEventType;
+    }
+}
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/package-info.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/package-info.java
similarity index 79%
copy from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/package-info.java
copy to apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/package-info.java
index 28427d4..d33c0d7 100644
--- a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/package-info.java
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/converter/package-info.java
@@ -13,6 +13,8 @@
  */
 
 /**
- * Application specific Exception classes.
+ * Converters for converting various ONOS events to their corresponding Protocol
+ * Buffer format.
+ *
  */
-package org.onosproject.kafkaintegration.errors;
+package org.onosproject.kafkaintegration.converter;
diff --git a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/impl/Dispatcher.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/impl/Dispatcher.java
new file mode 100644
index 0000000..d211458
--- /dev/null
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/impl/Dispatcher.java
@@ -0,0 +1,62 @@
+/**
+ * Copyright 2016 Open Networking Laboratory
+ * 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 org.onosproject.kafkaintegration.impl;
+
+import org.onosproject.event.AbstractListenerManager;
+import org.onosproject.kafkaintegration.api.ExportableEventListener;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.protobuf.GeneratedMessage;
+
+/**
+ * Dispatch ONOS Events to all interested Listeners.
+ *
+ */
+public final class Dispatcher
+        extends AbstractListenerManager<OnosEvent, ExportableEventListener> {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    // Exists to defeat instantiation
+    private Dispatcher() {
+    }
+
+    private static class SingletonHolder {
+        private static final Dispatcher INSTANCE = new Dispatcher();
+    }
+
+    /**
+     * Returns a static reference to the Listener Factory.
+     *
+     * @return singleton object
+     */
+    public static Dispatcher getInstance() {
+        return SingletonHolder.INSTANCE;
+    }
+
+    /**
+     * Publish the ONOS Event to all listeners.
+     *
+     * @param eventType the ONOS eventtype
+     * @param message generated Protocol buffer message from ONOS event data
+     */
+    public void publish(Type eventType, GeneratedMessage message) {
+        log.debug("Dispatching ONOS Event {}", eventType);
+        post(new OnosEvent(eventType, message));
+    }
+}
diff --git a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/impl/EventExporterManager.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/impl/EventExporterManager.java
new file mode 100644
index 0000000..6e1a0e8
--- /dev/null
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/impl/EventExporterManager.java
@@ -0,0 +1,323 @@
+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * 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 org.onosproject.kafkaintegration.impl;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreService;
+import org.onosproject.kafkaintegration.api.EventExporterService;
+import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
+import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+import org.onosproject.kafkaintegration.errors.InvalidApplicationException;
+import org.onosproject.kafkaintegration.errors.InvalidGroupIdException;
+import org.onosproject.kafkaintegration.listener.ListenerFactory;
+import org.onosproject.kafkaintegration.listener.OnosEventListener;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.link.LinkService;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Implementation of Event Exporter Service.
+ *
+ */
+@Component(immediate = true)
+@Service
+public class EventExporterManager implements EventExporterService {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    // Stores the currently registered applications for event export service.
+    private Map<ApplicationId, EventSubscriberGroupId> registeredApps;
+
+    private Map<Type, List<EventSubscriber>> subscriptions;
+
+    private static final String REGISTERED_APPS = "registered-applications";
+
+    private static final String SUBSCRIBED_APPS = "event-subscriptions";
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected DeviceService deviceService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected LinkService linkService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected CoreService coreService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected StorageService storageService;
+
+    private ApplicationId appId;
+
+    @Activate
+    protected void activate() {
+        appId = coreService
+                .registerApplication("org.onosproject.kafkaintegration");
+
+        registeredApps = storageService
+                .<ApplicationId, EventSubscriberGroupId>consistentMapBuilder()
+                .withName(REGISTERED_APPS)
+                .withSerializer(Serializer.using(KryoNamespaces.API,
+                                                 EventSubscriberGroupId.class,
+                                                 UUID.class))
+                .build().asJavaMap();
+
+        subscriptions = storageService
+                .<Type, List<EventSubscriber>>consistentMapBuilder()
+                .withName(SUBSCRIBED_APPS)
+                .withSerializer(Serializer.using(KryoNamespaces.API,
+                                                 EventSubscriber.class))
+                .build().asJavaMap();
+
+        log.info("Started");
+    }
+
+    @Deactivate
+    protected void deactivate() {
+        log.info("Stopped");
+    }
+
+    @Override
+    public EventSubscriberGroupId registerListener(String appName) {
+
+        // TODO: Remove it once ONOS provides a mechanism for external apps
+        // to register with the core service. See Jira - 4409
+        ApplicationId externalAppId = coreService.registerApplication(appName);
+
+        return registeredApps.computeIfAbsent(externalAppId,
+                                              (key) -> new EventSubscriberGroupId(UUID
+                                                      .randomUUID()));
+
+    }
+
+    @Override
+    public void unregisterListener(String appName) {
+        ApplicationId externalAppId =
+                checkNotNull(coreService.getAppId(appName));
+        registeredApps.remove(externalAppId);
+    }
+
+    @Override
+    public void subscribe(EventSubscriber subscriber)
+            throws InvalidGroupIdException, InvalidApplicationException {
+
+        checkNotNull(subscriber);
+
+        if (!registeredApplication(subscriber.appName())) {
+            throw new InvalidApplicationException("Application is not "
+                    + "registered to make this request.");
+        }
+
+        if (!validGroupId(subscriber.subscriberGroupId(),
+                          subscriber.appName())) {
+            throw new InvalidGroupIdException("Incorrect group id in the request");
+        }
+
+        OnosEventListener onosListener = getListener(subscriber.eventType());
+        checkNotNull(onosListener,
+                     "No listener for the supported event type - {}",
+                     subscriber.eventType());
+
+        applyListenerAction(subscriber.eventType(), onosListener,
+                            ListenerAction.START);
+
+        // update internal state
+        List<EventSubscriber> subscriptionList =
+                subscriptions.get(subscriber.eventType());
+        if (subscriptionList == null) {
+            subscriptionList = new ArrayList<EventSubscriber>();
+        }
+        subscriptionList.add(subscriber);
+        subscriptions.put(subscriber.eventType(), subscriptionList);
+
+        log.info("Subscription for {} event by {} successfull",
+                 subscriber.eventType(), subscriber.appName());
+    }
+
+    /**
+     * Checks if the application has registered.
+     *
+     * @param appName application name
+     * @return true if application has registered
+     */
+    private boolean registeredApplication(String appName) {
+
+        checkNotNull(appName);
+        ApplicationId appId = checkNotNull(coreService.getAppId(appName));
+        if (registeredApps.containsKey(appId)) {
+            return true;
+        }
+
+        log.debug("{} is not registered", appName);
+        return false;
+    }
+
+    /**
+     * Actions that can be performed on the ONOS Event Listeners.
+     *
+     */
+    private enum ListenerAction {
+        START, STOP;
+    }
+
+    /**
+     * Applies the specified action on the Listener.
+     *
+     * @param eventType the ONOS Event type registered by the application
+     * @param onosListener ONOS event listener
+     * @param action to be performed on the listener
+     */
+    private void applyListenerAction(Type eventType,
+                                     OnosEventListener onosListener,
+                                     ListenerAction action) {
+        switch (eventType) {
+        case DEVICE:
+            if (action == ListenerAction.START) {
+                onosListener.startListener(DEVICE, deviceService);
+            } else {
+                onosListener.stopListener(DEVICE, deviceService);
+            }
+            break;
+        case LINK:
+            if (action == ListenerAction.START) {
+                onosListener.startListener(LINK, linkService);
+            } else {
+                onosListener.stopListener(LINK, linkService);
+            }
+            break;
+        default:
+            log.error("Cannot {} listener. Unsupported event type {} ",
+                      action.toString(), eventType.toString());
+        }
+    }
+
+    /**
+     * Returns the ONOS event listener corresponding to the ONOS Event type.
+     *
+     * @param eventType ONOS event type
+     * @return ONOS event listener
+     */
+    private OnosEventListener getListener(Type eventType) {
+        checkNotNull(eventType);
+        ListenerFactory factory = ListenerFactory.getInstance();
+        OnosEventListener onosListener = factory.getListener(eventType);
+        return onosListener;
+    }
+
+    /**
+     * Checks if the group id is valid for this registered application.
+     *
+     * @param groupId GroupId assigned to the subscriber
+     * @param appName Registered Application name
+     * @return true if valid groupId and false otherwise
+     */
+    private boolean validGroupId(EventSubscriberGroupId groupId,
+                                 String appName) {
+
+        checkNotNull(groupId);
+
+        ApplicationId appId = coreService.getAppId(appName);
+        EventSubscriberGroupId registeredGroupId = registeredApps.get(appId);
+        if (registeredGroupId.equals(groupId)) {
+            return true;
+        }
+
+        return false;
+    }
+
+    @Override
+    public void unsubscribe(EventSubscriber subscriber)
+            throws InvalidGroupIdException, InvalidApplicationException {
+
+        checkNotNull(subscriber);
+
+        if (!registeredApplication(subscriber.appName())) {
+            throw new InvalidApplicationException("Application is not "
+                    + "registered to make this request.");
+        }
+
+        if (!validGroupId(subscriber.subscriberGroupId(),
+                          subscriber.appName())) {
+            throw new InvalidGroupIdException("Incorrect group id in the request");
+        }
+
+        if (!eventSubscribed(subscriber)) {
+            log.error("No subscription to {} was found",
+                      subscriber.eventType());
+            return;
+        }
+
+        // If this is the only subscriber listening for this event,
+        // stop the listener.
+        List<EventSubscriber> subscribers =
+                subscriptions.get(subscriber.eventType());
+        if (subscribers.size() == 1) {
+            OnosEventListener onosListener =
+                    getListener(subscriber.eventType());
+            checkNotNull(onosListener,
+                         "No listener for the supported event type - {}",
+                         subscriber.eventType());
+            applyListenerAction(subscriber.eventType(), onosListener,
+                                ListenerAction.STOP);
+        }
+
+        // update internal state.
+        subscribers.remove(subscriber);
+        subscriptions.put(subscriber.eventType(), subscribers);
+
+        log.info("Unsubscribed {} for {} events", subscriber.appName(),
+                 subscriber.eventType());
+    }
+
+    /**
+     * Checks if the subscriber has already subscribed to the requested event
+     * type.
+     *
+     * @param subscriber the subscriber to a specific ONOS event
+     * @return true if subscriber has subscribed to the ONOS event
+     */
+    private boolean eventSubscribed(EventSubscriber subscriber) {
+
+        List<EventSubscriber> subscriberList =
+                subscriptions.get(subscriber.eventType());
+
+        if (subscriberList == null) {
+            return false;
+        }
+
+        return subscriberList.contains(subscriber);
+    }
+
+}
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/package-info.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/impl/package-info.java
similarity index 92%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/package-info.java
rename to apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/impl/package-info.java
index 72f4843..a2950fa 100644
--- a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/package-info.java
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/impl/package-info.java
@@ -15,4 +15,4 @@
 /**
  * API implementation classes.
  */
-package org.onosproject.kafkaintegration;
+package org.onosproject.kafkaintegration.impl;
diff --git a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/DeviceEventsListener.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/DeviceEventsListener.java
new file mode 100644
index 0000000..22c4bf2
--- /dev/null
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/DeviceEventsListener.java
@@ -0,0 +1,94 @@
+/**
+ * Copyright 2016 Open Networking Laboratory
+ * 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 org.onosproject.kafkaintegration.listener;
+
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+
+import org.onosproject.event.ListenerService;
+import org.onosproject.kafkaintegration.impl.Dispatcher;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+import org.onosproject.kafkaintegration.converter.ConversionFactory;
+import org.onosproject.kafkaintegration.converter.EventConverter;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.net.device.DeviceListener;
+import org.onosproject.net.device.DeviceService;
+
+import com.google.protobuf.GeneratedMessage;
+
+/**
+ * Listens for ONOS Device events.
+ *
+ */
+final class DeviceEventsListener implements OnosEventListener {
+
+    private boolean listenerRunning = false;
+
+    private InnerListener listener = null;
+
+    // Exists to defeat instantiation
+    private DeviceEventsListener() {
+    }
+
+    private static class SingletonHolder {
+        private static final DeviceEventsListener INSTANCE =
+                new DeviceEventsListener();
+    }
+
+    /**
+     * Returns a static reference to the Listener Factory.
+     *
+     * @return singleton object
+     */
+    public static DeviceEventsListener getInstance() {
+        return SingletonHolder.INSTANCE;
+    }
+
+    @Override
+    public void startListener(Type e, ListenerService<?, ?> service) {
+        if (!listenerRunning) {
+            listener = new InnerListener();
+            DeviceService deviceService = (DeviceService) service;
+            deviceService.addListener(listener);
+            listenerRunning = true;
+        }
+    }
+
+    private class InnerListener implements DeviceListener {
+
+        @Override
+        public void event(DeviceEvent arg0) {
+
+            // Convert the event to GPB format
+            ConversionFactory conversionFactory =
+                    ConversionFactory.getInstance();
+            EventConverter converter = conversionFactory.getConverter(DEVICE);
+            GeneratedMessage message = converter.convertToProtoMessage(arg0);
+
+            // Call Dispatcher and publish event
+            Dispatcher.getInstance().publish(DEVICE, message);
+        }
+    }
+
+    @Override
+    public void stopListener(Type e, ListenerService<?, ?> service) {
+        if (listenerRunning) {
+            DeviceService deviceService = (DeviceService) service;
+            deviceService.removeListener(listener);
+            listener = null;
+            listenerRunning = false;
+        }
+    }
+
+}
diff --git a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/LinkEventsListener.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/LinkEventsListener.java
new file mode 100644
index 0000000..769d8b4
--- /dev/null
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/LinkEventsListener.java
@@ -0,0 +1,92 @@
+/**
+ * Copyright 2016 Open Networking Laboratory
+ * 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 org.onosproject.kafkaintegration.listener;
+
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+
+import org.onosproject.event.ListenerService;
+import org.onosproject.kafkaintegration.impl.Dispatcher;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+import org.onosproject.kafkaintegration.converter.ConversionFactory;
+import org.onosproject.kafkaintegration.converter.EventConverter;
+import org.onosproject.net.link.LinkEvent;
+import org.onosproject.net.link.LinkListener;
+import org.onosproject.net.link.LinkService;
+
+import com.google.protobuf.GeneratedMessage;
+
+/**
+ * Listens for ONOS Link Events.
+ *
+ */
+final class LinkEventsListener implements OnosEventListener {
+
+    private boolean listenerRunning = false;
+
+    private InnerListener listener = null;
+
+    // Exists to defeat instantiation
+    private LinkEventsListener() {
+    }
+
+    private static class SingletonHolder {
+        private static final LinkEventsListener INSTANCE =
+                new LinkEventsListener();
+    }
+
+    /**
+     * Returns a static reference to the Listener Factory.
+     *
+     * @return singleton object
+     */
+    public static LinkEventsListener getInstance() {
+        return SingletonHolder.INSTANCE;
+    }
+
+    @Override
+    public void startListener(Type e, ListenerService<?, ?> service) {
+        if (!listenerRunning) {
+            listener = new InnerListener();
+            LinkService linkService = (LinkService) service;
+            linkService.addListener(listener);
+            listenerRunning = true;
+        }
+    }
+
+    private class InnerListener implements LinkListener {
+
+        @Override
+        public void event(LinkEvent arg0) {
+
+            // Convert the event to GPB format
+            ConversionFactory conversionFactory =
+                    ConversionFactory.getInstance();
+            EventConverter converter = conversionFactory.getConverter(LINK);
+            GeneratedMessage message = converter.convertToProtoMessage(arg0);
+
+            // Call Dispatcher and publish event
+            Dispatcher.getInstance().publish(LINK, message);
+        }
+    }
+
+    @Override
+    public void stopListener(Type e, ListenerService<?, ?> service) {
+        if (listenerRunning) {
+            LinkService linkService = (LinkService) service;
+            linkService.removeListener(listener);
+            listenerRunning = false;
+        }
+    }
+}
diff --git a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/ListenerFactory.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/ListenerFactory.java
new file mode 100644
index 0000000..496ba2b
--- /dev/null
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/ListenerFactory.java
@@ -0,0 +1,67 @@
+/**
+ * Copyright 2016 Open Networking Laboratory
+ * 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 org.onosproject.kafkaintegration.listener;
+
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+
+/**
+ * Returns the appropriate listener object based on the ONOS event type.
+ *
+ */
+public final class ListenerFactory {
+
+    // Store listeners for all supported events
+    private Map<Type, OnosEventListener> listeners =
+            new HashMap<Type, OnosEventListener>() {
+                {
+                    put(DEVICE, DeviceEventsListener.getInstance());
+                    put(LINK, LinkEventsListener.getInstance());
+                }
+            };
+
+    // Exists to defeat instantiation
+    private ListenerFactory() {
+    }
+
+    private static class SingletonHolder {
+        private static final ListenerFactory INSTANCE = new ListenerFactory();
+    }
+
+    /**
+     * Returns a static reference to the Listener Factory.
+     *
+     * @return singleton object
+     */
+    public static ListenerFactory getInstance() {
+        return SingletonHolder.INSTANCE;
+    }
+
+    /**
+     * Returns the listener object for the specified ONOS event type.
+     *
+     * @param event ONOS Event type
+     * @return return listener object
+     */
+    public OnosEventListener getListener(Type event) {
+        return listeners.get(event);
+    }
+
+}
diff --git a/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/OnosEventListener.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/OnosEventListener.java
new file mode 100644
index 0000000..22cde21
--- /dev/null
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/OnosEventListener.java
@@ -0,0 +1,42 @@
+/**
+ * Copyright 2016 Open Networking Laboratory
+ * 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 org.onosproject.kafkaintegration.listener;
+
+import org.onosproject.event.ListenerService;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+
+/**
+ * APIs for starting and stopping a ONOS Event listener.
+ *
+ */
+public interface OnosEventListener {
+
+    /**
+     * Start the listener for the specific ONOS event type.
+     *
+     * @param event ONOS event type
+     * @param service ONOS event listener for the specific event type
+     */
+    void startListener(Type event, ListenerService<?, ?> service);
+
+    /**
+     * Stop the Listener for the specific ONOS event type.
+     *
+     * @param event ONOS event type
+     * @param service ONOS event listener for the specific event type
+     */
+    void stopListener(Type event, ListenerService<?, ?> service);
+
+}
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/package-info.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/package-info.java
similarity index 85%
copy from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/package-info.java
copy to apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/package-info.java
index 28427d4..f6f602f 100644
--- a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/errors/package-info.java
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/listener/package-info.java
@@ -13,6 +13,7 @@
  */
 
 /**
- * Application specific Exception classes.
+ * Listeners for listening to various ONOS events.
+ *
  */
-package org.onosproject.kafkaintegration.errors;
+package org.onosproject.kafkaintegration.listener;
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/rest/EventExporterWebResource.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/rest/EventExporterWebResource.java
similarity index 82%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/rest/EventExporterWebResource.java
rename to apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/rest/EventExporterWebResource.java
index 9677265..520976a 100644
--- a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/rest/EventExporterWebResource.java
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/rest/EventExporterWebResource.java
@@ -14,15 +14,11 @@
  */
 package org.onosproject.kafkaintegration.rest;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import org.onosproject.codec.JsonCodec;
-import org.onosproject.kafkaintegration.api.EventExporterService;
-import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
-import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
-import org.onosproject.rest.AbstractWebResource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static javax.ws.rs.core.Response.Status.BAD_REQUEST;
+
+import java.io.IOException;
+import java.io.InputStream;
 
 import javax.ws.rs.Consumes;
 import javax.ws.rs.DELETE;
@@ -31,11 +27,17 @@
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
-import java.io.IOException;
-import java.io.InputStream;
 
-import static com.google.common.base.Preconditions.checkNotNull;
-import static javax.ws.rs.core.Response.Status.BAD_REQUEST;
+import org.onosproject.codec.JsonCodec;
+import org.onosproject.kafkaintegration.api.EventExporterService;
+import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
+import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
+import org.onosproject.rest.AbstractWebResource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 
 /**
  * Rest Interfaces for subscribing/unsubscribing to event notifications.
@@ -44,14 +46,19 @@
 public class EventExporterWebResource extends AbstractWebResource {
 
     private final Logger log = LoggerFactory.getLogger(getClass());
-    public static final String JSON_NOT_NULL = "Registration Data cannot be empty";
-    public static final String REGISTRATION_SUCCESSFUL = "Registered Listener successfully";
-    public static final String DEREGISTRATION_SUCCESSFUL = "De-Registered Listener successfully";
-    public static final String EVENT_SUBSCRIPTION_SUCCESSFUL = "Event Registration successfull";
-    public static final String EVENT_SUBSCRIPTION_REMOVED = "Event De-Registration successfull";
+    public static final String JSON_NOT_NULL =
+            "Registration Data cannot be empty";
+    public static final String REGISTRATION_SUCCESSFUL =
+            "Registered Listener successfully";
+    public static final String DEREGISTRATION_SUCCESSFUL =
+            "De-Registered Listener successfully";
+    public static final String EVENT_SUBSCRIPTION_SUCCESSFUL =
+            "Event Registration successfull";
+    public static final String EVENT_SUBSCRIPTION_REMOVED =
+            "Event De-Registration successfull";
 
     /**
-     * Registers a listener for Onos Events.
+     * Registers a listener for ONOS Events.
      *
      * @param appName The application trying to register
      * @return 200 OK with UUID string which should be used as Kafka Consumer
@@ -76,7 +83,7 @@
     }
 
     /**
-     * Unregisters a listener for Onos Events.
+     * Unregisters a listener for ONOS Events.
      *
      * @param appName The application trying to unregister
      * @return 200 OK
@@ -93,9 +100,9 @@
     }
 
     /**
-     * Creates subscription to a specific Onos event.
+     * Creates subscription to a specific ONOS event.
      *
-     * @param input Subscription Data in Json format
+     * @param input Subscription Data in JSON format
      * @return 200 OK if successful or 400 BAD REQUEST
      * @onos.rsModel KafkaSubscription
      */
@@ -118,9 +125,9 @@
     }
 
     /**
-     * Parses Json Subscription Data from the external application.
+     * Parses JSON Subscription Data from the external application.
      *
-     * @param node node within the parsed json tree.
+     * @param input Subscription Data in JSON format
      * @return parsed DTO object
      * @throws IOException
      */
@@ -137,9 +144,9 @@
     }
 
     /**
-     * Deletes subscription from a specific Onos event.
+     * Deletes subscription from a specific ONOS event.
      *
-     * @param input data in json format
+     * @param input data in JSON format
      * @return 200 OK if successful or 400 BAD REQUEST
      * @onos.rsModel KafkaSubscription
      */
@@ -152,7 +159,7 @@
 
         try {
             EventSubscriber sub = parseSubscriptionData(input);
-            service.subscribe(sub);
+            service.unsubscribe(sub);
         } catch (Exception e) {
             log.error(e.getMessage());
             return Response.status(BAD_REQUEST).entity(e.getMessage()).build();
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/rest/SubscriberCodec.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/rest/SubscriberCodec.java
similarity index 100%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/rest/SubscriberCodec.java
rename to apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/rest/SubscriberCodec.java
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/rest/package-info.java b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/rest/package-info.java
similarity index 99%
rename from apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/rest/package-info.java
rename to apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/rest/package-info.java
index 424c4a5..7885851 100644
--- a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/rest/package-info.java
+++ b/apps/kafka-integration/app/src/main/java/org/onosproject/kafkaintegration/rest/package-info.java
@@ -14,5 +14,6 @@
 
 /**
  * REST API Definitions.
+ *
  */
 package org.onosproject.kafkaintegration.rest;
diff --git a/apps/kafka-integration/src/main/resources/definitions/KafkaRegistration.json b/apps/kafka-integration/app/src/main/resources/definitions/KafkaRegistration.json
similarity index 100%
rename from apps/kafka-integration/src/main/resources/definitions/KafkaRegistration.json
rename to apps/kafka-integration/app/src/main/resources/definitions/KafkaRegistration.json
diff --git a/apps/kafka-integration/src/main/resources/definitions/KafkaSubscription.json b/apps/kafka-integration/app/src/main/resources/definitions/KafkaSubscription.json
similarity index 100%
rename from apps/kafka-integration/src/main/resources/definitions/KafkaSubscription.json
rename to apps/kafka-integration/app/src/main/resources/definitions/KafkaSubscription.json
diff --git a/apps/kafka-integration/src/main/webapp/WEB-INF/web.xml b/apps/kafka-integration/app/src/main/webapp/WEB-INF/web.xml
similarity index 100%
rename from apps/kafka-integration/src/main/webapp/WEB-INF/web.xml
rename to apps/kafka-integration/app/src/main/webapp/WEB-INF/web.xml
diff --git a/apps/kafka-integration/pom.xml b/apps/kafka-integration/pom.xml
index 72f1c77..6f806a7 100644
--- a/apps/kafka-integration/pom.xml
+++ b/apps/kafka-integration/pom.xml
@@ -1,6 +1,6 @@
-<?sxml version="1.0" encoding="UTF-8"?>
+<?xml version="1.0" encoding="UTF-8"?>
 <!--
-  ~ Copyright 2016 Open Networking Laboratory
+  ~ Copyright 2016-present Open Networking Laboratory
   ~
   ~ Licensed under the Apache License, Version 2.0 (the "License");
   ~ you may not use this file except in compliance with the License.
@@ -14,218 +14,26 @@
   ~ See the License for the specific language governing permissions and
   ~ limitations under the License.
   -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <modelVersion>4.0.0</modelVersion>
-
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <parent>
         <groupId>org.onosproject</groupId>
         <artifactId>onos-apps</artifactId>
         <version>1.7.0-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
+    <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>onos-app-kafka</artifactId>
-    <packaging>bundle</packaging>
+    <artifactId>onos-kafka</artifactId>
 
-    <description>
-        ONOS Kafka Integration Application.
-        This will export ONOS events to an external Kafka Server
-    </description>
-    <url>http://onosproject.org</url>
+    <packaging>pom</packaging>
+    <description>Kafka Integration Application</description>
 
-    <properties>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <onos.version>1.6.0-SNAPSHOT</onos.version>
-        <onos.app.name>org.onosproject.kafkaintegration</onos.app.name>
-        <onos.app.title>Kafka Integration Application</onos.app.title>
-        <onos.app.origin>Calix, Inc.</onos.app.origin>
-        <web.context>/onos/kafka</web.context>
-        <api.version>1.0.0</api.version>
-        <api.package>org.onosproject.kafkaintegration.rest</api.package>
-        <api.title>Kafka Integration Application REST API</api.title>
-        <api.description>
-               APIs for subscribing to Events generated by onos
-        </api.description>
-        <onos.app.category>Utility</onos.app.category>
-        <onos.app.url>https://wiki.onosproject.org/display/ONOS/Kafka+Integration</onos.app.url>
-        <onos.app.readme>Export onos events to a Northbound Kafka server</onos.app.readme>
-    </properties>
-
-    <dependencies>
-        <dependency>
-            <groupId>org.onosproject</groupId>
-            <artifactId>onos-api</artifactId>
-            <version>${onos.version}</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.onosproject</groupId>
-            <artifactId>onlab-osgi</artifactId>
-            <version>${onos.version}</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.onosproject</groupId>
-            <artifactId>onos-rest</artifactId>
-            <version>${onos.version}</version>
-        </dependency>
-
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <version>4.12</version>
-            <scope>test</scope>
-        </dependency>
-
-        <dependency>
-            <groupId>org.onosproject</groupId>
-            <artifactId>onos-api</artifactId>
-            <version>${onos.version}</version>
-            <scope>test</scope>
-            <classifier>tests</classifier>
-        </dependency>
-
-        <dependency>
-            <groupId>javax.ws.rs</groupId>
-            <artifactId>javax.ws.rs-api</artifactId>
-            <version>2.0.1</version>
-        </dependency>
-
-        <dependency>
-            <groupId>com.google.protobuf</groupId>
-            <artifactId>protobuf-java</artifactId>
-            <version>2.5.0</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.codehaus.jackson</groupId>
-            <artifactId>jackson-core-asl</artifactId>
-            <version>1.9.13</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.codehaus.jackson</groupId>
-            <artifactId>jackson-mapper-asl</artifactId>
-            <version>1.9.13</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.glassfish.jersey.containers</groupId>
-            <artifactId>jersey-container-servlet</artifactId>
-            <version>2.22.2</version>
-        </dependency>
-
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-annotations</artifactId>
-            <version>2.6.4</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.onosproject</groupId>
-            <artifactId>onos-core-serializers</artifactId>
-            <version>${onos.version}</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.felix</groupId>
-            <artifactId>org.apache.felix.scr.annotations</artifactId>
-            <version>1.9.12</version>
-            <scope>provided</scope>
-        </dependency>
-    </dependencies>
-
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.felix</groupId>
-                <artifactId>maven-bundle-plugin</artifactId>
-                <version>3.0.1</version>
-                <extensions>true</extensions>
-                <configuration>
-                    <instructions>
-                    <Bundle-SymbolicName>
-                        ${project.groupId}.${project.artifactId}
-                    </Bundle-SymbolicName>
-                    <_wab>src/main/webapp/</_wab>
-                    <Include-Resource>
-                        WEB-INF/classes/apidoc/swagger.json=target/swagger.json,
-                                    {maven-resources}
-                    </Include-Resource>
-                    <Import-Package>
-                        org.slf4j,
-                        org.osgi.framework,
-                        javax.ws.rs,
-                        javax.ws.rs.core,
-                        org.glassfish.jersey.servlet,
-                        com.fasterxml.jackson.databind,
-                        com.fasterxml.jackson.databind.node,
-                        com.fasterxml.jackson.core,
-                        org.onlab.packet.*,
-                        org.onosproject.*,
-                        com.google.common.*
-                    </Import-Package>
-                    <Web-ContextPath>${web.context}</Web-ContextPath>
-                </instructions>
-                </configuration>
-            </plugin>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-compiler-plugin</artifactId>
-                <version>2.5.1</version>
-                <configuration>
-                    <source>1.8</source>
-                    <target>1.8</target>
-                </configuration>
-            </plugin>
-            <plugin>
-                <groupId>org.apache.felix</groupId>
-                <artifactId>maven-scr-plugin</artifactId>
-                <version>1.21.0</version>
-                <executions>
-                    <execution>
-                        <id>generate-scr-srcdescriptor</id>
-                        <goals>
-                            <goal>scr</goal>
-                        </goals>
-                    </execution>
-                </executions>
-                <configuration>
-                    <supportedProjectTypes>
-                        <supportedProjectType>bundle</supportedProjectType>
-                        <supportedProjectType>war</supportedProjectType>
-                    </supportedProjectTypes>
-                </configuration>
-            </plugin>
-            <plugin>
-                <groupId>org.onosproject</groupId>
-                <artifactId>onos-maven-plugin</artifactId>
-                <version>1.9</version>
-                <executions>
-                    <execution>
-                        <id>cfg</id>
-                        <phase>generate-resources</phase>
-                        <goals>
-                            <goal>cfg</goal>
-                        </goals>
-                    </execution>
-                    <execution>
-                        <id>swagger</id>
-                        <phase>generate-sources</phase>
-                        <goals>
-                            <goal>swagger</goal>
-                        </goals>
-                    </execution>
-                    <execution>
-                        <id>app</id>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>app</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
-        </plugins>
-    </build>
+    <modules>
+        <module>api</module>
+        <module>app</module>
+    </modules>
 
 </project>
+
diff --git a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/EventExporterManager.java b/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/EventExporterManager.java
deleted file mode 100644
index 53d9173..0000000
--- a/apps/kafka-integration/src/main/java/org/onosproject/kafkaintegration/EventExporterManager.java
+++ /dev/null
@@ -1,132 +0,0 @@
-/*
- * Copyright 2016-present Open Networking Laboratory
- *
- * 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 org.onosproject.kafkaintegration;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-
-import java.util.Map;
-import java.util.UUID;
-
-import org.apache.felix.scr.annotations.Activate;
-import org.apache.felix.scr.annotations.Component;
-import org.apache.felix.scr.annotations.Deactivate;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.ReferenceCardinality;
-import org.apache.felix.scr.annotations.Service;
-import org.onosproject.core.ApplicationId;
-import org.onosproject.core.CoreService;
-import org.onosproject.kafkaintegration.api.EventExporterService;
-import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
-import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
-import org.onosproject.kafkaintegration.errors.InvalidApplicationException;
-import org.onosproject.kafkaintegration.errors.InvalidGroupIdException;
-import org.onosproject.kafkaintegration.errors.UnsupportedEventException;
-import org.onosproject.net.device.DeviceService;
-import org.onosproject.net.link.LinkService;
-import org.onosproject.store.serializers.KryoNamespaces;
-import org.onosproject.store.service.Serializer;
-import org.onosproject.store.service.StorageService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Implementation of Event Exporter Service.
- *
- */
-@Component(immediate = true)
-@Service
-public class EventExporterManager implements EventExporterService {
-
-    private final Logger log = LoggerFactory.getLogger(getClass());
-
-    // Stores the currently registered applications for event export service.
-    // Map of Appname to groupId
-    private Map<ApplicationId, EventSubscriberGroupId> registeredApps;
-
-    private static final String REGISTERED_APPS = "registered-applications";
-
-    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
-    protected DeviceService deviceService;
-
-    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
-    protected LinkService linkService;
-
-    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
-    protected CoreService coreService;
-
-    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
-    protected StorageService storageService;
-
-    private static final String NOT_YET_SUPPORTED = "Not yet supported.";
-
-    private ApplicationId appId;
-
-    @Activate
-    protected void activate() {
-        appId = coreService
-                .registerApplication("org.onosproject.kafkaintegration");
-
-        registeredApps = storageService
-                .<ApplicationId, EventSubscriberGroupId>consistentMapBuilder()
-                .withName(REGISTERED_APPS)
-                .withSerializer(Serializer.using(KryoNamespaces.API,
-                                                 EventSubscriberGroupId.class,
-                                                 UUID.class))
-                .build().asJavaMap();
-
-        log.info("Started");
-    }
-
-    @Deactivate
-    protected void deactivate() {
-        log.info("Stopped");
-    }
-
-    @Override
-    public EventSubscriberGroupId registerListener(String appName) {
-
-        // TODO: Remove it once ONOS provides a mechanism for external apps
-        // to register with the core service. See Jira - 4409
-        ApplicationId externalAppId = coreService.registerApplication(appName);
-
-        return registeredApps.computeIfAbsent(externalAppId,
-                                              (key) -> new EventSubscriberGroupId(UUID
-                                                      .randomUUID()));
-
-    }
-
-    @Override
-    public void unregisterListener(String appName) {
-        ApplicationId externalAppId =
-                checkNotNull(coreService.getAppId(appName));
-        registeredApps.remove(externalAppId);
-    }
-
-    @Override
-    public void subscribe(EventSubscriber subscriber)
-            throws UnsupportedEventException, InvalidGroupIdException,
-            InvalidApplicationException {
-
-        throw new UnsupportedOperationException(NOT_YET_SUPPORTED);
-    }
-
-    @Override
-    public void unsubscribe(EventSubscriber subscriber)
-            throws InvalidGroupIdException, InvalidApplicationException {
-
-        throw new UnsupportedOperationException(NOT_YET_SUPPORTED);
-    }
-}
diff --git a/incubator/protobuf/src/main/proto/Device.proto b/incubator/protobuf/src/main/proto/Device.proto
index ec138e0..a01790e 100644
--- a/incubator/protobuf/src/main/proto/Device.proto
+++ b/incubator/protobuf/src/main/proto/Device.proto
@@ -3,6 +3,23 @@
 
 package Device;
 
+message DeviceDescription {
+  string device_Uri = 1;
+  DeviceType type = 2;
+  string manufacturer = 3;
+  string hw_version = 4;
+  string sw_version = 5;
+  string serial_number = 6;
+  string chassis_id = 7;
+  map<string, string> annotations = 8;
+}
+
+enum MastershipRole {
+  NONE = 0;
+  MASTER = 1;
+  STANDBY = 2;
+}
+
 enum DeviceType {
   OTHER = 0;
   SWITCH = 1;
@@ -20,8 +37,9 @@
   MICROWAVE = 13;
 }
 
-message DeviceDescription {
-  string device_Uri = 1;
+// Corresponds to org.onosproject.net.Device.
+message DeviceCore {
+  string deviceId = 1;
   DeviceType type = 2;
   string manufacturer = 3;
   string hw_version = 4;
@@ -29,10 +47,4 @@
   string serial_number = 6;
   string chassis_id = 7;
   map<string, string> annotations = 8;
-}
-
-enum MastershipRole {
-  NONE = 0;
-  MASTER = 1;
-  STANDBY = 2;
-}
+} 
\ No newline at end of file
diff --git a/incubator/protobuf/src/main/proto/DeviceEvent.proto b/incubator/protobuf/src/main/proto/DeviceEvent.proto
new file mode 100644
index 0000000..9bf7a2f
--- /dev/null
+++ b/incubator/protobuf/src/main/proto/DeviceEvent.proto
@@ -0,0 +1,30 @@
+syntax = "proto3";
+option java_package = "org.onosproject.grpc.net";
+
+
+import "Device.proto";
+import "Port.proto";
+
+package DeviceEvent;
+
+// Corresponds to org.onosproject.net.device.DeviceEvent.
+message DeviceNotification {
+    Device.DeviceCore device = 1;
+    DeviceEventType deviceEventType = 2;
+    Port.PortCore port = 3;
+}
+
+enum DeviceEventType {
+    DEVICE_ADDED = 0;
+    DEVICE_UPDATED = 1;
+    DEVICE_REMOVED = 2;
+    DEVICE_SUSPENDED = 3;
+    DEVICE_AVAILABILITY_CHANGED = 4;
+    PORT_ADDED = 5;
+    PORT_UPDATED = 6;
+    PORT_REMOVED = 7;
+    PORT_STATS_UPDATED = 8;
+}
+
+
+
diff --git a/incubator/protobuf/src/main/proto/Link.proto b/incubator/protobuf/src/main/proto/Link.proto
index 45b9e90..b5fcbfc 100644
--- a/incubator/protobuf/src/main/proto/Link.proto
+++ b/incubator/protobuf/src/main/proto/Link.proto
@@ -38,6 +38,20 @@
   string port_number = 2;
 }
 
+enum LinkState {
+    ACTIVE = 0;
+    INACTIVE = 1;
+}
+
+// Corresponds to org.onosproject.net.Link.
+message LinkCore {
+  LinkState state = 1;
+  ConnectPoint src = 2;
+  ConnectPoint dst = 3;
+  LinkType type = 4;
+  map<string, string> annotations = 5;
+}
+
 message LinkDescription {
   ConnectPoint src = 1;
   ConnectPoint dst = 2;
diff --git a/incubator/protobuf/src/main/proto/LinkEvent.proto b/incubator/protobuf/src/main/proto/LinkEvent.proto
new file mode 100644
index 0000000..fdd6580
--- /dev/null
+++ b/incubator/protobuf/src/main/proto/LinkEvent.proto
@@ -0,0 +1,19 @@
+syntax = "proto3";
+option java_package = "org.onosproject.grpc.net";
+
+package LinkEvent;
+
+import "Link.proto";
+
+// Corresponds to org.onosproject.net.link.LinkEvent.
+message LinkNotification {
+    LinkEventType linkEventType = 2;
+    Link.LinkCore link = 3;
+}
+
+// Link Event Types
+enum LinkEventType {
+    LINK_ADDED = 0;
+    LINK_UPDATED = 1;
+    LINK_REMOVED = 2;
+}
diff --git a/incubator/protobuf/src/main/proto/Port.proto b/incubator/protobuf/src/main/proto/Port.proto
index 0d9ed2e..a3fd34d 100644
--- a/incubator/protobuf/src/main/proto/Port.proto
+++ b/incubator/protobuf/src/main/proto/Port.proto
@@ -31,6 +31,15 @@
   map<string, string> annotations = 8;
 }
 
+// Corresponds to org.onosproject.net.Port.
+message PortCore {
+  string port_number = 1;
+  bool is_enabled = 2;
+  PortType type = 3;
+  int64 port_speed = 4;
+  map<string, string> annotations = 5;
+}
+
 message PortStatistics {
   int32 port = 1;
   int64 packets_received = 2;