diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/ConversionFactory.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/ConversionFactory.java
new file mode 100644
index 0000000..26b5060
--- /dev/null
+++ b/apps/kafka-integration/core/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 org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+
+/**
+ * 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/core/src/main/java/org/onosproject/kafkaintegration/converter/DeviceEventConverter.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/DeviceEventConverter.java
new file mode 100644
index 0000000..6b2ee24
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/DeviceEventConverter.java
@@ -0,0 +1,122 @@
+/**
+ * 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 com.google.protobuf.GeneratedMessage;
+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;
+
+/**
+ * Converts ONOS Device event message to protobuf format.
+ */
+public class DeviceEventConverter implements EventConverter {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public GeneratedMessage convertToProtoMessage(Event<?, ?> event) {
+
+        DeviceEvent deviceEvent = (DeviceEvent) event;
+
+        if (!deviceEventTypeSupported(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 deviceEventTypeSupported(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.Builder notificationBuilder =
+                DeviceNotification.newBuilder();
+
+        DeviceCore deviceCore =
+                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.subject().type().name()))
+                        .build();
+
+        PortCore portCore = null;
+        if (deviceEvent.port() != null) {
+            portCore =
+                    PortCore.newBuilder()
+                            .setIsEnabled(deviceEvent.port().isEnabled())
+                            .setPortNumber(deviceEvent.port().number()
+                                                   .toString())
+                            .setPortSpeed(deviceEvent.port().portSpeed())
+                            .setType(PortType
+                                             .valueOf(deviceEvent.port().type().name()))
+                            .build();
+
+            notificationBuilder.setPort(portCore);
+        }
+
+        notificationBuilder.setDeviceEventType(getProtoType(deviceEvent))
+                .setDevice(deviceCore);
+
+        return notificationBuilder.build();
+    }
+
+    /**
+     * 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/core/src/main/java/org/onosproject/kafkaintegration/converter/EventConverter.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/EventConverter.java
new file mode 100644
index 0000000..c1e7739
--- /dev/null
+++ b/apps/kafka-integration/core/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 protobuf 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 protobuf format.
+     */
+    GeneratedMessage convertToProtoMessage(Event<?, ?> event);
+}
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java
new file mode 100644
index 0000000..febe020
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/LinkEventConverter.java
@@ -0,0 +1,101 @@
+/**
+ * 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 protobuf format.
+ */
+public class LinkEventConverter implements EventConverter {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public GeneratedMessage convertToProtoMessage(Event<?, ?> event) {
+
+        LinkEvent linkEvent = (LinkEvent) event;
+
+        if (!linkEventTypeSupported(linkEvent)) {
+            log.error("Unsupported Onos Event {}. There is no matching "
+                    + "proto Event type", linkEvent.type().toString());
+            return null;
+        }
+
+        return buildDeviceProtoMessage(linkEvent);
+    }
+
+    private boolean linkEventTypeSupported(LinkEvent event) {
+        LinkEventType[] kafkaLinkEvents = LinkEventType.values();
+        for (LinkEventType 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/core/src/main/java/org/onosproject/kafkaintegration/converter/package-info.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/package-info.java
new file mode 100644
index 0000000..d33c0d7
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/converter/package-info.java
@@ -0,0 +1,20 @@
+/**
+ * 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.
+ */
+
+/**
+ * Converters for converting various ONOS events to their corresponding Protocol
+ * Buffer format.
+ *
+ */
+package org.onosproject.kafkaintegration.converter;
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/EventConversionManager.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/EventConversionManager.java
new file mode 100644
index 0000000..f1d0c97
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/EventConversionManager.java
@@ -0,0 +1,72 @@
+/**
+ * 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 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.Service;
+import org.onosproject.event.Event;
+import org.onosproject.kafkaintegration.api.EventConversionService;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent;
+import org.onosproject.kafkaintegration.converter.DeviceEventConverter;
+import org.onosproject.kafkaintegration.converter.EventConverter;
+import org.onosproject.kafkaintegration.converter.LinkEventConverter;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.net.link.LinkEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+
+/**
+ * Implementation of Event Conversion Service.
+ *
+ */
+@Component(immediate = true)
+@Service
+public class EventConversionManager implements EventConversionService {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private EventConverter deviceEventConverter;
+    private EventConverter linkEventConverter;
+
+    @Activate
+    protected void activate() {
+        deviceEventConverter = new DeviceEventConverter();
+        linkEventConverter = new LinkEventConverter();
+
+        log.info("Started");
+    }
+
+    @Deactivate
+    protected void deactivate() {
+        log.info("Stopped");
+    }
+
+    @Override
+    public OnosEvent convertEvent(Event<?, ?> event) {
+        if (event instanceof DeviceEvent) {
+            return new OnosEvent(DEVICE, deviceEventConverter.convertToProtoMessage(event));
+        } else if (event instanceof LinkEvent) {
+            return new OnosEvent(LINK, linkEventConverter.convertToProtoMessage(event));
+        } else {
+            throw new IllegalArgumentException("Unsupported event type");
+        }
+    }
+}
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/EventSubscriptionManager.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/EventSubscriptionManager.java
new file mode 100644
index 0000000..ebbac6b
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/EventSubscriptionManager.java
@@ -0,0 +1,333 @@
+/**
+ * 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 com.google.common.collect.ImmutableList;
+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.EventSubscriptionService;
+import org.onosproject.kafkaintegration.api.dto.DefaultEventSubscriber;
+import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
+import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent;
+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;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+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;
+
+/**
+ * Implementation of Event Subscription Manager.
+ *
+ */
+@Component(immediate = true)
+@Service
+public class EventSubscriptionManager implements EventSubscriptionService {
+
+    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,
+                                                 OnosEvent.class,
+                                                 OnosEvent.Type.class,
+                                                 DefaultEventSubscriber.class,
+                                                 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 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 {} successful",
+                 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());
+    }
+
+    @Override
+    public List<EventSubscriber> getEventSubscribers(Type type) {
+        return subscriptions.getOrDefault(type, ImmutableList.of());
+
+    }
+
+    /**
+     * 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/core/src/main/java/org/onosproject/kafkaintegration/impl/KafkaCodecRegistrator.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/KafkaCodecRegistrator.java
new file mode 100644
index 0000000..22705d5
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/KafkaCodecRegistrator.java
@@ -0,0 +1,47 @@
+/**
+ * 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 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.onosproject.codec.CodecService;
+import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Implementation of the JSON codec brokering service for Kafka app.
+ */
+@Component(immediate = true)
+public class KafkaCodecRegistrator {
+    private static Logger log = LoggerFactory.getLogger(KafkaCodecRegistrator
+                                                                .class);
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected CodecService codecService;
+
+    @Activate
+    public void activate() {
+        codecService.registerCodec(EventSubscriber.class, new SubscriberCodec());
+        log.info("Started");
+    }
+
+    @Deactivate
+    public void deactivate() {
+        log.info("Stopped");
+    }
+}
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/KafkaPublisherManager.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/KafkaPublisherManager.java
new file mode 100644
index 0000000..104619a
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/KafkaPublisherManager.java
@@ -0,0 +1,60 @@
+/**
+ * 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.KafkaPublisherService;
+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 KafkaPublisherManager
+        extends AbstractListenerManager<OnosEvent, ExportableEventListener> implements KafkaPublisherService {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    // Exists to defeat instantiation
+    private KafkaPublisherManager() {
+    }
+
+    //TODO: If possible, get rid of Singleton implementation.
+    private static class SingletonHolder {
+        private static final KafkaPublisherManager INSTANCE = new KafkaPublisherManager();
+    }
+
+    /**
+     * Returns a static reference to the Listener Factory.
+     *
+     * @return singleton object
+     */
+    public static KafkaPublisherManager getInstance() {
+        return SingletonHolder.INSTANCE;
+    }
+
+    @Override
+    public void publish(Type eventType, GeneratedMessage message) {
+        log.debug("Dispatching ONOS Event {}", eventType);
+        post(new OnosEvent(eventType, message));
+    }
+}
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/KafkaStorageManager.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/KafkaStorageManager.java
new file mode 100644
index 0000000..64af4e8
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/KafkaStorageManager.java
@@ -0,0 +1,108 @@
+/**
+ * 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.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.onosproject.kafkaintegration.api.KafkaEventStorageService;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent;
+import org.onosproject.store.service.AtomicValue;
+import org.onosproject.store.service.StorageService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.TreeMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+@Component(immediate = true)
+public class KafkaStorageManager implements KafkaEventStorageService {
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected StorageService storageService;
+
+    private TreeMap<Long, OnosEvent> kafkaEventStore;
+
+    private AtomicValue<Long> lastPublishedEvent;
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    private ScheduledExecutorService gcExService;
+
+    private InternalGarbageCollector gcTask;
+
+    // Thread scheduler parameters.
+    private final long delay = 0;
+    private final long period = 1;
+
+    @Activate
+    protected void activate() {
+        kafkaEventStore = new TreeMap<Long, OnosEvent>();
+        lastPublishedEvent = storageService.<Long>atomicValueBuilder()
+                .withName("onos-app-kafka-published-seqNumber").build()
+                .asAtomicValue();
+
+        startGC();
+
+        log.info("Started");
+    }
+
+    private void startGC() {
+        log.info("Starting Garbage Collection Service");
+        gcExService = Executors.newSingleThreadScheduledExecutor();
+        gcTask = new InternalGarbageCollector();
+        gcExService.scheduleAtFixedRate(gcTask, delay, period,
+                                        TimeUnit.SECONDS);
+    }
+
+    @Deactivate
+    protected void deactivate() {
+        stopGC();
+        log.info("Stopped");
+    }
+
+    private void stopGC() {
+        log.info("Stopping Garbage Collection Service");
+        gcExService.shutdown();
+    }
+
+    @Override
+    public boolean insertCacheEntry(OnosEvent e) {
+        // TODO: Fill in the code once the event carries timestamp info.
+        return true;
+    }
+
+    @Override
+    public void updateLastPublishedEntry(Long sequenceNumber) {
+        this.lastPublishedEvent.set(sequenceNumber);
+    }
+
+    /**
+     * Removes events from the Kafka Event Store which have been published.
+     *
+     */
+    private class InternalGarbageCollector implements Runnable {
+
+        @Override
+        public void run() {
+            kafkaEventStore.headMap(lastPublishedEvent.get(), true).clear();
+        }
+    }
+
+}
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/SubscriberCodec.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/SubscriberCodec.java
new file mode 100644
index 0000000..3489f6f
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/SubscriberCodec.java
@@ -0,0 +1,64 @@
+/**
+ * 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 com.fasterxml.jackson.databind.node.ObjectNode;
+import org.onosproject.codec.CodecContext;
+import org.onosproject.codec.JsonCodec;
+import org.onosproject.kafkaintegration.api.dto.DefaultEventSubscriber;
+import org.onosproject.kafkaintegration.api.dto.EventSubscriber;
+import org.onosproject.kafkaintegration.api.dto.EventSubscriberGroupId;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+
+import java.util.UUID;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Codec for encoding/decoding a Subscriber object to/from JSON.
+ *
+ */
+public final class SubscriberCodec extends JsonCodec<EventSubscriber> {
+
+    // JSON field names
+    private static final String NAME = "appName";
+    private static final String GROUP_ID = "groupId";
+    private static final String EVENT_TYPE = "eventType";
+
+    @Override
+    public ObjectNode encode(EventSubscriber data, CodecContext context) {
+        checkNotNull(data, "Subscriber cannot be null");
+        return context.mapper().createObjectNode().put(NAME, data.appName())
+                .put(GROUP_ID, data.subscriberGroupId().getId().toString())
+                .put(EVENT_TYPE, data.eventType().toString());
+    }
+
+    @Override
+    public EventSubscriber decode(ObjectNode json, CodecContext context) {
+
+        EventSubscriber.Builder resultBuilder = new DefaultEventSubscriber
+                .Builder();
+        String appName = json.get(NAME).asText();
+        resultBuilder.setAppName(appName);
+
+        String subscriberGroupId = json.get(GROUP_ID).asText();
+        resultBuilder.setSubscriberGroupId(new EventSubscriberGroupId(UUID.
+                fromString(subscriberGroupId)));
+
+        String eventType = json.get(EVENT_TYPE).asText();
+        resultBuilder.setEventType(Type.valueOf(eventType));
+        return resultBuilder.build();
+    }
+}
\ No newline at end of file
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/package-info.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/package-info.java
new file mode 100644
index 0000000..a2950fa
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/impl/package-info.java
@@ -0,0 +1,18 @@
+/**
+ * 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.
+ */
+
+/**
+ * API implementation classes.
+ */
+package org.onosproject.kafkaintegration.impl;
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/kafka/EventMonitor.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/kafka/EventMonitor.java
new file mode 100644
index 0000000..6e7291b
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/kafka/EventMonitor.java
@@ -0,0 +1,242 @@
+/**
+ * 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.kafka;
+
+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.Modified;
+import org.apache.felix.scr.annotations.Property;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.onosproject.cfg.ComponentConfigService;
+import org.onosproject.kafkaintegration.api.EventConversionService;
+import org.onosproject.kafkaintegration.api.EventSubscriptionService;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.net.device.DeviceListener;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.link.LinkEvent;
+import org.onosproject.net.link.LinkListener;
+import org.onosproject.net.link.LinkService;
+import org.osgi.service.component.ComponentContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Dictionary;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+
+import static com.google.common.base.Strings.isNullOrEmpty;
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import static org.onlab.util.Tools.get;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+
+/**
+ * Encapsulates the behavior of monitoring various ONOS events.
+ * */
+@Component(immediate = true)
+public class EventMonitor {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected EventSubscriptionService eventSubscriptionService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected EventConversionService eventConversionService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected DeviceService deviceService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected LinkService linkService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected ComponentConfigService componentConfigService;
+
+    private final DeviceListener deviceListener = new InternalDeviceListener();
+    private final LinkListener linkListener = new InternalLinkListener();
+
+    protected ExecutorService eventExecutor;
+
+    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
+    private static final int RETRIES = 1;
+    private static final int MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = 5;
+    private static final int REQUEST_REQUIRED_ACKS = 1;
+    private static final String KEY_SERIALIZER = "org.apache.kafka.common.serialization.StringSerializer";
+    private static final String VALUE_SERIALIZER = "org.apache.kafka.common.serialization.ByteArraySerializer";
+
+    @Property(name = "bootstrap.servers", value = BOOTSTRAP_SERVERS,
+            label = "Default host/post pair to establish initial connection to Kafka cluster.")
+    private String bootstrapServers = BOOTSTRAP_SERVERS;
+
+    @Property(name = "retries", intValue = RETRIES,
+            label = "Number of times the producer can retry to send after first failure")
+    private int retries = RETRIES;
+
+    @Property(name = "max.in.flight.requests.per.connection", intValue = MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,
+            label = "The maximum number of unacknowledged requests the client will send before blocking")
+    private int maxInFlightRequestsPerConnection = MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION;
+
+    @Property(name = "request.required.acks", intValue = 1,
+            label = "Producer will get an acknowledgement after the leader has replicated the data")
+    private int requestRequiredAcks = REQUEST_REQUIRED_ACKS;
+
+    @Property(name = "key.serializer", value = KEY_SERIALIZER,
+            label = "Serializer class for key that implements the Serializer interface.")
+    private String keySerializer = KEY_SERIALIZER;
+
+    @Property(name = "value.serializer", value = VALUE_SERIALIZER,
+            label = "Serializer class for value that implements the Serializer interface.")
+    private String valueSerializer = VALUE_SERIALIZER;
+
+    private Producer producer;
+
+    @Activate
+    protected void activate(ComponentContext context) {
+        componentConfigService.registerProperties(getClass());
+        eventExecutor = newSingleThreadScheduledExecutor(groupedThreads("onos/onosEvents", "events-%d", log));
+        deviceService.addListener(deviceListener);
+        linkService.addListener(linkListener);
+        producer = new Producer(bootstrapServers, retries, maxInFlightRequestsPerConnection,
+                                requestRequiredAcks, keySerializer, valueSerializer);
+        producer.start();
+
+        log.info("Started");
+    }
+
+    @Deactivate
+    protected void deactivate() {
+        componentConfigService.unregisterProperties(getClass(), false);
+        deviceService.removeListener(deviceListener);
+        linkService.removeListener(linkListener);
+        producer.stop();
+        eventExecutor.shutdownNow();
+        eventExecutor = null;
+
+        log.info("Stopped");
+    }
+
+    @Modified
+    private void modified(ComponentContext context) {
+        if (context == null) {
+            bootstrapServers = BOOTSTRAP_SERVERS;
+            retries = RETRIES;
+            maxInFlightRequestsPerConnection = MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION;
+            requestRequiredAcks = REQUEST_REQUIRED_ACKS;
+            keySerializer = KEY_SERIALIZER;
+            valueSerializer = VALUE_SERIALIZER;
+            return;
+        }
+        Dictionary properties = context.getProperties();
+
+        String newBootstrapServers = BOOTSTRAP_SERVERS;
+        int newRetries = RETRIES;
+        int newMaxInFlightRequestsPerConnection = MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION;
+        int newRequestRequiredAcks = REQUEST_REQUIRED_ACKS;
+        try {
+            String s = get(properties, "bootstrapServers");
+            newBootstrapServers = isNullOrEmpty(s)
+                    ? bootstrapServers : s.trim();
+
+            s = get(properties, "retries");
+            newRetries = isNullOrEmpty(s)
+                    ? retries : Integer.parseInt(s.trim());
+
+            s = get(properties, "maxInFlightRequestsPerConnection");
+            newMaxInFlightRequestsPerConnection = isNullOrEmpty(s)
+                    ? maxInFlightRequestsPerConnection : Integer.parseInt(s.trim());
+
+            s = get(properties, "requestRequiredAcks");
+            newRequestRequiredAcks = isNullOrEmpty(s)
+                    ? requestRequiredAcks : Integer.parseInt(s.trim());
+
+        } catch (NumberFormatException | ClassCastException e) {
+            return;
+        }
+
+        boolean modified = newBootstrapServers != bootstrapServers ||
+                newRetries != retries ||
+                newMaxInFlightRequestsPerConnection != maxInFlightRequestsPerConnection ||
+                newRequestRequiredAcks != requestRequiredAcks;
+
+        if (modified) {
+            bootstrapServers = newBootstrapServers;
+            retries = newRetries;
+            maxInFlightRequestsPerConnection = newMaxInFlightRequestsPerConnection;
+            requestRequiredAcks = newRequestRequiredAcks;
+            if (producer != null) {
+                producer.stop();
+            }
+            producer = new Producer(bootstrapServers, retries, maxInFlightRequestsPerConnection,
+                                        requestRequiredAcks, keySerializer, valueSerializer);
+            producer.start();
+            log.info("Modified");
+        } else {
+            return;
+        }
+    }
+
+    private class InternalDeviceListener implements DeviceListener {
+
+        @Override
+        public void event(DeviceEvent event) {
+            if (!eventSubscriptionService.getEventSubscribers(DEVICE).isEmpty()) {
+                eventExecutor.execute(() -> {
+                    try {
+                        String id = UUID.randomUUID().toString();
+                        producer.send(new ProducerRecord<>(DEVICE.toString(),
+                                                                id, event.subject().toString().getBytes())).get();
+                        log.debug("Device event sent successfully.");
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                    } catch (ExecutionException e) {
+                        log.error("Exception thrown {}", e);
+                    }
+                });
+            } else {
+                log.debug("No device listeners");
+            }
+        }
+    }
+
+    private class InternalLinkListener implements LinkListener {
+
+        @Override
+        public void event(LinkEvent event) {
+            if (!eventSubscriptionService.getEventSubscribers(LINK).isEmpty()) {
+                eventExecutor.execute(() -> {
+                    try {
+                        String id = UUID.randomUUID().toString();
+                        producer.send(new ProducerRecord<>(LINK.toString(),
+                                                                id, event.subject().toString().getBytes())).get();
+                        log.debug("Link event sent successfully.");
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                    } catch (ExecutionException e) {
+                        log.error("Exception thrown {}", e);
+                    }
+                });
+            } else {
+                log.debug("No link listeners");
+            }
+        }
+    }
+}
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/kafka/Producer.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/kafka/Producer.java
new file mode 100644
index 0000000..bbdff81
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/kafka/Producer.java
@@ -0,0 +1,66 @@
+/**
+ * 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.kafka;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Properties;
+import java.util.concurrent.Future;
+
+/**
+ * Implementation of Kafka Producer.
+ */
+public class Producer {
+    private KafkaProducer<String, byte[]> kafkaProducer = null;
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    Producer(String bootstrapServers, int retries, int maxInFlightRequestsPerConnection,
+             int requestRequiredAcks, String keySerializer, String valueSerializer) {
+
+        Properties prop = new Properties();
+        prop.put("bootstrap.servers", bootstrapServers);
+        prop.put("retries", retries);
+        prop.put("max.in.flight.requests.per.connection", maxInFlightRequestsPerConnection);
+        prop.put("request.required.acks", requestRequiredAcks);
+        prop.put("key.serializer", keySerializer);
+        prop.put("value.serializer", valueSerializer);
+
+        kafkaProducer = new KafkaProducer<>(prop);
+    }
+
+    public void start() {
+        log.info("Started");
+    }
+
+    public void stop() {
+        if (kafkaProducer != null) {
+            kafkaProducer.close();
+            kafkaProducer = null;
+        }
+
+        log.info("Stopped");
+    }
+
+    public Future<RecordMetadata> send(ProducerRecord<String, byte[]> record) {
+        return kafkaProducer.send(record);
+    }
+}
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/kafka/package-info.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/kafka/package-info.java
new file mode 100644
index 0000000..e222a64
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/kafka/package-info.java
@@ -0,0 +1,18 @@
+/**
+ * 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.
+ */
+
+/**
+ * API implementation classes.
+ */
+package org.onosproject.kafkaintegration.kafka;
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/listener/DeviceEventsListener.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/listener/DeviceEventsListener.java
new file mode 100644
index 0000000..e2f23c0
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/listener/DeviceEventsListener.java
@@ -0,0 +1,96 @@
+/**
+ * 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 com.google.protobuf.GeneratedMessage;
+import org.onosproject.event.ListenerService;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+import org.onosproject.kafkaintegration.converter.ConversionFactory;
+import org.onosproject.kafkaintegration.converter.EventConverter;
+import org.onosproject.kafkaintegration.impl.KafkaPublisherManager;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.net.device.DeviceListener;
+import org.onosproject.net.device.DeviceService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+
+/**
+ * Listens for ONOS Device events.
+ *
+ */
+public final class DeviceEventsListener implements OnosEventListener {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    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 protobuf format
+            ConversionFactory conversionFactory =
+                    ConversionFactory.getInstance();
+            EventConverter converter = conversionFactory.getConverter(DEVICE);
+            GeneratedMessage message = converter.convertToProtoMessage(arg0);
+
+            // Call Dispatcher and publish event
+            KafkaPublisherManager.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;
+        }
+    }
+
+}
\ No newline at end of file
diff --git a/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/listener/LinkEventsListener.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/listener/LinkEventsListener.java
new file mode 100644
index 0000000..39ab078
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/listener/LinkEventsListener.java
@@ -0,0 +1,90 @@
+/**
+ * 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 com.google.protobuf.GeneratedMessage;
+import org.onosproject.event.ListenerService;
+import org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+import org.onosproject.kafkaintegration.converter.ConversionFactory;
+import org.onosproject.kafkaintegration.converter.EventConverter;
+import org.onosproject.kafkaintegration.impl.KafkaPublisherManager;
+import org.onosproject.net.link.LinkEvent;
+import org.onosproject.net.link.LinkListener;
+import org.onosproject.net.link.LinkService;
+
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+
+/**
+ * Listens for ONOS Link Events.
+ *
+ */
+public 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 protobuf format
+            ConversionFactory conversionFactory =
+                    ConversionFactory.getInstance();
+            EventConverter converter = conversionFactory.getConverter(LINK);
+            GeneratedMessage message = converter.convertToProtoMessage(arg0);
+
+            // Call Dispatcher and publish event
+            KafkaPublisherManager.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/core/src/main/java/org/onosproject/kafkaintegration/listener/ListenerFactory.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/listener/ListenerFactory.java
new file mode 100644
index 0000000..65b7cf1
--- /dev/null
+++ b/apps/kafka-integration/core/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 org.onosproject.kafkaintegration.api.dto.OnosEvent.Type;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.DEVICE;
+import static org.onosproject.kafkaintegration.api.dto.OnosEvent.Type.LINK;
+
+/**
+ * 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/core/src/main/java/org/onosproject/kafkaintegration/listener/OnosEventListener.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/listener/OnosEventListener.java
new file mode 100644
index 0000000..ad14104
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/listener/OnosEventListener.java
@@ -0,0 +1,41 @@
+/**
+ * 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/core/src/main/java/org/onosproject/kafkaintegration/listener/package-info.java b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/listener/package-info.java
new file mode 100644
index 0000000..f6f602f
--- /dev/null
+++ b/apps/kafka-integration/core/src/main/java/org/onosproject/kafkaintegration/listener/package-info.java
@@ -0,0 +1,19 @@
+/**
+ * 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.
+ */
+
+/**
+ * Listeners for listening to various ONOS events.
+ *
+ */
+package org.onosproject.kafkaintegration.listener;