diff --git a/app/app.xml b/app/app.xml
new file mode 100644
index 0000000..d441b88
--- /dev/null
+++ b/app/app.xml
@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2021-present Open Networking Foundation
+  ~
+  ~ 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.opencord.pppoeagent" origin="Open Networking Foundation" version="${project.version}"
+     category="vOlt" url="http://onosproject.org" title="PPPoE Agent"
+     featuresRepo="mvn:${project.groupId}/${project.artifactId}/${project.version}/xml/features"
+     features="${project.artifactId}" apps="org.opencord.sadis">
+    <description>${project.description}</description>
+    <artifact>mvn:${project.groupId}/${project.artifactId}/${project.version}</artifact>
+    <artifact>mvn:${project.groupId}/pppoeagent-api/${project.version}</artifact>
+</app>
diff --git a/app/features.xml b/app/features.xml
new file mode 100644
index 0000000..7b8b9a3
--- /dev/null
+++ b/app/features.xml
@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<!--
+  ~ Copyright 2021-present Open Networking Foundation
+  ~
+  ~ 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}/pppoeagent-api/${project.version}</bundle>
+        <bundle>mvn:${project.groupId}/${project.artifactId}/${project.version}</bundle>
+    </feature>
+</features>
diff --git a/app/pom.xml b/app/pom.xml
new file mode 100644
index 0000000..187a9b7
--- /dev/null
+++ b/app/pom.xml
@@ -0,0 +1,190 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2021-present Open Networking Foundation
+  ~
+  ~ 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/maven-v4_0_0.xsd">
+    <parent>
+        <groupId>org.opencord</groupId>
+        <artifactId>pppoeagent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>pppoeagent-app</artifactId>
+
+    <packaging>bundle</packaging>
+    <description>PPPoE Intermediate Agent application</description>
+
+    <properties>
+        <web.context>/onos/pppoeagent</web.context>
+        <api.version>1.0.0-SNAPSHOT</api.version>
+        <api.title>PPPoE Agent REST API</api.title>
+        <api.description>REST API to query PPPoE agent sessions information.</api.description>
+        <api.package>org.opencord.pppoeagent.rest</api.package>
+        <onos.app.requires>
+            org.opencord.sadis
+        </onos.app.requires>
+        <onos.app.name>org.opencord.pppoeagent</onos.app.name>
+        <onos.app.title>PPPoE Intermediate Agent App</onos.app.title>
+        <onos.app.category>vOLT</onos.app.category>
+        <onos.app.url>http://opencord.org</onos.app.url>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <onos.app.readme>PPPoE Intermediate Agent</onos.app.readme>
+        <onos.app.origin>Furukawa Electric LatAm</onos.app.origin>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-api</artifactId>
+            <version>${onos.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-core-serializers</artifactId>
+            <version>${onos.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onlab-rest</artifactId>
+            <version>${onos.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.opencord</groupId>
+            <artifactId>pppoeagent-api</artifactId>
+            <version>1.0.0-SNAPSHOT</version>
+            <scope>compile</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onlab-junit</artifactId>
+            <version>${onos.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-cli</artifactId>
+            <version>${onos.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-api</artifactId>
+            <version>${onos.version}</version>
+            <scope>test</scope>
+            <classifier>tests</classifier>
+        </dependency>
+
+        <dependency>
+            <groupId>org.opencord</groupId>
+            <artifactId>sadis-api</artifactId>
+            <version>${sadis.api.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-core-common</artifactId>
+            <version>${onos.version}</version>
+            <classifier>tests</classifier>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onlab-misc</artifactId>
+            <version>${onos.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-api</artifactId>
+            <version>${onos.version}</version>
+            <scope>test</scope>
+            <classifier>tests</classifier>
+        </dependency>
+
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onlab-osgi</artifactId>
+            <version>${onos.version}</version>
+            <classifier>tests</classifier>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.karaf.shell</groupId>
+            <artifactId>org.apache.karaf.shell.core</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.karaf.shell</groupId>
+            <artifactId>org.apache.karaf.shell.console</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>javax.ws.rs</groupId>
+            <artifactId>javax.ws.rs-api</artifactId>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.onosproject</groupId>
+                <artifactId>onos-maven-plugin</artifactId>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.felix</groupId>
+                <artifactId>maven-bundle-plugin</artifactId>
+                <configuration>
+                    <instructions>
+                        <_wab>src/main/webapp/</_wab>
+                        <Include-Resource>
+                            WEB-INF/classes/apidoc/swagger.json=target/swagger.json,
+                            {maven-resources}
+                        </Include-Resource>
+                        <Import-Package>
+                            *,org.glassfish.jersey.servlet
+                        </Import-Package>
+                        <Web-ContextPath>${web.context}</Web-ContextPath>
+                        <Karaf-Commands>org.opencord.pppoeagent.cli</Karaf-Commands>
+                    </instructions>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>
diff --git a/app/src/main/java/org/opencord/pppoeagent/cli/PppoeAgentShowUsersCommand.java b/app/src/main/java/org/opencord/pppoeagent/cli/PppoeAgentShowUsersCommand.java
new file mode 100644
index 0000000..a67ffca
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/cli/PppoeAgentShowUsersCommand.java
@@ -0,0 +1,87 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.cli;
+
+import org.opencord.pppoeagent.PppoeAgentService;
+import org.opencord.pppoeagent.PppoeSessionInfo;
+import org.apache.karaf.shell.api.action.Argument;
+import org.apache.karaf.shell.api.action.Command;
+import org.apache.karaf.shell.api.action.lifecycle.Service;
+import org.onlab.packet.MacAddress;
+import org.onlab.packet.PPPoED;
+import org.onosproject.cli.AbstractShellCommand;
+import org.onosproject.net.AnnotationKeys;
+import org.onosproject.net.ConnectPoint;
+import org.onosproject.net.Port;
+import org.onosproject.net.device.DeviceService;
+
+/**
+ *  Shows the PPPoE sessions/users learned by the agent.
+ */
+@Service
+@Command(scope = "pppoe", name = "pppoe-users",
+        description = "Shows the PPPoE users")
+public class PppoeAgentShowUsersCommand extends AbstractShellCommand {
+
+    @Argument(index = 0, name = "mac", description = "MAC address related to PPPoE session.",
+            required = false, multiValued = false)
+    private String macStr = null;
+
+    @Override
+    protected void doExecute() {
+        MacAddress macAddress = null;
+        if (macStr != null && !macStr.isEmpty()) {
+            try {
+                macAddress = MacAddress.valueOf(macStr);
+            } catch (IllegalArgumentException e) {
+                log.error(e.getMessage());
+                return;
+            }
+        }
+
+        DeviceService deviceService = AbstractShellCommand.get(DeviceService.class);
+        PppoeAgentService pppoeAgentService = AbstractShellCommand.get(PppoeAgentService.class);
+
+        if (macAddress != null) {
+            PppoeSessionInfo singleInfo = pppoeAgentService.getSessionsMap().get(macAddress);
+            if (singleInfo != null) {
+                Port devicePort = deviceService.getPort(singleInfo.getClientCp());
+                printPppoeInfo(macAddress, singleInfo, devicePort);
+            } else {
+                print("No session information found for provided MAC address %s", macAddress.toString());
+            }
+        } else {
+            pppoeAgentService.getSessionsMap().forEach((mac, sessionInfo) -> {
+                final Port devicePortFinal = deviceService.getPort(sessionInfo.getClientCp());
+                printPppoeInfo(mac, sessionInfo, devicePortFinal);
+            });
+        }
+    }
+
+    private void printPppoeInfo(MacAddress macAddr, PppoeSessionInfo sessionInfo, Port devicePort) {
+        PPPoED.Type lastReceivedPkt = PPPoED.Type.getTypeByValue(sessionInfo.getPacketCode());
+        ConnectPoint cp = sessionInfo.getClientCp();
+        String subscriberId = devicePort != null ? devicePort.annotations().value(AnnotationKeys.PORT_NAME) :
+                "UNKNOWN";
+
+        print("MacAddress=%s,SessionId=%s,CurrentState=%s,LastReceivedPacket=%s,DeviceId=%s,PortNumber=%s," +
+                        "SubscriberId=%s",
+                macAddr.toString(), String.valueOf(sessionInfo.getSessionId()),
+                sessionInfo.getCurrentState(), lastReceivedPkt.name(),
+                cp.deviceId().toString(), cp.port().toString(), subscriberId);
+    }
+}
diff --git a/app/src/main/java/org/opencord/pppoeagent/cli/PppoeAgentStatsCommand.java b/app/src/main/java/org/opencord/pppoeagent/cli/PppoeAgentStatsCommand.java
new file mode 100644
index 0000000..885bf36
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/cli/PppoeAgentStatsCommand.java
@@ -0,0 +1,137 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.cli;
+
+import org.opencord.pppoeagent.PppoeAgentEvent;
+import org.apache.karaf.shell.api.action.Argument;
+import org.apache.karaf.shell.api.action.Command;
+import org.apache.karaf.shell.api.action.Option;
+import org.apache.karaf.shell.api.action.lifecycle.Service;
+import org.onosproject.cli.AbstractShellCommand;
+import java.util.Collections;
+import java.util.Map;
+import org.opencord.pppoeagent.impl.PppoeAgentCountersIdentifier;
+import org.opencord.pppoeagent.impl.PppoeAgentCountersStore;
+import org.opencord.pppoeagent.impl.PppoeAgentCounterNames;
+
+/**
+ * Display/Reset the PPPoE Agent application statistics.
+ */
+@Service
+@Command(scope = "pppoe", name = "pppoeagent-stats",
+        description = "Display or Reset the PPPoE Agent application statistics.")
+public class PppoeAgentStatsCommand extends AbstractShellCommand {
+    private static final String CONFIRM_PHRASE = "please";
+
+    @Option(name = "-r", aliases = "--reset", description = "Resets a specific counter.\n",
+            required = false, multiValued = false)
+    PppoeAgentCounterNames reset = null;
+
+    @Option(name = "-R", aliases = "--reset-all", description = "Resets all counters.\n",
+            required = false, multiValued = false)
+    boolean resetAll = false;
+
+    @Option(name = "-s", aliases = "--subscriberId", description = "Subscriber Id.\n",
+            required = false, multiValued = false)
+    String subscriberId = null;
+
+    @Option(name = "-p", aliases = "--please", description = "Confirmation phrase.",
+            required = false, multiValued = false)
+    String please = null;
+
+    @Argument(index = 0, name = "counter",
+            description = "The counter to display. In case not specified, all counters will be displayed.",
+            required = false, multiValued = false)
+    PppoeAgentCounterNames counter = null;
+
+    @Override
+    protected void doExecute() {
+        PppoeAgentCountersStore pppoeCounters = AbstractShellCommand.get(PppoeAgentCountersStore.class);
+
+        if ((subscriberId == null) || (subscriberId.equals("global"))) {
+            // All subscriber Ids
+            subscriberId = PppoeAgentEvent.GLOBAL_COUNTER;
+        }
+
+        if (resetAll || reset != null) {
+            if (please == null || !please.equals(CONFIRM_PHRASE)) {
+                print("WARNING: Be aware that you are going to reset the counters. " +
+                        "Enter confirmation phrase to continue.");
+                return;
+            }
+            if (resetAll) {
+                // Reset all counters.
+                pppoeCounters.resetCounters(subscriberId);
+            } else {
+                // Reset the specified counter.
+                pppoeCounters.setCounter(subscriberId, reset, (long) 0);
+            }
+        } else {
+            Map<PppoeAgentCountersIdentifier, Long> countersMap = pppoeCounters.getCounters().counters();
+            if (countersMap.size() > 0) {
+                if (counter == null) {
+                    String jsonString = "";
+                    if (outputJson()) {
+                        jsonString = String.format("{\"%s\":{", pppoeCounters.NAME);
+                    } else {
+                        print("%s [%s] :", pppoeCounters.NAME, subscriberId);
+                    }
+                    PppoeAgentCounterNames[] counters = PppoeAgentCounterNames.values();
+                    for (int i = 0; i < counters.length; i++) {
+                        PppoeAgentCounterNames counterType = counters[i];
+                        Long value = countersMap.get(new PppoeAgentCountersIdentifier(subscriberId, counterType));
+                        if (value == null) {
+                            value = 0L;
+                        }
+                        if (outputJson()) {
+                            jsonString += String.format("\"%s\":%d", counterType, value);
+                            if (i < counters.length - 1) {
+                                jsonString += ",";
+                            }
+                        } else {
+                            printCounter(counterType, value);
+                        }
+                    }
+                    if (outputJson()) {
+                        jsonString += "}}";
+                        print("%s", jsonString);
+                    }
+                } else {
+                    // Show only the specified counter
+                    Long value = countersMap.get(new PppoeAgentCountersIdentifier(subscriberId, counter));
+                    if (value == null) {
+                        value = 0L;
+                    }
+                    if (outputJson()) {
+                        print("{\"%s\":%d}", counter, value);
+                    } else {
+                        printCounter(counter, value);
+                    }
+                }
+            } else {
+                print("No PPPoE Agent Counters were created yet for counter class [%s]",
+                        PppoeAgentEvent.GLOBAL_COUNTER);
+            }
+        }
+    }
+
+    void printCounter(PppoeAgentCounterNames c, long value) {
+        // print in non-JSON format
+        print("  %s %s %-4d", c,
+                String.join("", Collections.nCopies(50 - c.toString().length(), ".")), value);
+    }
+}
\ No newline at end of file
diff --git a/app/src/main/java/org/opencord/pppoeagent/cli/package-info.java b/app/src/main/java/org/opencord/pppoeagent/cli/package-info.java
new file mode 100644
index 0000000..515908f
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/cli/package-info.java
@@ -0,0 +1,19 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.
+ */
+/**
+ * CLI commands for the PPPoE agent.
+ */
+package org.opencord.pppoeagent.cli;
diff --git a/app/src/main/java/org/opencord/pppoeagent/impl/OsgiPropertyConstants.java b/app/src/main/java/org/opencord/pppoeagent/impl/OsgiPropertyConstants.java
new file mode 100644
index 0000000..65c8310
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/impl/OsgiPropertyConstants.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.impl;
+
+/**
+ * Constants for default values of configurable properties.
+ */
+public final class OsgiPropertyConstants {
+
+    private OsgiPropertyConstants() {
+    }
+
+    public static final String ENABLE_CIRCUIT_ID_VALIDATION = "enableCircuitIdValidation";
+    public static final boolean ENABLE_CIRCUIT_ID_VALIDATION_DEFAULT = true;
+
+    public static final String PPPOE_COUNTERS_TOPIC = "pppoeCountersTopic";
+    public static final String PPPOE_COUNTERS_TOPIC_DEFAULT = "onos.pppoe.stats.kpis";
+
+    public static final String PUBLISH_COUNTERS_RATE = "publishCountersRate";
+    public static final int PUBLISH_COUNTERS_RATE_DEFAULT = 10;
+
+    public static final String PPPOE_MAX_MTU = "pppoeMaxMtu";
+    public static final int PPPOE_MAX_MTU_DEFAULT = 1500;
+
+    public static final String PACKET_PROCESSOR_THREADS = "packetProcessorThreads";
+    public static final int PACKET_PROCESSOR_THREADS_DEFAULT = 10;
+
+    public static final String SYNC_COUNTERS_RATE = "syncCountersRate";
+    public static final int SYNC_COUNTERS_RATE_DEFAULT = 5;
+}
diff --git a/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgent.java b/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgent.java
new file mode 100644
index 0000000..67c71da
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgent.java
@@ -0,0 +1,929 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.impl;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableSet;
+
+import com.google.common.collect.Sets;
+import org.opencord.pppoeagent.PppoeAgentEvent;
+import org.opencord.pppoeagent.PppoeAgentListener;
+import org.opencord.pppoeagent.PppoeAgentService;
+import org.opencord.pppoeagent.PppoeAgentStoreDelegate;
+import org.opencord.pppoeagent.PPPoEDVendorSpecificTag;
+import org.opencord.pppoeagent.PppoeSessionInfo;
+import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.Serializer;
+import com.google.common.collect.ImmutableMap;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.ConsistentMap;
+
+import org.onlab.util.KryoNamespace;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.MacAddress;
+import org.onlab.packet.PPPoED;
+import org.onlab.packet.PPPoEDTag;
+import org.onlab.util.Tools;
+import org.onosproject.cfg.ComponentConfigService;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreService;
+import org.onosproject.event.AbstractListenerManager;
+import org.onosproject.mastership.MastershipEvent;
+import org.onosproject.mastership.MastershipListener;
+import org.onosproject.mastership.MastershipService;
+import org.onosproject.net.AnnotationKeys;
+import org.onosproject.net.ConnectPoint;
+import org.onosproject.net.Device;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.Port;
+import org.onosproject.net.PortNumber;
+import org.onosproject.net.config.ConfigFactory;
+import org.onosproject.net.config.NetworkConfigEvent;
+import org.onosproject.net.config.NetworkConfigListener;
+import org.onosproject.net.config.NetworkConfigRegistry;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.net.device.DeviceListener;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.flow.DefaultTrafficTreatment;
+import org.onosproject.net.flow.TrafficTreatment;
+import org.onosproject.net.packet.DefaultOutboundPacket;
+import org.onosproject.net.packet.OutboundPacket;
+import org.onosproject.net.packet.PacketContext;
+import org.onosproject.net.packet.PacketProcessor;
+import org.onosproject.net.packet.PacketService;
+import org.opencord.sadis.BaseInformationService;
+import org.opencord.sadis.SadisService;
+import org.opencord.sadis.SubscriberAndDeviceInformation;
+import org.opencord.sadis.UniTagInformation;
+import org.opencord.pppoeagent.util.CircuitIdBuilder;
+import org.opencord.pppoeagent.util.CircuitIdFieldName;
+import org.osgi.service.component.ComponentContext;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Modified;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Dictionary;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicReference;
+import org.onosproject.store.service.Versioned;
+
+import static java.util.concurrent.Executors.newFixedThreadPool;
+
+import static org.onlab.packet.PPPoED.PPPOED_CODE_PADI;
+import static org.onlab.packet.PPPoED.PPPOED_CODE_PADO;
+import static org.onlab.packet.PPPoED.PPPOED_CODE_PADR;
+import static org.onlab.packet.PPPoED.PPPOED_CODE_PADS;
+import static org.onlab.packet.PPPoED.PPPOED_CODE_PADT;
+import static org.onlab.packet.PPPoEDTag.PPPOED_TAG_AC_SYSTEM_ERROR;
+import static org.onlab.packet.PPPoEDTag.PPPOED_TAG_GENERIC_ERROR;
+import static org.onlab.packet.PPPoEDTag.PPPOED_TAG_SERVICE_NAME_ERROR;
+import static org.onlab.packet.PPPoEDTag.PPPOED_TAG_VENDOR_SPECIFIC;
+
+import static org.onlab.util.Tools.getIntegerProperty;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.onosproject.net.config.basics.SubjectFactories.APP_SUBJECT_FACTORY;
+
+import static org.opencord.pppoeagent.impl.OsgiPropertyConstants.PPPOE_MAX_MTU;
+import static org.opencord.pppoeagent.impl.OsgiPropertyConstants.PPPOE_MAX_MTU_DEFAULT;
+import static org.opencord.pppoeagent.impl.OsgiPropertyConstants.ENABLE_CIRCUIT_ID_VALIDATION;
+import static org.opencord.pppoeagent.impl.OsgiPropertyConstants.ENABLE_CIRCUIT_ID_VALIDATION_DEFAULT;
+import static org.opencord.pppoeagent.impl.OsgiPropertyConstants.PACKET_PROCESSOR_THREADS;
+import static org.opencord.pppoeagent.impl.OsgiPropertyConstants.PACKET_PROCESSOR_THREADS_DEFAULT;
+
+/**
+ * PPPoE Intermediate Agent application.
+ */
+@Component(immediate = true,
+property = {
+        PPPOE_MAX_MTU + ":Integer=" + PPPOE_MAX_MTU_DEFAULT,
+        ENABLE_CIRCUIT_ID_VALIDATION + ":Boolean=" + ENABLE_CIRCUIT_ID_VALIDATION_DEFAULT,
+        PACKET_PROCESSOR_THREADS + ":Integer=" + PACKET_PROCESSOR_THREADS_DEFAULT,
+})
+public class PppoeAgent
+        extends AbstractListenerManager<PppoeAgentEvent, PppoeAgentListener>
+        implements PppoeAgentService {
+    private static final String APP_NAME = "org.opencord.pppoeagent";
+    private static final short QINQ_VID_NONE = (short) -1;
+
+    private final InternalConfigListener cfgListener = new InternalConfigListener();
+    private final Set<ConfigFactory> factories = ImmutableSet.of(
+            new ConfigFactory<>(APP_SUBJECT_FACTORY, PppoeAgentConfig.class, "pppoeagent") {
+                @Override
+                public PppoeAgentConfig createConfig() {
+                    return new PppoeAgentConfig();
+                }
+            }
+    );
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected CoreService coreService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected NetworkConfigRegistry cfgService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected PacketService packetService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected ComponentConfigService componentConfigService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected SadisService sadisService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected DeviceService deviceService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected MastershipService mastershipService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected StorageService storageService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected PppoeAgentCountersStore pppoeAgentCounters;
+
+    // OSGi Properties
+    protected int pppoeMaxMtu = PPPOE_MAX_MTU_DEFAULT;
+    protected boolean enableCircuitIdValidation = ENABLE_CIRCUIT_ID_VALIDATION_DEFAULT;
+    protected int packetProcessorThreads = PACKET_PROCESSOR_THREADS_DEFAULT;
+
+    private ApplicationId appId;
+    private InnerDeviceListener deviceListener = new InnerDeviceListener();
+    private InnerMastershipListener changeListener = new InnerMastershipListener();
+    private PppoeAgentPacketProcessor pppoeAgentPacketProcessor = new PppoeAgentPacketProcessor();
+    protected BaseInformationService<SubscriberAndDeviceInformation> subsService;
+    private PppoeAgentStoreDelegate delegate = new InnerPppoeAgentStoreDelegate();
+
+    Set<ConnectPoint> pppoeConnectPoints;
+    protected AtomicReference<ConnectPoint> pppoeServerConnectPoint = new AtomicReference<>();
+    protected boolean useOltUplink = false;
+
+    static ConsistentMap<MacAddress, PppoeSessionInfo> sessionsMap;
+
+    @Override
+    public Map<MacAddress, PppoeSessionInfo> getSessionsMap() {
+        return ImmutableMap.copyOf(sessionsMap.asJavaMap());
+    }
+
+    @Override
+    public void clearSessionsMap() {
+        sessionsMap.clear();
+    }
+
+    private final ArrayList<CircuitIdFieldName> circuitIdfields = new ArrayList<>(Arrays.asList(
+            CircuitIdFieldName.AcessNodeIdentifier,
+            CircuitIdFieldName.Slot,
+            CircuitIdFieldName.Port,
+            CircuitIdFieldName.OnuSerialNumber));
+
+    protected ExecutorService packetProcessorExecutor;
+
+    @Activate
+    protected void activate(ComponentContext context) {
+        eventDispatcher.addSink(PppoeAgentEvent.class, listenerRegistry);
+
+        appId = coreService.registerApplication(APP_NAME);
+        cfgService.addListener(cfgListener);
+        componentConfigService.registerProperties(getClass());
+
+        KryoNamespace serializer = KryoNamespace.newBuilder()
+                .register(KryoNamespaces.API)
+                .register(PppoeSessionInfo.class)
+                .register(MacAddress.class)
+                .register(SubscriberAndDeviceInformation.class)
+                .register(UniTagInformation.class)
+                .register(ConnectPoint.class)
+                .build();
+
+        sessionsMap = storageService.<MacAddress, PppoeSessionInfo>consistentMapBuilder()
+                .withName("pppoeagent-sessions")
+                .withSerializer(Serializer.using(serializer))
+                .withApplicationId(appId)
+                .build();
+
+        factories.forEach(cfgService::registerConfigFactory);
+        deviceService.addListener(deviceListener);
+        subsService = sadisService.getSubscriberInfoService();
+        mastershipService.addListener(changeListener);
+        pppoeAgentCounters.setDelegate(delegate);
+        updateConfig();
+        packetService.addProcessor(pppoeAgentPacketProcessor, PacketProcessor.director(0));
+        if (context != null) {
+            modified(context);
+        }
+        log.info("PPPoE Intermediate Agent Started");
+    }
+
+    @Deactivate
+    protected void deactivate() {
+        cfgService.removeListener(cfgListener);
+        factories.forEach(cfgService::unregisterConfigFactory);
+        packetService.removeProcessor(pppoeAgentPacketProcessor);
+        packetProcessorExecutor.shutdown();
+        componentConfigService.unregisterProperties(getClass(), false);
+        deviceService.removeListener(deviceListener);
+        eventDispatcher.removeSink(PppoeAgentEvent.class);
+        pppoeAgentCounters.unsetDelegate(delegate);
+        log.info("PPPoE Intermediate Agent Stopped");
+    }
+
+    private void updateConfig() {
+        PppoeAgentConfig cfg = cfgService.getConfig(appId, PppoeAgentConfig.class);
+        if (cfg == null) {
+            log.warn("PPPoE server info not available");
+            return;
+        }
+
+        synchronized (this) {
+            pppoeConnectPoints = Sets.newConcurrentHashSet(cfg.getPppoeServerConnectPoint());
+        }
+        useOltUplink = cfg.getUseOltUplinkForServerPktInOut();
+    }
+
+    /**
+     * Returns whether the passed port is the uplink port of the olt device.
+     */
+    private boolean isUplinkPortOfOlt(DeviceId dId, Port p) {
+        log.debug("isUplinkPortOfOlt: DeviceId: {} Port: {}", dId, p);
+        if (!mastershipService.isLocalMaster(dId)) {
+            return false;
+        }
+
+        Device d = deviceService.getDevice(dId);
+        SubscriberAndDeviceInformation deviceInfo = subsService.get(d.serialNumber());
+        if (deviceInfo != null) {
+            return (deviceInfo.uplinkPort() == p.number().toLong());
+        }
+
+        return false;
+    }
+
+    /**
+     * Returns the connectPoint which is the uplink port of the OLT.
+     */
+    private ConnectPoint getUplinkConnectPointOfOlt(DeviceId dId) {
+        Device d = deviceService.getDevice(dId);
+        SubscriberAndDeviceInformation deviceInfo = subsService.get(d.serialNumber());
+        log.debug("getUplinkConnectPointOfOlt DeviceId: {} devInfo: {}", dId, deviceInfo);
+        if (deviceInfo != null) {
+            PortNumber pNum = PortNumber.portNumber(deviceInfo.uplinkPort());
+            Port port = deviceService.getPort(d.id(), pNum);
+            if (port != null) {
+                return new ConnectPoint(d.id(), pNum);
+            }
+        }
+        return null;
+    }
+
+    @Modified
+    protected void modified(ComponentContext context) {
+        Dictionary<?, ?> properties = context.getProperties();
+
+        Integer newPppoeMaxMtu = getIntegerProperty(properties, PPPOE_MAX_MTU);
+        if (newPppoeMaxMtu != null) {
+            if (newPppoeMaxMtu != pppoeMaxMtu && newPppoeMaxMtu >= 0) {
+                log.info("PPPPOE MTU modified from {} to {}", pppoeMaxMtu, newPppoeMaxMtu);
+                pppoeMaxMtu = newPppoeMaxMtu;
+            } else if (newPppoeMaxMtu < 0) {
+                log.error("Invalid newPppoeMaxMtu : {}, defaulting to 1492", newPppoeMaxMtu);
+                pppoeMaxMtu = PPPOE_MAX_MTU_DEFAULT;
+            }
+        }
+
+        Boolean newEnableCircuitIdValidation = Tools.isPropertyEnabled(properties, ENABLE_CIRCUIT_ID_VALIDATION);
+        if (newEnableCircuitIdValidation != null) {
+            if (enableCircuitIdValidation != newEnableCircuitIdValidation) {
+                log.info("Property enableCircuitIdValidation modified from {} to {}",
+                        enableCircuitIdValidation, newEnableCircuitIdValidation);
+                enableCircuitIdValidation = newEnableCircuitIdValidation;
+            }
+        }
+
+        String s = Tools.get(properties, PACKET_PROCESSOR_THREADS);
+
+        int oldpacketProcessorThreads = packetProcessorThreads;
+        packetProcessorThreads = Strings.isNullOrEmpty(s) ? oldpacketProcessorThreads
+                : Integer.parseInt(s.trim());
+        if (packetProcessorExecutor == null || oldpacketProcessorThreads != packetProcessorThreads) {
+            if (packetProcessorExecutor != null) {
+                packetProcessorExecutor.shutdown();
+            }
+            packetProcessorExecutor = newFixedThreadPool(packetProcessorThreads,
+                    groupedThreads("onos/pppoe",
+                            "pppoe-packet-%d", log));
+        }
+    }
+
+    /**
+     * Selects a connect point through an available device for which it is the master.
+     */
+    private void selectServerConnectPoint() {
+        synchronized (this) {
+            pppoeServerConnectPoint.set(null);
+            if (pppoeConnectPoints != null) {
+                // find a connect point through a device for which we are master
+                for (ConnectPoint cp: pppoeConnectPoints) {
+                    if (mastershipService.isLocalMaster(cp.deviceId())) {
+                        if (deviceService.isAvailable(cp.deviceId())) {
+                            pppoeServerConnectPoint.set(cp);
+                        }
+                        log.info("PPPOE connectPoint selected is {}", cp);
+                        break;
+                    }
+                }
+            }
+            log.info("PPPOE Server connectPoint is {}", pppoeServerConnectPoint.get());
+            if (pppoeServerConnectPoint.get() == null) {
+                log.error("Master of none, can't relay PPPOE messages to server");
+            }
+        }
+    }
+
+    private SubscriberAndDeviceInformation getSubscriber(ConnectPoint cp) {
+        Port p = deviceService.getPort(cp);
+        String subscriberId = p.annotations().value(AnnotationKeys.PORT_NAME);
+        return subsService.get(subscriberId);
+    }
+
+    private SubscriberAndDeviceInformation getDevice(PacketContext context) {
+        String serialNo = deviceService.getDevice(context.inPacket().
+                receivedFrom().deviceId()).serialNumber();
+
+        return subsService.get(serialNo);
+    }
+
+    private UniTagInformation getUnitagInformationFromPacketContext(PacketContext context,
+                                                                    SubscriberAndDeviceInformation sub) {
+        return sub.uniTagList()
+                .stream()
+                .filter(u -> u.getPonCTag().toShort() == context.inPacket().parsed().getVlanID())
+                .findFirst()
+                .orElse(null);
+    }
+
+    private boolean removeSessionsByConnectPoint(ConnectPoint cp) {
+        boolean removed = false;
+        for (MacAddress key : sessionsMap.keySet()) {
+            PppoeSessionInfo entry = sessionsMap.asJavaMap().get(key);
+            if (entry.getClientCp().equals(cp)) {
+                sessionsMap.remove(key);
+                removed = true;
+            }
+        }
+        return removed;
+    }
+
+    private boolean removeSessionsByDevice(DeviceId devid) {
+        boolean removed = false;
+        for (MacAddress key : sessionsMap.keySet()) {
+            PppoeSessionInfo entry = sessionsMap.asJavaMap().get(key);
+            if (entry.getClientCp().deviceId().equals(devid)) {
+                sessionsMap.remove(key);
+                removed = true;
+            }
+        }
+        return removed;
+    }
+
+    private class PppoeAgentPacketProcessor implements PacketProcessor {
+        @Override
+        public void process(PacketContext context) {
+            packetProcessorExecutor.execute(() -> {
+                processInternal(context);
+            });
+        }
+
+        private void processInternal(PacketContext context) {
+            Ethernet packet = (Ethernet) context.inPacket().parsed().clone();
+            if (packet.getEtherType() == Ethernet.TYPE_PPPOED) {
+                processPppoedPacket(context, packet);
+            }
+        }
+
+        private void processPppoedPacket(PacketContext context, Ethernet packet) {
+            PPPoED pppoed = (PPPoED) packet.getPayload();
+            if (pppoed == null) {
+                log.warn("PPPoED payload is null");
+                return;
+            }
+
+            final byte pppoedCode = pppoed.getCode();
+            final short sessionId = pppoed.getSessionId();
+            final MacAddress clientMacAddress;
+            final ConnectPoint packetCp = context.inPacket().receivedFrom();
+            boolean serverMessage = false;
+
+            // Get the client MAC address
+            switch (pppoedCode) {
+                case PPPOED_CODE_PADT: {
+                    if (sessionsMap.containsKey(packet.getDestinationMAC())) {
+                        clientMacAddress = packet.getDestinationMAC();
+                        serverMessage = true;
+                    } else if (sessionsMap.containsKey(packet.getSourceMAC())) {
+                        clientMacAddress = packet.getSourceMAC();
+                    } else {
+                        // In the unlikely case of receiving a PADT without an existing session
+                        log.warn("PADT received for unknown session. Dropping packet.");
+                        return;
+                    }
+                    break;
+                }
+                case PPPOED_CODE_PADI:
+                case PPPOED_CODE_PADR: {
+                    clientMacAddress = packet.getSourceMAC();
+                    break;
+                }
+                default: {
+                    clientMacAddress = packet.getDestinationMAC();
+                    serverMessage = true;
+                    break;
+                }
+            }
+
+            SubscriberAndDeviceInformation subsInfo;
+            if (serverMessage) {
+                if (!sessionsMap.containsKey(clientMacAddress)) {
+                    log.error("PPPoED message received from server without an existing session. Message not relayed.");
+                    return;
+                } else {
+                    PppoeSessionInfo sessInfo = sessionsMap.get(clientMacAddress).value();
+                    subsInfo = getSubscriber(sessInfo.getClientCp());
+                }
+            } else {
+                subsInfo = getSubscriber(packetCp);
+            }
+
+            if (subsInfo == null) {
+                log.error("No Sadis info for subscriber on connect point {}. Message not relayed.", packetCp);
+                return;
+            }
+
+            log.trace("{} received from {} at {} with client mac: {}",
+                    PPPoED.Type.getTypeByValue(pppoedCode).toString(),
+                    serverMessage ? "server" : "client", packetCp, clientMacAddress);
+
+            if (log.isTraceEnabled()) {
+                log.trace("PPPoED message received from {} at {} {}",
+                        serverMessage ? "server" : "client", packetCp, packet);
+            }
+
+            // In case of PADI, force the removal of the previous session entry
+            if ((pppoedCode == PPPOED_CODE_PADI) && (sessionsMap.containsKey(clientMacAddress))) {
+                log.trace("PADI received from MAC: {} with existing session data. Removing the existing data.",
+                        clientMacAddress.toString());
+                sessionsMap.remove(clientMacAddress);
+            }
+
+            // Fill the session map entry
+            PppoeSessionInfo sessionInfo;
+            if (!sessionsMap.containsKey(clientMacAddress)) {
+                if (!serverMessage)  {
+                    ConnectPoint serverCp = getServerConnectPoint(packetCp.deviceId());
+                    SubscriberAndDeviceInformation subscriber = getSubscriber(packetCp);
+                    sessionInfo = new PppoeSessionInfo(packetCp, serverCp, pppoedCode,
+                            sessionId, subscriber, clientMacAddress);
+                    sessionsMap.put(clientMacAddress, sessionInfo);
+                } else {
+                    // This case was already covered.
+                    return;
+                }
+            } else {
+                sessionInfo = sessionsMap.get(clientMacAddress).value();
+            }
+
+            switch (pppoedCode) {
+                case PPPOED_CODE_PADI:
+                case PPPOED_CODE_PADR:
+                    updatePppoeAgentCountersStore(sessionInfo.getSubscriber(),
+                            pppoedCode == PPPOED_CODE_PADI ? PppoeAgentCounterNames.PADI : PppoeAgentCounterNames.PADR);
+
+                    Ethernet padir = processPacketFromClient(context, packet, pppoed, sessionInfo, clientMacAddress);
+                    if (padir != null) {
+                        if (padir.serialize().length <= pppoeMaxMtu) {
+                            forwardPacketToServer(padir, sessionInfo);
+                        } else {
+                            log.debug("MTU message size: {} exceeded configured pppoeMaxMtu: {}. Dropping Packet.",
+                                    padir.serialize().length, pppoeMaxMtu);
+                            forwardPacketToClient(errorToClient(packet, pppoed, "MTU message size exceeded"),
+                                                                sessionInfo, clientMacAddress);
+                            updatePppoeAgentCountersStore(sessionInfo.getSubscriber(),
+                                                          PppoeAgentCounterNames.MTU_EXCEEDED);
+                        }
+                    }
+                    break;
+                case PPPOED_CODE_PADO:
+                case PPPOED_CODE_PADS:
+                    updatePppoeAgentCountersStore(sessionInfo.getSubscriber(),
+                            pppoedCode == PPPOED_CODE_PADO ? PppoeAgentCounterNames.PADO : PppoeAgentCounterNames.PADS);
+                    Ethernet pados = processPacketFromServer(packet, pppoed, sessionInfo, clientMacAddress);
+                    if (pados != null) {
+                        forwardPacketToClient(pados, sessionInfo, clientMacAddress);
+                    }
+                    break;
+                case PPPOED_CODE_PADT:
+                    if (serverMessage) {
+                        updatePppoeAgentCountersStore(sessionInfo.getSubscriber(),
+                                                      PppoeAgentCounterNames.PADT_FROM_SERVER);
+                        forwardPacketToClient(packet, sessionInfo, clientMacAddress);
+                    } else {
+                        updatePppoeAgentCountersStore(sessionInfo.getSubscriber(),
+                                                      PppoeAgentCounterNames.PADT_FROM_CLIENT);
+                        forwardPacketToServer(packet, sessionInfo);
+                    }
+
+                    String reason = "";
+                    PPPoEDTag genericErrorTag = pppoed.getTags()
+                            .stream()
+                            .filter(tag -> tag.getType() == PPPOED_TAG_GENERIC_ERROR)
+                            .findFirst()
+                            .orElse(null);
+
+                    if (genericErrorTag != null) {
+                        reason = new String(genericErrorTag.getValue(), StandardCharsets.UTF_8);
+                    }
+                    log.debug("PADT sessionId:{}  client MAC:{}  Terminate reason:{}.",
+                            Integer.toHexString(sessionId & 0xFFFF), clientMacAddress, reason);
+
+                    boolean knownSessionId = sessionInfo.getSessionId() == sessionId;
+                    if (knownSessionId) {
+                        PppoeSessionInfo removedSessionInfo = Versioned
+                                .valueOrNull(sessionsMap.remove(clientMacAddress));
+                        if (removedSessionInfo != null) {
+                            post(new PppoeAgentEvent(PppoeAgentEvent.Type.TERMINATE, removedSessionInfo,
+                                                     packetCp, clientMacAddress, reason));
+                        }
+                    } else {
+                        log.warn("PADT received for a known MAC address but for unknown session.");
+                    }
+                    break;
+                default:
+                    break;
+            }
+        }
+
+        private Ethernet processPacketFromClient(PacketContext context, Ethernet ethernetPacket, PPPoED pppoed,
+                                                 PppoeSessionInfo sessionInfo, MacAddress clientMacAddress) {
+            byte pppoedCode = pppoed.getCode();
+
+            sessionInfo.setPacketCode(pppoedCode);
+            sessionsMap.put(clientMacAddress, sessionInfo);
+
+            // Update Counters
+            for (PPPoEDTag tag : pppoed.getTags()) {
+                if (tag.getType() == PPPOED_TAG_GENERIC_ERROR) {
+                    updatePppoeAgentCountersStore(sessionInfo.getSubscriber(),
+                            PppoeAgentCounterNames.GENERIC_ERROR_FROM_CLIENT);
+                    break;
+                }
+            }
+
+            Ethernet ethFwd = ethernetPacket;
+
+            // If this is a PADI packet, there'll be a START event.
+            if (pppoedCode == PPPOED_CODE_PADI) {
+                post(new PppoeAgentEvent(PppoeAgentEvent.Type.START, sessionInfo, sessionInfo.getClientCp(),
+                        clientMacAddress));
+            }
+
+            // Creates the vendor specific tag.
+            String circuitId = getCircuitId(sessionInfo.getClientCp());
+            if (circuitId == null) {
+                log.error("Failed to build circuid-id for client on connect point {}. Dropping packet.",
+                        sessionInfo.getClientCp());
+                return null;
+            }
+
+            // Checks whether the circuit-id is valid, if it's not it drops the packet.
+            if (!isCircuitIdValid(circuitId, sessionInfo.getSubscriber())) {
+                log.warn("Invalid circuit ID, dropping packet.");
+                PppoeAgentEvent invalidCidEvent = new PppoeAgentEvent(PppoeAgentEvent.Type.INVALID_CID, sessionInfo,
+                        context.inPacket().receivedFrom(), clientMacAddress);
+                post(invalidCidEvent);
+                return null;
+            }
+
+            String remoteId = sessionInfo.getSubscriber().remoteId();
+            byte[] vendorSpecificTag = new PPPoEDVendorSpecificTag(circuitId, remoteId).toByteArray();
+
+            // According to TR-101, R-149 (by Broadband Forum), agent must REPLACE vendor-specific tag that may come
+            // from client message with its own tag.
+            // The following block ensures that agent removes any previous vendor-specific tag.
+            List<PPPoEDTag> originalTags = pppoed.getTags();
+            if (originalTags != null) {
+                PPPoEDTag originalVendorSpecificTag = originalTags.stream()
+                        .filter(tag -> tag.getType() == PPPOED_TAG_VENDOR_SPECIFIC)
+                        .findFirst()
+                        .orElse(null);
+
+                if (originalVendorSpecificTag != null) {
+                    int tagToRemoveLength = originalVendorSpecificTag.getLength();
+                    originalTags.removeIf(tag -> tag.getType() == PPPOED_TAG_VENDOR_SPECIFIC);
+                    pppoed.setPayloadLength((short) (pppoed.getPayloadLength() - tagToRemoveLength));
+                }
+            }
+
+            pppoed.setTag(PPPOED_TAG_VENDOR_SPECIFIC, vendorSpecificTag);
+
+            ethFwd.setPayload(pppoed);
+            ethFwd.setQinQTPID(Ethernet.TYPE_VLAN);
+
+            UniTagInformation uniTagInformation = getUnitagInformationFromPacketContext(context,
+                    sessionInfo.getSubscriber());
+            if (uniTagInformation == null) {
+                log.warn("Missing service information for connectPoint {} / cTag {}",
+                        context.inPacket().receivedFrom(),  context.inPacket().parsed().getVlanID());
+                return null;
+            }
+            ethFwd.setQinQVID(uniTagInformation.getPonSTag().toShort());
+            ethFwd.setPad(true);
+            return ethFwd;
+        }
+
+        private Ethernet processPacketFromServer(Ethernet ethernetPacket, PPPoED pppoed,
+                                                 PppoeSessionInfo sessionInfo, MacAddress clientMacAddress) {
+            // Update counters
+            List<PPPoEDTag> tags = pppoed.getTags();
+            for (PPPoEDTag tag : tags) {
+                switch (tag.getType()) {
+                    case PPPOED_TAG_GENERIC_ERROR:
+                        updatePppoeAgentCountersStore(sessionInfo.getSubscriber(),
+                                PppoeAgentCounterNames.GENERIC_ERROR_FROM_SERVER);
+                        break;
+                    case PPPOED_TAG_SERVICE_NAME_ERROR:
+                        updatePppoeAgentCountersStore(sessionInfo.getSubscriber(),
+                                PppoeAgentCounterNames.SERVICE_NAME_ERROR);
+                        break;
+                    case PPPOED_TAG_AC_SYSTEM_ERROR:
+                        updatePppoeAgentCountersStore(sessionInfo.getSubscriber(),
+                                PppoeAgentCounterNames.AC_SYSTEM_ERROR);
+                        break;
+                    default:
+                        break;
+                }
+            }
+
+            byte pppoedCode = pppoed.getCode();
+
+            if (pppoedCode == PPPOED_CODE_PADS) {
+                log.debug("PADS sessionId:{}  client MAC:{}", Integer.toHexString(pppoed.getSessionId() & 0xFFFF),
+                        clientMacAddress);
+                sessionInfo.setSessionId(pppoed.getSessionId());
+            }
+            sessionInfo.setPacketCode(pppoedCode);
+            sessionsMap.put(clientMacAddress, sessionInfo);
+
+            PppoeAgentEvent.Type eventType = pppoedCode == PPPOED_CODE_PADS ?
+                    PppoeAgentEvent.Type.SESSION_ESTABLISHED :
+                    PppoeAgentEvent.Type.NEGOTIATION;
+
+            post(new PppoeAgentEvent(eventType, sessionInfo, sessionInfo.getClientCp(), clientMacAddress));
+
+            ethernetPacket.setQinQVID(QINQ_VID_NONE);
+            ethernetPacket.setPad(true);
+            return ethernetPacket;
+        }
+
+        private void updatePppoeAgentCountersStore(SubscriberAndDeviceInformation sub,
+                                                   PppoeAgentCounterNames counterType) {
+            // Update global counter stats
+            pppoeAgentCounters.incrementCounter(PppoeAgentEvent.GLOBAL_COUNTER, counterType);
+            if (sub == null) {
+                log.warn("Counter not updated as subscriber info not found.");
+            } else {
+                // Update subscriber counter stats
+                pppoeAgentCounters.incrementCounter(sub.id(), counterType);
+            }
+        }
+
+        private String getCircuitId(ConnectPoint cp) {
+            return new CircuitIdBuilder()
+                    .setConnectPoint(cp)
+                    .setDeviceService(deviceService)
+                    .setSubsService(subsService)
+                    .setCircuitIdConfig(circuitIdfields)
+                    .addCustomSeparator(CircuitIdFieldName.AcessNodeIdentifier, " ")
+                    .addCustomSeparator(CircuitIdFieldName.Port, ":")
+                    .build();
+        }
+
+        protected ConnectPoint getServerConnectPoint(DeviceId deviceId) {
+            ConnectPoint serverCp;
+            if (!useOltUplink) {
+                serverCp = pppoeServerConnectPoint.get();
+            } else {
+                serverCp = getUplinkConnectPointOfOlt(deviceId);
+            }
+            return serverCp;
+        }
+
+        private boolean isCircuitIdValid(String cId, SubscriberAndDeviceInformation entry) {
+            if (!enableCircuitIdValidation) {
+                log.debug("Circuit ID validation is disabled.");
+                return true;
+            }
+
+            if (entry == null) {
+                log.error("SubscriberAndDeviceInformation cannot be null.");
+                return false;
+            }
+
+            if (entry.circuitId() == null || entry.circuitId().isEmpty()) {
+                log.debug("Circuit ID not configured in SADIS entry. No check is done.");
+                return true;
+            } else {
+                if (cId.equals(entry.circuitId())) {
+                    log.info("Circuit ID in packet: {} matched the configured entry on SADIS.", cId);
+                    return true;
+                } else {
+                    log.warn("Circuit ID in packet: {} did not match the configured entry on SADIS: {}.",
+                            cId, entry.circuitId());
+                    return false;
+                }
+            }
+        }
+
+        private void forwardPacketToServer(Ethernet packet, PppoeSessionInfo sessionInfo) {
+            ConnectPoint toSendTo = sessionInfo.getServerCp();
+            if (toSendTo != null) {
+                log.info("Sending PPPOE packet to server at {}", toSendTo);
+                TrafficTreatment t = DefaultTrafficTreatment.builder().setOutput(toSendTo.port()).build();
+                OutboundPacket o = new DefaultOutboundPacket(toSendTo.deviceId(), t,
+                        ByteBuffer.wrap(packet.serialize()));
+                if (log.isTraceEnabled()) {
+                    log.trace("Relaying packet to pppoe server at {} {}", toSendTo, packet);
+                }
+                packetService.emit(o);
+
+                updatePppoeAgentCountersStore(sessionInfo.getSubscriber(),
+                        PppoeAgentCounterNames.PPPOED_PACKETS_TO_SERVER);
+            } else {
+                log.error("No connect point to send msg to PPPOE Server");
+            }
+        }
+
+        private void forwardPacketToClient(Ethernet packet, PppoeSessionInfo sessionInfo, MacAddress clientMacAddress) {
+            ConnectPoint subCp = sessionInfo.getClientCp();
+            if (subCp == null) {
+                log.error("Dropping PPPOE packet, can't find a connectpoint for MAC {}", clientMacAddress);
+                return;
+            }
+
+            log.info("Sending PPPOE packet to client at {}", subCp);
+            TrafficTreatment t = DefaultTrafficTreatment.builder()
+                    .setOutput(subCp.port()).build();
+            OutboundPacket o = new DefaultOutboundPacket(
+                    subCp.deviceId(), t, ByteBuffer.wrap(packet.serialize()));
+            if (log.isTraceEnabled()) {
+                log.trace("Relaying packet to pppoe client at {} {}", subCp, packet);
+            }
+
+            packetService.emit(o);
+
+            updatePppoeAgentCountersStore(sessionInfo.getSubscriber(),
+                    PppoeAgentCounterNames.PPPOED_PACKETS_FROM_SERVER);
+        }
+
+        private Ethernet errorToClient(Ethernet packet, PPPoED p, String errString) {
+            PPPoED err = new PPPoED();
+            err.setVersion(p.getVersion());
+            err.setType(p.getType());
+            switch (p.getCode()) {
+                case PPPOED_CODE_PADI:
+                    err.setCode(PPPOED_CODE_PADO);
+                    break;
+                case PPPOED_CODE_PADR:
+                    err.setCode(PPPOED_CODE_PADS);
+                    break;
+                default:
+                    break;
+            }
+            err.setCode(p.getCode());
+            err.setSessionId(p.getSessionId());
+            err.setTag(PPPOED_TAG_GENERIC_ERROR, errString.getBytes(StandardCharsets.UTF_8));
+
+            Ethernet ethPacket = new Ethernet();
+            ethPacket.setPayload(err);
+            ethPacket.setSourceMACAddress(packet.getDestinationMACAddress());
+            ethPacket.setDestinationMACAddress(packet.getSourceMACAddress());
+            ethPacket.setQinQVID(QINQ_VID_NONE);
+            ethPacket.setPad(true);
+
+            return ethPacket;
+        }
+    }
+
+    private class InternalConfigListener implements NetworkConfigListener {
+        @Override
+        public void event(NetworkConfigEvent event) {
+
+            if ((event.type() == NetworkConfigEvent.Type.CONFIG_ADDED ||
+                    event.type() == NetworkConfigEvent.Type.CONFIG_UPDATED) &&
+                    event.configClass().equals(PppoeAgentConfig.class)) {
+                updateConfig();
+                log.info("Reconfigured");
+            }
+        }
+    }
+
+    /**
+     * Handles Mastership changes for the devices which connect to the PPPOE server.
+     */
+    private class InnerMastershipListener implements MastershipListener {
+        @Override
+        public void event(MastershipEvent event) {
+            if (!useOltUplink) {
+                if (pppoeServerConnectPoint.get() != null &&
+                        pppoeServerConnectPoint.get().deviceId().equals(event.subject())) {
+                    log.trace("Mastership Event recevived for {}", event.subject());
+                    // mastership of the device for our connect point has changed, reselect
+                    selectServerConnectPoint();
+                }
+            }
+        }
+    }
+
+    private class InnerDeviceListener implements DeviceListener {
+        @Override
+        public void event(DeviceEvent event) {
+            DeviceId devId = event.subject().id();
+
+            if (log.isTraceEnabled() && !event.type().equals(DeviceEvent.Type.PORT_STATS_UPDATED)) {
+                log.trace("Device Event received for {} event {}", event.subject(), event.type());
+            }
+
+            // Handle events from any other device
+            switch (event.type()) {
+                case PORT_UPDATED:
+                    if (!event.port().isEnabled()) {
+                        ConnectPoint cp = new ConnectPoint(devId, event.port().number());
+                        removeSessionsByConnectPoint(cp);
+                    }
+                    break;
+                case PORT_REMOVED:
+                    // Remove all entries related to this port from sessions map
+                    ConnectPoint cp = new ConnectPoint(devId, event.port().number());
+                    removeSessionsByConnectPoint(cp);
+                    break;
+                case DEVICE_REMOVED:
+                    // Remove all entries related to this device from sessions map
+                    removeSessionsByDevice(devId);
+                    break;
+                default:
+                    break;
+            }
+        }
+    }
+
+    private class InnerPppoeAgentStoreDelegate implements PppoeAgentStoreDelegate {
+        @Override
+        public void notify(PppoeAgentEvent event) {
+            if (event.type().equals(PppoeAgentEvent.Type.STATS_UPDATE)) {
+                PppoeAgentEvent toPost = event;
+                if (event.getSubscriberId() != null) {
+                    // infuse the event with the allocation info before posting
+                    PppoeSessionInfo info = Versioned.valueOrNull(
+                            sessionsMap.stream().filter(entry -> event.getSubscriberId()
+                                    .equals(entry.getValue().value().getSubscriber().id()))
+                                    .map(Map.Entry::getValue)
+                                    .findFirst()
+                                    .orElse(null));
+                    if (info == null) {
+                        log.debug("Not handling STATS_UPDATE event for session that no longer exists. {}.", event);
+                        return;
+                    }
+
+                    toPost = new PppoeAgentEvent(event.type(), info, event.getCounterName(), event.getCounterValue(),
+                                                 info.getClientMac(), event.getSubscriberId());
+                }
+                post(toPost);
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentConfig.java b/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentConfig.java
new file mode 100644
index 0000000..5d14c76
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentConfig.java
@@ -0,0 +1,86 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.impl;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.google.common.collect.ImmutableSet;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.net.ConnectPoint;
+import org.onosproject.net.config.Config;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class PppoeAgentConfig extends Config<ApplicationId> {
+    private static final String PPPOE_CONNECT_POINTS = "pppoeServerConnectPoints";
+    private static final String USE_OLT_ULPORT_FOR_PKT_INOUT = "useOltUplinkForServerPktInOut";
+
+    protected static final Boolean DEFAULT_USE_OLT_ULPORT_FOR_PKT_INOUT = true;
+
+    @Override
+    public boolean isValid() {
+        return hasOnlyFields(PPPOE_CONNECT_POINTS, USE_OLT_ULPORT_FOR_PKT_INOUT);
+    }
+
+    /**
+     * Returns whether the app would use the uplink port of OLT for sending/receving
+     * messages to/from the server.
+     *
+     * @return true if OLT uplink port is to be used, false otherwise
+     */
+    public boolean getUseOltUplinkForServerPktInOut() {
+        if (object == null) {
+            return DEFAULT_USE_OLT_ULPORT_FOR_PKT_INOUT;
+        }
+        if (!object.has(USE_OLT_ULPORT_FOR_PKT_INOUT)) {
+            return DEFAULT_USE_OLT_ULPORT_FOR_PKT_INOUT;
+        }
+        return object.path(USE_OLT_ULPORT_FOR_PKT_INOUT).asBoolean();
+    }
+
+    /**
+     * Returns the pppoe server connect points.
+     *
+     * @return pppoe server connect points
+     */
+    public Set<ConnectPoint> getPppoeServerConnectPoint() {
+        if (object == null) {
+            return new HashSet<ConnectPoint>();
+        }
+
+        if (!object.has(PPPOE_CONNECT_POINTS)) {
+            return ImmutableSet.of();
+        }
+
+        ImmutableSet.Builder<ConnectPoint> builder = ImmutableSet.builder();
+        ArrayNode arrayNode = (ArrayNode) object.path(PPPOE_CONNECT_POINTS);
+        for (JsonNode jsonNode : arrayNode) {
+            String portName = jsonNode.asText(null);
+            if (portName == null) {
+                return null;
+            }
+            try {
+                builder.add(ConnectPoint.deviceConnectPoint(portName));
+            } catch (IllegalArgumentException e) {
+                return null;
+            }
+        }
+        return builder.build();
+    }
+
+
+}
diff --git a/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentCounterNames.java b/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentCounterNames.java
new file mode 100644
index 0000000..0d58078
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentCounterNames.java
@@ -0,0 +1,91 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.impl;
+
+import com.google.common.collect.ImmutableSet;
+
+import java.util.Set;
+
+/**
+ * Represents PPPoE agent counters type.
+ */
+public enum PppoeAgentCounterNames {
+    /**
+     * Number of PADI messages received from client.
+     */
+    PADI,
+    /**
+     * Number of PADO messages received from server.
+     */
+    PADO,
+    /**
+     * Number of PADR messages received from client.
+     */
+    PADR,
+    /**
+     * Number of PADS messages received from server.
+     */
+    PADS,
+    /**
+     * Number of PADT messages received from server.
+     */
+    PADT_FROM_SERVER,
+    /**
+     * Number of PADT messages received from client.
+     */
+    PADT_FROM_CLIENT,
+    /**
+     * Number of PPPoED messages sent to server.
+     */
+    PPPOED_PACKETS_TO_SERVER,
+    /**
+     * Number  of PPPoED messages received from server.
+     */
+    PPPOED_PACKETS_FROM_SERVER,
+    /**
+     * Number of MTU Exceeded errors generated by the PPPoED agent.
+     */
+    MTU_EXCEEDED,
+    /**
+     * Number of Generic Errors received from server.
+     */
+    GENERIC_ERROR_FROM_SERVER,
+    /**
+     * Number of Generic Errors received from client.
+     */
+    GENERIC_ERROR_FROM_CLIENT,
+    /**
+     * Number of ServiceName Errors received from server.
+     */
+    SERVICE_NAME_ERROR,
+    /**
+     * Number of AC-System Errors received from server.
+     */
+    AC_SYSTEM_ERROR;
+
+    /**
+     * Supported types of PPPoED agent counters.
+     */
+    public  static final Set<PppoeAgentCounterNames> SUPPORTED_COUNTERS = ImmutableSet.of(
+            PppoeAgentCounterNames.PADI, PppoeAgentCounterNames.PADO,
+            PppoeAgentCounterNames.PADR, PppoeAgentCounterNames.PADS,
+            PppoeAgentCounterNames.PADT_FROM_SERVER, PppoeAgentCounterNames.PADT_FROM_CLIENT,
+            PppoeAgentCounterNames.PPPOED_PACKETS_TO_SERVER, PppoeAgentCounterNames.PPPOED_PACKETS_FROM_SERVER,
+            PppoeAgentCounterNames.MTU_EXCEEDED, PppoeAgentCounterNames.GENERIC_ERROR_FROM_SERVER,
+            PppoeAgentCounterNames.GENERIC_ERROR_FROM_CLIENT, PppoeAgentCounterNames.SERVICE_NAME_ERROR,
+            PppoeAgentCounterNames.AC_SYSTEM_ERROR);
+}
\ No newline at end of file
diff --git a/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentCountersIdentifier.java b/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentCountersIdentifier.java
new file mode 100644
index 0000000..cca163f
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentCountersIdentifier.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.impl;
+
+import org.opencord.pppoeagent.PppoeAgentEvent;
+
+import java.util.Objects;
+
+/**
+ * Represents PPPoED agent counters identifier.
+ */
+public final class PppoeAgentCountersIdentifier {
+    final String counterClassKey;
+    final Enum<PppoeAgentCounterNames> counterTypeKey;
+
+    /**
+     * Creates a default global counter identifier for a given counterType.
+     *
+     * @param counterTypeKey Identifies the supported type of pppoe agent counters
+     */
+    public PppoeAgentCountersIdentifier(PppoeAgentCounterNames counterTypeKey) {
+        this.counterClassKey = PppoeAgentEvent.GLOBAL_COUNTER;
+        this.counterTypeKey = counterTypeKey;
+    }
+
+    /**
+     * Creates a counter identifier. A counter is defined by the key pair [counterClass, counterType],
+     * where counterClass can be global or the subscriber ID and counterType is the supported pppoe counter.
+     *
+     * @param counterClassKey Identifies which class the counter is assigned (global or per subscriber)
+     * @param counterTypeKey Identifies the supported type of pppoed relay counters
+     */
+    public PppoeAgentCountersIdentifier(String counterClassKey, PppoeAgentCounterNames counterTypeKey) {
+        this.counterClassKey = counterClassKey;
+        this.counterTypeKey = counterTypeKey;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj instanceof PppoeAgentCountersIdentifier) {
+            final PppoeAgentCountersIdentifier other = (PppoeAgentCountersIdentifier) obj;
+            return Objects.equals(this.counterClassKey, other.counterClassKey)
+                    && Objects.equals(this.counterTypeKey, other.counterTypeKey);
+        }
+
+        return false;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(counterClassKey, counterTypeKey);
+    }
+
+    @Override
+    public String toString() {
+        return "PppoeAgentCountersIdentifier{" +
+                "counterClassKey='" + counterClassKey + '\'' +
+                ", counterTypeKey=" + counterTypeKey +
+                '}';
+    }
+}
\ No newline at end of file
diff --git a/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentCountersStore.java b/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentCountersStore.java
new file mode 100644
index 0000000..5b4ddfd
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentCountersStore.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.impl;
+
+import org.onosproject.store.Store;
+import org.opencord.pppoeagent.PppoeAgentEvent;
+import org.opencord.pppoeagent.PppoeAgentStoreDelegate;
+
+/**
+ * Represents a stored Pppoe Agent Counters. A counter entry is defined by the pair [counterClass, counterType],
+ * where counterClass can be maybe global or subscriber ID and counterType is the pppoe counter.
+ */
+public interface PppoeAgentCountersStore extends Store<PppoeAgentEvent, PppoeAgentStoreDelegate> {
+
+    String NAME = "PPPOE_Agent_stats";
+
+    /**
+     * Creates or updates PPPOE Agent counter.
+     *
+     * @param counterClass class of counters (global, per subscriber).
+     * @param counterType name of counter
+     */
+    void incrementCounter(String counterClass, PppoeAgentCounterNames counterType);
+
+    /**
+     * Sets the value of a PPPOE Agent counter.
+     *
+     * @param counterClass class of counters (global, per subscriber).
+     * @param counterType name of counter
+     * @param value The value of the counter
+     */
+    void setCounter(String counterClass, PppoeAgentCounterNames counterType, Long value);
+
+    /**
+     * Gets the current PPPoE Agent counter values.
+     *
+     * @return PPPoE Agent counter values
+     */
+    PppoeAgentStatistics getCounters();
+
+    /**
+     * Resets counter values for a given counter class.
+     *
+     * @param counterClass class of counters (global, per subscriber).
+     */
+    void resetCounters(String counterClass);
+}
\ No newline at end of file
diff --git a/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentStatistics.java b/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentStatistics.java
new file mode 100644
index 0000000..dce1ac7
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/impl/PppoeAgentStatistics.java
@@ -0,0 +1,89 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.impl;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Sets;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Snapshot of PPPoE Agent statistics.
+ */
+public class PppoeAgentStatistics {
+    private final ImmutableMap<PppoeAgentCountersIdentifier, Long> counters;
+    private PppoeAgentStatistics(ImmutableMap<PppoeAgentCountersIdentifier, Long> counters) {
+        this.counters = counters;
+    }
+    /**
+     * Creates a new empty statistics instance.
+     */
+    public PppoeAgentStatistics() {
+        counters = ImmutableMap.of();
+    }
+    /**
+     * Gets the value of the counter with the given ID. Defaults to 0 if counter is not present.
+     *
+     * @param id counter ID
+     * @return counter value
+     */
+    public long get(PppoeAgentCountersIdentifier id) {
+        return counters.getOrDefault(id, 0L);
+    }
+    /**
+     * Gets the map of counters.
+     *
+     * @return map of counters
+     */
+    public Map<PppoeAgentCountersIdentifier, Long> counters() {
+        return counters;
+    }
+    /**
+     * Creates a new statistics instance with the given counter values.
+     *
+     * @param counters counters
+     * @return statistics
+     */
+    public static PppoeAgentStatistics withCounters(Map<PppoeAgentCountersIdentifier, Long> counters) {
+        ImmutableMap.Builder<PppoeAgentCountersIdentifier, Long> builder = ImmutableMap.builder();
+        counters.forEach(builder::put);
+        return new PppoeAgentStatistics(builder.build());
+    }
+    /**
+     * Adds the given statistics instance to this one (sums the common counters) and returns
+     * a new instance containing the result.
+     *
+     * @param other other instance
+     * @return result
+     */
+    public PppoeAgentStatistics add(PppoeAgentStatistics other) {
+        ImmutableMap.Builder<PppoeAgentCountersIdentifier, Long> builder = ImmutableMap.builder();
+        Set<PppoeAgentCountersIdentifier> keys = Sets.newHashSet(other.counters.keySet());
+        counters.forEach((id, value) -> {
+            builder.put(id, value + other.counters.getOrDefault(id, 0L));
+            keys.remove(id);
+        });
+        keys.forEach(i -> builder.put(i, other.counters.get(i)));
+        return new PppoeAgentStatistics(builder.build());
+    }
+    @Override
+    public String toString() {
+        MoreObjects.ToStringHelper helper = MoreObjects.toStringHelper(this.getClass());
+        counters.forEach((id, v) -> helper.add(id.toString(), v));
+        return helper.toString();
+    }
+}
\ No newline at end of file
diff --git a/app/src/main/java/org/opencord/pppoeagent/impl/SimplePppoeAgentCountersStore.java b/app/src/main/java/org/opencord/pppoeagent/impl/SimplePppoeAgentCountersStore.java
new file mode 100644
index 0000000..8be88b7
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/impl/SimplePppoeAgentCountersStore.java
@@ -0,0 +1,305 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.impl;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableMap;
+
+import org.onlab.util.KryoNamespace;
+import org.onlab.util.SafeRecurringTask;
+import org.onlab.util.Tools;
+import org.onosproject.cfg.ComponentConfigService;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.LeadershipService;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.store.AbstractStore;
+import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
+import org.onosproject.store.cluster.messaging.ClusterMessage;
+import org.onosproject.store.cluster.messaging.MessageSubject;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.EventuallyConsistentMap;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.WallClockTimestamp;
+import org.opencord.pppoeagent.PppoeAgentEvent;
+import org.opencord.pppoeagent.PppoeAgentStoreDelegate;
+import org.osgi.service.component.ComponentContext;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Modified;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
+
+import org.slf4j.Logger;
+import java.nio.charset.StandardCharsets;
+
+import java.util.Dictionary;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.opencord.pppoeagent.impl.OsgiPropertyConstants.PUBLISH_COUNTERS_RATE;
+import static org.opencord.pppoeagent.impl.OsgiPropertyConstants.PUBLISH_COUNTERS_RATE_DEFAULT;
+import static org.opencord.pppoeagent.impl.OsgiPropertyConstants.SYNC_COUNTERS_RATE;
+import static org.opencord.pppoeagent.impl.OsgiPropertyConstants.SYNC_COUNTERS_RATE_DEFAULT;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * PPPoE Agent Counters Manager Component.
+ */
+@Component(immediate = true,
+        property = {
+                PUBLISH_COUNTERS_RATE + ":Integer=" + PUBLISH_COUNTERS_RATE_DEFAULT,
+                SYNC_COUNTERS_RATE + ":Integer=" + SYNC_COUNTERS_RATE_DEFAULT,
+        }
+)
+public class SimplePppoeAgentCountersStore extends AbstractStore<PppoeAgentEvent, PppoeAgentStoreDelegate>
+        implements PppoeAgentCountersStore {
+    private static final String PPPOE_STATISTICS_LEADERSHIP = "pppoeagent-statistics";
+    private static final MessageSubject RESET_SUBJECT = new MessageSubject("pppoeagent-statistics-reset");
+
+    private final Logger log = getLogger(getClass());
+    private ConcurrentMap<PppoeAgentCountersIdentifier, Long> countersMap;
+
+    private EventuallyConsistentMap<NodeId, PppoeAgentStatistics> statistics;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected StorageService storageService;
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected ClusterService clusterService;
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected LeadershipService leadershipService;
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected ComponentConfigService componentConfigService;
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected ClusterCommunicationService clusterCommunicationService;
+    protected int publishCountersRate = PUBLISH_COUNTERS_RATE_DEFAULT;
+    protected int syncCountersRate = SYNC_COUNTERS_RATE_DEFAULT;
+    KryoNamespace serializer = KryoNamespace.newBuilder()
+            .register(KryoNamespaces.API)
+            .register(PppoeAgentStatistics.class)
+            .register(PppoeAgentCountersIdentifier.class)
+            .register(PppoeAgentCounterNames.class)
+            .register(ClusterMessage.class)
+            .register(MessageSubject.class)
+            .build();
+    private ScheduledExecutorService executor;
+    private ScheduledFuture<?> publisherTask;
+    private ScheduledFuture<?> syncTask;
+    private AtomicBoolean dirty = new AtomicBoolean(true);
+
+    @Activate
+    public void activate(ComponentContext context) {
+        log.info("Activate PPPoE Agent Counters Manager");
+        countersMap = new ConcurrentHashMap<>();
+        componentConfigService.registerProperties(getClass());
+        modified(context);
+        statistics = storageService.<NodeId, PppoeAgentStatistics>eventuallyConsistentMapBuilder()
+                .withName("pppoeagent-statistics")
+                .withSerializer(serializer)
+                .withTimestampProvider((k, v) -> new WallClockTimestamp())
+                .build();
+        // Initialize counter values for the global counters
+        initCounters(PppoeAgentEvent.GLOBAL_COUNTER, statistics.get(clusterService.getLocalNode().id()));
+        syncStats();
+        leadershipService.runForLeadership(PPPOE_STATISTICS_LEADERSHIP);
+        executor = Executors.newScheduledThreadPool(1);
+        clusterCommunicationService.addSubscriber(RESET_SUBJECT, Serializer.using(serializer)::decode,
+                this::resetLocal, executor);
+        startSyncTask();
+        startPublishTask();
+    }
+
+    @Deactivate
+    public void deactivate() {
+        clusterCommunicationService.removeSubscriber(RESET_SUBJECT);
+        leadershipService.withdraw(PPPOE_STATISTICS_LEADERSHIP);
+        stopPublishTask();
+        stopSyncTask();
+        executor.shutdownNow();
+        componentConfigService.unregisterProperties(getClass(), false);
+    }
+    @Modified
+    public void modified(ComponentContext context) {
+        Dictionary<String, Object> properties = context.getProperties();
+        String s = Tools.get(properties, PUBLISH_COUNTERS_RATE);
+        int oldPublishCountersRate = publishCountersRate;
+        publishCountersRate = Strings.isNullOrEmpty(s) ? PUBLISH_COUNTERS_RATE_DEFAULT
+                : Integer.parseInt(s.trim());
+        if (oldPublishCountersRate != publishCountersRate) {
+            stopPublishTask();
+            startPublishTask();
+        }
+        s = Tools.get(properties, SYNC_COUNTERS_RATE);
+        int oldSyncCountersRate = syncCountersRate;
+        syncCountersRate = Strings.isNullOrEmpty(s) ? SYNC_COUNTERS_RATE_DEFAULT
+                : Integer.parseInt(s.trim());
+        if (oldSyncCountersRate != syncCountersRate) {
+            stopSyncTask();
+            startSyncTask();
+        }
+    }
+    private ScheduledFuture<?> startTask(Runnable r, int rate) {
+        return executor.scheduleAtFixedRate(SafeRecurringTask.wrap(r),
+                0, rate, TimeUnit.SECONDS);
+    }
+    private void stopTask(ScheduledFuture<?> task) {
+        task.cancel(true);
+    }
+    private void startSyncTask() {
+        syncTask = startTask(this::syncStats, syncCountersRate);
+    }
+    private void stopSyncTask() {
+        stopTask(syncTask);
+    }
+    private void startPublishTask() {
+        publisherTask = startTask(this::publishStats, publishCountersRate);
+    }
+    private void stopPublishTask() {
+        stopTask(publisherTask);
+    }
+
+    ImmutableMap<PppoeAgentCountersIdentifier, Long> getCountersMap() {
+        return ImmutableMap.copyOf(countersMap);
+    }
+
+    public PppoeAgentStatistics getCounters() {
+        return aggregate();
+    }
+
+    /**
+     * Initialize the supported counters map for the given counter class.
+     * @param counterClass class of counters (global, per subscriber)
+     * @param existingStats existing values to intialise the counters to
+     */
+    public void initCounters(String counterClass, PppoeAgentStatistics existingStats) {
+        checkNotNull(counterClass, "counter class can't be null");
+        for (PppoeAgentCounterNames counterType : PppoeAgentCounterNames.SUPPORTED_COUNTERS) {
+            PppoeAgentCountersIdentifier id = new PppoeAgentCountersIdentifier(counterClass, counterType);
+            countersMap.put(id, existingStats == null ? 0L : existingStats.get(id));
+        }
+    }
+
+    /**
+     * Inserts the counter entry if it is not already in the set otherwise increment the existing counter entry.
+     * @param counterClass class of counters (global, per subscriber)
+     * @param counterType name of counter
+     */
+    public void incrementCounter(String counterClass, PppoeAgentCounterNames counterType) {
+        checkNotNull(counterClass, "counter class can't be null");
+        if (PppoeAgentCounterNames.SUPPORTED_COUNTERS.contains(counterType)) {
+            PppoeAgentCountersIdentifier counterIdentifier =
+                    new PppoeAgentCountersIdentifier(counterClass, counterType);
+            countersMap.compute(counterIdentifier, (key, counterValue) ->
+                    (counterValue != null) ? counterValue + 1 : 1L);
+        } else {
+            log.error("Failed to increment counter class {} of type {}", counterClass, counterType);
+        }
+        dirty.set(true);
+    }
+
+    /**
+     * Reset the counters map for the given counter class.
+     * @param counterClass class of counters (global, per subscriber)
+     */
+    public void resetCounters(String counterClass) {
+        byte[] payload = counterClass.getBytes(StandardCharsets.UTF_8);
+        ClusterMessage reset = new ClusterMessage(clusterService.getLocalNode().id(), RESET_SUBJECT, payload);
+        clusterCommunicationService.broadcastIncludeSelf(reset, RESET_SUBJECT, Serializer.using(serializer)::encode);
+    }
+    private void resetLocal(ClusterMessage m) {
+        String counterClass = new String(m.payload(), StandardCharsets.UTF_8);
+        checkNotNull(counterClass, "counter class can't be null");
+        for (PppoeAgentCounterNames counterType : PppoeAgentCounterNames.SUPPORTED_COUNTERS) {
+            PppoeAgentCountersIdentifier counterIdentifier =
+                    new PppoeAgentCountersIdentifier(counterClass, counterType);
+            countersMap.computeIfPresent(counterIdentifier, (key, counterValue) -> 0L);
+        }
+        dirty.set(true);
+        syncStats();
+    }
+
+    /**
+     * Inserts the counter entry if it is not already in the set otherwise update the existing counter entry.
+     * @param counterClass class of counters (global, per subscriber).
+     * @param counterType name of counter
+     * @param value counter value
+     */
+    public void setCounter(String counterClass, PppoeAgentCounterNames counterType, Long value) {
+        checkNotNull(counterClass, "counter class can't be null");
+        if (PppoeAgentCounterNames.SUPPORTED_COUNTERS.contains(counterType)) {
+            PppoeAgentCountersIdentifier counterIdentifier =
+                    new PppoeAgentCountersIdentifier(counterClass, counterType);
+            countersMap.put(counterIdentifier, value);
+        } else {
+            log.error("Failed to increment counter class {} of type {}", counterClass, counterType);
+        }
+        dirty.set(true);
+        syncStats();
+    }
+
+    private PppoeAgentStatistics aggregate() {
+        return statistics.values().stream()
+                .reduce(new PppoeAgentStatistics(), PppoeAgentStatistics::add);
+    }
+    /**
+     * Creates a snapshot of the current in-memory statistics.
+     *
+     * @return snapshot of statistics
+     */
+    private PppoeAgentStatistics snapshot() {
+        return PppoeAgentStatistics.withCounters(countersMap);
+    }
+    /**
+     * Syncs in-memory stats to the eventually consistent map.
+     */
+    private void syncStats() {
+        if (dirty.get()) {
+            statistics.put(clusterService.getLocalNode().id(), snapshot());
+            dirty.set(false);
+        }
+    }
+    private void publishStats() {
+        // Only publish events if we are the cluster leader for PPPoE Agent stats
+        if (!Objects.equals(leadershipService.getLeader(PPPOE_STATISTICS_LEADERSHIP),
+                clusterService.getLocalNode().id())) {
+            return;
+        }
+        if (aggregate().counters() != null) {
+            aggregate().counters().forEach((counterKey, counterValue) -> {
+                // Subscriber-specific counters have the subscriber ID set
+                String subscriberId = null;
+                if (!counterKey.counterClassKey.equals(PppoeAgentEvent.GLOBAL_COUNTER)) {
+                    subscriberId = counterKey.counterClassKey;
+                }
+                if (delegate != null) {
+                    delegate.notify(new PppoeAgentEvent(PppoeAgentEvent.Type.STATS_UPDATE, null,
+                                                        counterKey.counterTypeKey.toString(), counterValue,
+                                       null, subscriberId));
+                }
+            });
+        } else {
+            log.debug("Ignoring 'publishStats' request since counters are null.");
+        }
+    }
+}
\ No newline at end of file
diff --git a/app/src/main/java/org/opencord/pppoeagent/impl/package-info.java b/app/src/main/java/org/opencord/pppoeagent/impl/package-info.java
new file mode 100644
index 0000000..84ceb21
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/impl/package-info.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.
+ */
+
+/**
+ * PPPoE agent implementation.
+ */
+package org.opencord.pppoeagent.impl;
diff --git a/app/src/main/java/org/opencord/pppoeagent/rest/PppoeAgentWebResource.java b/app/src/main/java/org/opencord/pppoeagent/rest/PppoeAgentWebResource.java
new file mode 100644
index 0000000..56d5598
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/rest/PppoeAgentWebResource.java
@@ -0,0 +1,186 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.rest;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.util.Map;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+
+import org.onlab.packet.MacAddress;
+import org.onlab.packet.PPPoED;
+import org.onlab.util.ItemNotFoundException;
+import org.onosproject.cli.AbstractShellCommand;
+import org.onosproject.net.AnnotationKeys;
+import org.onosproject.net.ConnectPoint;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.Port;
+import org.onosproject.rest.AbstractWebResource;
+
+import org.opencord.pppoeagent.impl.PppoeAgentCounterNames;
+import org.opencord.pppoeagent.impl.PppoeAgentCountersStore;
+import org.opencord.pppoeagent.impl.PppoeAgentCountersIdentifier;
+import org.opencord.pppoeagent.impl.PppoeAgentStatistics;
+import org.opencord.pppoeagent.PppoeAgentEvent;
+import org.opencord.pppoeagent.PppoeAgentService;
+import org.opencord.pppoeagent.PppoeSessionInfo;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static javax.ws.rs.core.Response.Status.INTERNAL_SERVER_ERROR;
+
+/**
+ * PppoeAgent web resource.
+ */
+@Path("pppoeagent-app")
+public class PppoeAgentWebResource extends AbstractWebResource {
+    private final ObjectNode root = mapper().createObjectNode();
+    private final ArrayNode node = root.putArray("entry");
+    private static final String SESSION_NOT_FOUND = "Session not found";
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    DeviceService deviceService = AbstractShellCommand.get(DeviceService.class);
+
+    /**
+     * Get session info object.
+     *
+     * @param mac Session MAC address
+     *
+     * @return 200 OK
+     */
+    @GET
+    @Path("/session/{mac}")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getSubscriber(@PathParam("mac") String mac) {
+        MacAddress macAddress = MacAddress.valueOf(mac);
+        PppoeAgentService pppoeAgent = get(PppoeAgentService.class);
+        PppoeSessionInfo entry = pppoeAgent.getSessionsMap().get(macAddress);
+        if (entry == null) {
+            throw new ItemNotFoundException(SESSION_NOT_FOUND);
+        }
+
+        try {
+            node.add(encodePppoeSessionInfo(entry, macAddress));
+            return ok(mapper().writeValueAsString(root)).build();
+        } catch (IllegalArgumentException e) {
+            log.error("Error while fetching PPPoE session info for MAC {} through REST API: {}", mac, e.getMessage());
+            return Response.status(INTERNAL_SERVER_ERROR).build();
+        } catch (JsonProcessingException e) {
+            log.error("Error assembling JSON response for PPPoE session info request for MAC {} " +
+                    "through REST API: {}", mac, e.getMessage());
+            return Response.status(INTERNAL_SERVER_ERROR).build();
+        }
+    }
+
+    /**
+     * Get all session info objects.
+     *
+     * @return 200 OK
+     */
+    @GET
+    @Path("/session")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getSubscribers() {
+        try {
+            PppoeAgentService pppoeAgent = get(PppoeAgentService.class);
+            pppoeAgent.getSessionsMap().forEach((mac, entry) -> {
+                node.add(encodePppoeSessionInfo(entry, mac));
+            });
+
+            return ok(mapper().writeValueAsString(root)).build();
+        } catch (Exception e) {
+            log.error("Error while fetching PPPoE sessions information through REST API: {}", e.getMessage());
+            return Response.status(INTERNAL_SERVER_ERROR).build();
+        }
+    }
+
+    private ObjectNode encodePppoeSessionInfo(PppoeSessionInfo entry, MacAddress macAddress) {
+        ConnectPoint cp = entry.getClientCp();
+        Port devicePort = deviceService.getPort(cp);
+        String portLabel = "uni-" + ((cp.port().toLong() & 0xF) + 1);
+        String subscriberId = devicePort != null ? devicePort.annotations().value(AnnotationKeys.PORT_NAME) :
+                "UNKNOWN";
+
+        return mapper().createObjectNode()
+                .put("macAddress", macAddress.toString())
+                .put("sessionId", entry.getSessionId())
+                .put("currentState", entry.getCurrentState())
+                .put("lastReceivedPacket", PPPoED.Type.getTypeByValue(entry.getPacketCode()).name())
+                .put("deviceId", cp.deviceId().toString())
+                .put("portNumber", cp.port().toString())
+                .put("portLabel", portLabel)
+                .put("subscriberId", subscriberId);
+    }
+
+    /**
+     * Gets PPPoE Agent counters for global context.
+     *
+     * @return 200 OK
+     */
+    @GET
+    @Path("/stats")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getPppoeStats() {
+        return getStats(PppoeAgentEvent.GLOBAL_COUNTER);
+    }
+
+    /**
+     * Gets PPPoE Agent counters for specific subscriber.
+     *
+     * @param subscriberId Id of subscriber.
+     *
+     * @return 200 OK
+     */
+    @GET
+    @Path("/stats/{subscriberId}")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getPppoeSubscriberStats(@PathParam("subscriberId") String subscriberId) {
+        return getStats(subscriberId);
+    }
+    private Response getStats(String key) {
+        PppoeAgentCountersStore pppoeCounters = get(PppoeAgentCountersStore.class);
+        try {
+            PppoeAgentStatistics pppoeStatistics = pppoeCounters.getCounters();
+            JsonNode completeNode = buildPppoeCounterNodeObject(key, pppoeStatistics.counters());
+            return ok(mapper().writeValueAsString(completeNode)).build();
+        } catch (JsonProcessingException e) {
+            log.error("Error while fetching PPPoE agent counter stats through REST API: {}", e.getMessage());
+            return Response.status(INTERNAL_SERVER_ERROR).build();
+        }
+    }
+
+    private JsonNode buildPppoeCounterNodeObject(String key, Map<PppoeAgentCountersIdentifier, Long> countersMap) {
+        ObjectNode entryNode = mapper().createObjectNode();
+        for (PppoeAgentCounterNames counterType : PppoeAgentCounterNames.SUPPORTED_COUNTERS) {
+            Long value = countersMap.get(new PppoeAgentCountersIdentifier(key, counterType));
+            if (value == null) {
+                continue;
+            }
+            entryNode = entryNode.put(counterType.name(), String.valueOf(value));
+        }
+        return mapper().createObjectNode().set(key, entryNode);
+    }
+}
\ No newline at end of file
diff --git a/app/src/main/java/org/opencord/pppoeagent/rest/package-info.java b/app/src/main/java/org/opencord/pppoeagent/rest/package-info.java
new file mode 100644
index 0000000..70a18d3
--- /dev/null
+++ b/app/src/main/java/org/opencord/pppoeagent/rest/package-info.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.
+ */
+
+/**
+ * Rest interface for PppoeAgent.
+ */
+package org.opencord.pppoeagent.rest;
diff --git a/app/src/main/webapp/WEB-INF/web.xml b/app/src/main/webapp/WEB-INF/web.xml
new file mode 100644
index 0000000..9bb27a0
--- /dev/null
+++ b/app/src/main/webapp/WEB-INF/web.xml
@@ -0,0 +1,39 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2021-present Open Networking Foundation
+  ~
+  ~ 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.
+  -->
+<web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://java.sun.com/xml/ns/javaee"
+         xmlns:web="http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
+         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
+         id="ONOS" version="2.5">
+    <display-name>PPPoE Agent REST API v1.0</display-name>
+
+    <servlet>
+        <servlet-name>JAX-RS Service</servlet-name>
+        <servlet-class>org.glassfish.jersey.servlet.ServletContainer</servlet-class>
+        <init-param>
+            <param-name>jersey.config.server.provider.classnames</param-name>
+            <param-value>
+                org.opencord.pppoeagent.rest.PppoeAgentWebResource
+            </param-value>
+        </init-param>
+        <load-on-startup>1</load-on-startup>
+    </servlet>
+
+    <servlet-mapping>
+        <servlet-name>JAX-RS Service</servlet-name>
+        <url-pattern>/*</url-pattern>
+    </servlet-mapping>
+</web-app>
diff --git a/app/src/test/java/org/opencord/pppoeagent/impl/PppoeAgentTest.java b/app/src/test/java/org/opencord/pppoeagent/impl/PppoeAgentTest.java
new file mode 100644
index 0000000..6298acf
--- /dev/null
+++ b/app/src/test/java/org/opencord/pppoeagent/impl/PppoeAgentTest.java
@@ -0,0 +1,421 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.impl;
+
+import com.google.common.util.concurrent.MoreExecutors;
+import java.util.List;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.onlab.junit.TestUtils;
+import org.onlab.osgi.ComponentContextAdapter;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.MacAddress;
+import org.onlab.packet.PPPoED;
+import org.onlab.packet.PPPoEDTag;
+import org.onlab.packet.VlanId;
+
+import org.onosproject.net.Device;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.cfg.ComponentConfigService;
+import org.onosproject.cluster.ClusterServiceAdapter;
+import org.onosproject.cluster.LeadershipServiceAdapter;
+import org.onosproject.net.ConnectPoint;
+import org.onosproject.net.PortNumber;
+import org.onosproject.net.device.DeviceListener;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.TestStorageService;
+
+import org.opencord.pppoeagent.PppoeAgentEvent;
+import org.opencord.pppoeagent.PPPoEDVendorSpecificTag;
+import org.opencord.pppoeagent.PppoeSessionInfo;
+import org.opencord.sadis.SubscriberAndDeviceInformation;
+
+import static org.easymock.EasyMock.createMock;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertFalse;
+
+public class PppoeAgentTest extends PppoeAgentTestBase {
+    private PppoeAgent pppoeAgent;
+    private SimplePppoeAgentCountersStore store;
+
+    ComponentConfigService mockConfigService =
+            createMock(ComponentConfigService.class);
+
+    /**
+     * Sets up the services required by the PPPoE agent app.
+     */
+    @Before
+    public void setUp() {
+        pppoeAgent = new PppoeAgent();
+        pppoeAgent.cfgService = new MockNetworkConfigRegistry();
+        pppoeAgent.coreService = new MockCoreServiceAdapter();
+        pppoeAgent.packetService = new MockPacketService();
+        pppoeAgent.componentConfigService = mockConfigService;
+        pppoeAgent.deviceService = new MockDeviceService();
+        pppoeAgent.sadisService = new MockSadisService();
+        pppoeAgent.subsService = pppoeAgent.sadisService.getSubscriberInfoService();
+        pppoeAgent.mastershipService = new MockMastershipService();
+        pppoeAgent.storageService = new TestStorageService();
+        pppoeAgent.pppoeAgentCounters = this.store;
+
+        store = new SimplePppoeAgentCountersStore();
+        store.storageService = new TestStorageService();
+        store.clusterService = new ClusterServiceAdapter();
+        store.leadershipService = new LeadershipServiceAdapter();
+        store.clusterCommunicationService = new MockClusterCommunicationService<>();
+        store.componentConfigService = mockConfigService;
+        TestUtils.setField(store, "eventDispatcher", new MockEventDispatcher());
+        store.activate(new MockComponentContext());
+
+        pppoeAgent.pppoeAgentCounters = this.store;
+
+        TestUtils.setField(pppoeAgent, "eventDispatcher", new MockEventDispatcher());
+        TestUtils.setField(pppoeAgent, "packetProcessorExecutor", MoreExecutors.newDirectExecutorService());
+
+        pppoeAgent.activate(new ComponentContextAdapter());
+    }
+
+    /**
+     * Tears down the PPPoE agent app.
+     */
+    @After
+    public void tearDown() {
+        pppoeAgent.deactivate();
+    }
+
+    @Test
+    public void testPppoePadi() {
+        testPppoeUpstreamPacket(PPPoED.PPPOED_CODE_PADI);
+    }
+
+    @Test
+    public void testPppoePado() {
+        testPppoeDownstreamPacket(PPPoED.PPPOED_CODE_PADO);
+    }
+
+    @Test
+    public void testPppoePadr() {
+        testPppoeUpstreamPacket(PPPoED.PPPOED_CODE_PADR);
+    }
+
+    @Test
+    public void testPppoePads() {
+        testPppoeDownstreamPacket(PPPoED.PPPOED_CODE_PADS);
+    }
+
+    @Test
+    public void testPppoePadt() {
+        // To simulate a successful PADT from subscriber a session entry is needed.
+        PppoeSessionInfo sessionInfo = new PppoeSessionInfo(DEFAULT_CONNECT_POINT, SERVER_CONNECT_POINT,
+                                                            PPPoED.PPPOED_CODE_PADS, (short) 1,
+                                                            pppoeAgent.subsService.get(CLIENT_NAS_PORT_ID),
+                                                            CLIENT_MAC);
+        putInfoOnSessionMap(CLIENT_MAC, sessionInfo);
+        assertTrue(pppoeAgent.getSessionsMap().containsKey(CLIENT_MAC));
+
+        Ethernet padt = constructPppoedPacket(PPPoED.PPPOED_CODE_PADT, CLIENT_MAC, MacAddress.BROADCAST,
+                                              CLIENT_C_TAG, CLIENT_S_TAG, (short) 1);
+        sendPacket(padt, DEFAULT_CONNECT_POINT);
+        Ethernet processedPadt = (Ethernet) getPacket();
+        assertNotNull(processedPadt);
+        assertEquals(padt, processedPadt);
+        assertFalse(pppoeAgent.getSessionsMap().containsKey(CLIENT_MAC));
+        PppoeAgentEvent e = getEvent();
+        assertNotNull(e);
+        assertEquals(PppoeAgentEvent.Type.TERMINATE, e.type());
+        assertEquals(DEFAULT_CONNECT_POINT, e.getConnectPoint());
+        assertEquals(CLIENT_MAC, e.getSubscriberMacAddress());
+
+        // Simulating PADT from server.
+        putInfoOnSessionMap(CLIENT_MAC, sessionInfo);
+        assertTrue(pppoeAgent.getSessionsMap().containsKey(CLIENT_MAC));
+        padt = constructPppoedPacket(PPPoED.PPPOED_CODE_PADT, SERVER_MAC, CLIENT_MAC,
+                                     CLIENT_C_TAG, CLIENT_S_TAG, (short) 1);
+        sendPacket(padt, SERVER_CONNECT_POINT);
+        processedPadt = (Ethernet) getPacket();
+        assertNotNull(processedPadt);
+        assertEquals(padt, processedPadt);
+        assertFalse(pppoeAgent.getSessionsMap().containsKey(CLIENT_MAC));
+        e = getEvent();
+        assertNotNull(e);
+        assertEquals(PppoeAgentEvent.Type.TERMINATE, e.type());
+        assertEquals(SERVER_CONNECT_POINT, e.getConnectPoint());
+        assertEquals(CLIENT_MAC, e.getSubscriberMacAddress());
+
+        // Simulating PADT from client for unknown session (the packet must not be processed)..
+        padt = constructPppoedPacket(PPPoED.PPPOED_CODE_PADT, CLIENT_MAC, MacAddress.BROADCAST,
+                                     CLIENT_C_TAG, CLIENT_S_TAG, (short) 1);
+        sendPacket(padt, DEFAULT_CONNECT_POINT);
+        processedPadt = (Ethernet) getPacket();
+        assertNull(processedPadt);
+
+        // Simulating PADT from server for unknown session (the packet must not be processed).
+        padt = constructPppoedPacket(PPPoED.PPPOED_CODE_PADT, SERVER_MAC, CLIENT_MAC,
+                                     CLIENT_C_TAG, CLIENT_S_TAG, (short) 1);
+        sendPacket(padt, SERVER_CONNECT_POINT);
+        processedPadt = (Ethernet) getPacket();
+        assertNull(processedPadt);
+    }
+
+    @Test
+    public void testPppoeCircuitIdValidation() {
+        Ethernet packet = constructPppoedPacket(PPPoED.PPPOED_CODE_PADI, CLIENT_MAC, MacAddress.BROADCAST,
+                                                CLIENT_C_TAG, CLIENT_S_TAG, (short) 0);
+        // Send packet with a different port number, so there will be a circuit-id mismatch.
+        sendPacket(packet, new ConnectPoint(DEVICE_ID, PortNumber.portNumber(4096L)));
+        Ethernet processedPacket = (Ethernet) getPacket();
+        assertNull(processedPacket);
+        PppoeAgentEvent e = getEvent();
+        assertNotNull(e);
+        assertEquals(PppoeAgentEvent.Type.INVALID_CID, e.type());
+
+        // Now send it from the default connect point, which should generate the valid circuit-id.
+        sendPacket(packet, DEFAULT_CONNECT_POINT);
+        processedPacket = (Ethernet) getPacket();
+        assertNotNull(processedPacket);
+        PPPoED pppoeLayer = (PPPoED) processedPacket.getPayload();
+        assertNotNull(pppoeLayer);
+        PPPoEDTag tag = pppoeLayer.getTag(PPPoEDTag.PPPOED_TAG_VENDOR_SPECIFIC);
+        assertNotNull(tag);
+        PPPoEDVendorSpecificTag vendorSpecificTag = PPPoEDVendorSpecificTag.fromByteArray(tag.getValue());
+        assertNotNull(vendorSpecificTag);
+
+        // Checks if the configured circuit-id matches with the built one.
+        assertEquals(CLIENT_CIRCUIT_ID, vendorSpecificTag.getCircuitId());
+    }
+
+    @Test
+    public void testPppoeCounters() {
+        short sessionId = (short) 0;
+        Ethernet padi = constructPppoedPacket(PPPoED.PPPOED_CODE_PADI, CLIENT_MAC, MacAddress.BROADCAST,
+                                              CLIENT_C_TAG, CLIENT_S_TAG, sessionId);
+        Ethernet pado = constructPppoedPacket(PPPoED.PPPOED_CODE_PADO, SERVER_MAC, CLIENT_MAC,
+                                              CLIENT_C_TAG, CLIENT_S_TAG, sessionId);
+        Ethernet padr = constructPppoedPacket(PPPoED.PPPOED_CODE_PADR, CLIENT_MAC, MacAddress.BROADCAST,
+                                              CLIENT_C_TAG, CLIENT_S_TAG, sessionId);
+        sessionId++;
+        Ethernet pads = constructPppoedPacket(PPPoED.PPPOED_CODE_PADS, SERVER_MAC, CLIENT_MAC,
+                                              CLIENT_C_TAG, CLIENT_S_TAG, sessionId);
+
+        List.of(new CounterTester(PppoeAgentCounterNames.PADI, 6, padi, DEFAULT_CONNECT_POINT),
+                new CounterTester(PppoeAgentCounterNames.PADO, 2, pado, SERVER_CONNECT_POINT),
+                new CounterTester(PppoeAgentCounterNames.PADR, 5, padr, DEFAULT_CONNECT_POINT),
+                new CounterTester(PppoeAgentCounterNames.PADS, 3, pads, SERVER_CONNECT_POINT),
+                new CounterTester(PppoeAgentCounterNames.PPPOED_PACKETS_FROM_SERVER, 5, null, null),
+                new CounterTester(PppoeAgentCounterNames.PPPOED_PACKETS_TO_SERVER, 11, null, null),
+                new CounterTester(PppoeAgentCounterNames.AC_SYSTEM_ERROR, 0, null, null),
+                new CounterTester(PppoeAgentCounterNames.GENERIC_ERROR_FROM_CLIENT, 0, null, null),
+                new CounterTester(PppoeAgentCounterNames.GENERIC_ERROR_FROM_SERVER, 0, null, null),
+                new CounterTester(PppoeAgentCounterNames.MTU_EXCEEDED, 0, null, null),
+                new CounterTester(PppoeAgentCounterNames.SERVICE_NAME_ERROR, 0, null, null))
+        .forEach(CounterTester::test);
+    }
+
+    @Test
+    public void testSessionsMap() {
+        assertEquals(0, pppoeAgent.getSessionsMap().size());
+        Ethernet packet = constructPppoedPacket(PPPoED.PPPOED_CODE_PADI, CLIENT_MAC, MacAddress.BROADCAST,
+                                                CLIENT_C_TAG, CLIENT_S_TAG, (short) 0);
+        sendPacket(packet, DEFAULT_CONNECT_POINT);
+        assertEquals(1, pppoeAgent.getSessionsMap().size());
+
+        int randomPacketsNumber = 15;
+        sendMultiplePadi(randomPacketsNumber);
+        assertEquals(randomPacketsNumber + 1, pppoeAgent.getSessionsMap().size());
+        PppoeSessionInfo sessionInfo = pppoeAgent.getSessionsMap().get(CLIENT_MAC);
+        assertSessionInfo(sessionInfo, PPPoED.PPPOED_CODE_PADI, (short) 0);
+
+        packet = constructPppoedPacket(PPPoED.PPPOED_CODE_PADT, CLIENT_MAC, MacAddress.BROADCAST,
+                                       CLIENT_C_TAG, CLIENT_S_TAG, (short) 0);
+        sendPacket(packet, DEFAULT_CONNECT_POINT);
+
+        assertEquals(randomPacketsNumber, pppoeAgent.getSessionsMap().size());
+    }
+
+    @Test
+    public void testDeviceEvents() {
+        // Guarantee map is empty.
+        assertEquals(0, pppoeAgent.getSessionsMap().size());
+
+        // Fill sessionsMap by sending 10 PADI packets for random mac addresses.
+        int numPackets = 10;
+        sendMultiplePadi(numPackets);
+        assertEquals(numPackets, pppoeAgent.getSessionsMap().size());
+
+        // Generate PORT_REMOVED event and inject into the device listener.
+        DeviceListener deviceListener = TestUtils.getField(pppoeAgent, "deviceListener");
+        Device device = pppoeAgent.deviceService.getDevice(DEVICE_ID);
+        DeviceEvent deviceEvent = new DeviceEvent(DeviceEvent.Type.PORT_REMOVED,
+                                                  device,
+                                                  new MockPort(PortNumber.portNumber(1L)));
+        deviceListener.event(deviceEvent);
+
+        // Check if session map is empty again.
+        assertEquals(0, pppoeAgent.getSessionsMap().size());
+
+        // Perform the same test but for PORT_UPDATED event.
+        sendMultiplePadi(numPackets);
+        assertEquals(numPackets, pppoeAgent.getSessionsMap().size());
+        deviceEvent = new DeviceEvent(DeviceEvent.Type.PORT_UPDATED,
+                                      device,
+                                      new MockPort(PortNumber.portNumber(1L), false));
+        deviceListener.event(deviceEvent);
+        assertEquals(0, pppoeAgent.getSessionsMap().size());
+
+        // Same test for DEVICE_REMOVED.
+        sendMultiplePadi(numPackets);
+        assertEquals(numPackets, pppoeAgent.getSessionsMap().size());
+        deviceEvent = new DeviceEvent(DeviceEvent.Type.DEVICE_REMOVED, device, null);
+        deviceListener.event(deviceEvent);
+        assertEquals(0, pppoeAgent.getSessionsMap().size());
+    }
+
+    private void sendMultiplePadi(int num) {
+        for (int i = 0; i < num; i++) {
+            MacAddress macAddress;
+            // A trick to guarantee the Mac address won't repeat (this case may never occur).
+            do {
+                macAddress = randomizeMacAddress();
+            } while (pppoeAgent.getSessionsMap().containsKey(macAddress));
+
+            Ethernet packet = constructPppoedPacket(PPPoED.PPPOED_CODE_PADI, macAddress, MacAddress.BROADCAST,
+                    CLIENT_C_TAG, CLIENT_S_TAG, (short) 0);
+            sendPacket(packet, DEFAULT_CONNECT_POINT);
+        }
+    }
+
+    private void testPppoeUpstreamPacket(byte packetCode) {
+        Ethernet packet = constructPppoedPacket(packetCode, CLIENT_MAC, MacAddress.BROADCAST,
+                                                CLIENT_C_TAG, CLIENT_S_TAG, (short) 0);
+        sendPacket(packet, DEFAULT_CONNECT_POINT);
+
+        Ethernet processedPacket = (Ethernet) getPacket();
+        assertNotNull(processedPacket);
+
+        PPPoED pppoedLayer = (PPPoED) processedPacket.getPayload();
+        assertNotNull(pppoedLayer);
+
+        List<PPPoEDTag> pppoedTagList = pppoedLayer.getTags();
+        assertEquals(1, pppoedTagList.size());
+
+        PPPoEDTag ppPoEDTag = pppoedTagList.get(0);
+        assertEquals(PPPoEDTag.PPPOED_TAG_VENDOR_SPECIFIC, ppPoEDTag.getType());
+
+        PPPoEDVendorSpecificTag vendorSpecificTag = PPPoEDVendorSpecificTag.fromByteArray(ppPoEDTag.getValue());
+        assertEquals(CLIENT_CIRCUIT_ID, vendorSpecificTag.getCircuitId());
+        assertEquals(CLIENT_REMOTE_ID, vendorSpecificTag.getRemoteId());
+        assertEquals(Integer.valueOf(PPPoEDVendorSpecificTag.BBF_IANA_VENDOR_ID), vendorSpecificTag.getVendorId());
+
+        // The only difference between the original and processed is the tag list,
+        // so after removing it the packets must be equal.
+        pppoedLayer.setPayload(null);
+        pppoedLayer.setPayloadLength((short) 0);
+        processedPacket.setPayload(pppoedLayer);
+        assertEquals(packet, processedPacket);
+
+        PppoeSessionInfo sessionInfo = pppoeAgent.getSessionsMap().get(CLIENT_MAC);
+        assertNotNull(sessionInfo);
+        assertSessionInfo(sessionInfo, packetCode, (short) 0);
+    }
+
+    private void testPppoeDownstreamPacket(byte packetCode) {
+        // Simulating a session entry of a previous packet.
+        byte previousPacketCode = packetCode == PPPoED.PPPOED_CODE_PADO ? PPPoED.PPPOED_CODE_PADI :
+                                                                         PPPoED.PPPOED_CODE_PADR;
+
+        SubscriberAndDeviceInformation deviceInfo = pppoeAgent.subsService.get(CLIENT_NAS_PORT_ID);
+
+        putInfoOnSessionMap(CLIENT_MAC, new PppoeSessionInfo(DEFAULT_CONNECT_POINT, SERVER_CONNECT_POINT,
+                                                             previousPacketCode, (short) 0, deviceInfo, CLIENT_MAC));
+
+        short sessionId = (short) (packetCode == PPPoED.PPPOED_CODE_PADS ? 1 : 0);
+        Ethernet packet = constructPppoedPacket(packetCode, SERVER_MAC, CLIENT_MAC,
+                                                CLIENT_C_TAG, CLIENT_S_TAG, sessionId);
+        sendPacket(packet, SERVER_CONNECT_POINT);
+
+        Ethernet processedPacket = (Ethernet) getPacket();
+        assertNotNull(processedPacket);
+
+        // sTag is removed before sending the packet to client.
+        packet.setQinQVID(VlanId.UNTAGGED);
+        assertEquals(packet, processedPacket);
+
+        PppoeSessionInfo sessionInfo = pppoeAgent.getSessionsMap().get(CLIENT_MAC);
+        assertNotNull(sessionInfo);
+        assertSessionInfo(sessionInfo, packetCode, sessionId);
+    }
+
+    private void assertSessionInfo(PppoeSessionInfo sessionInfo, Byte packetCode, short sessionId) {
+        assertEquals(packetCode, sessionInfo.getPacketCode());
+        assertEquals(sessionId, sessionInfo.getSessionId());
+        assertEquals(DEFAULT_CONNECT_POINT, sessionInfo.getClientCp());
+        assertEquals(CLIENT_MAC, sessionInfo.getClientMac());
+    }
+
+    private void putInfoOnSessionMap(MacAddress key, PppoeSessionInfo sessionInfo) {
+        ConsistentMap<MacAddress, PppoeSessionInfo> sessionsMap = TestUtils.getField(pppoeAgent, "sessionsMap");
+        sessionsMap.put(key, sessionInfo);
+    }
+
+    class CounterTester {
+        CounterTester(String subscriber, PppoeAgentCounterNames counter,
+                      long expectedValue, Ethernet packetModel,
+                      ConnectPoint cp) {
+            this.subscriber = subscriber;
+            this.counter = counter;
+            this.expectedValue = expectedValue;
+            this.packetModel = packetModel;
+            this.cp = cp;
+        }
+
+        CounterTester(PppoeAgentCounterNames counter, long expectedValue,
+                      Ethernet packetModel, ConnectPoint cp) {
+            this(PppoeAgentEvent.GLOBAL_COUNTER, counter, expectedValue, packetModel, cp);
+        }
+
+        String subscriber;
+        PppoeAgentCounterNames counter;
+        long expectedValue;
+        Ethernet packetModel;
+        ConnectPoint cp;
+
+        void sendModel() {
+            for (int i = 0; i < expectedValue; i++) {
+                sendPacket(packetModel, cp);
+            }
+        }
+
+        void assertCounterValue() {
+            long actualValue = store.getCountersMap()
+                    .get(new PppoeAgentCountersIdentifier(subscriber, counter));
+            assertEquals(expectedValue, actualValue);
+        }
+
+        void test() {
+            if (packetModel != null && cp != null) {
+                sendModel();
+            }
+            assertCounterValue();
+        }
+    }
+}
diff --git a/app/src/test/java/org/opencord/pppoeagent/impl/PppoeAgentTestBase.java b/app/src/test/java/org/opencord/pppoeagent/impl/PppoeAgentTestBase.java
new file mode 100644
index 0000000..59dc216
--- /dev/null
+++ b/app/src/test/java/org/opencord/pppoeagent/impl/PppoeAgentTestBase.java
@@ -0,0 +1,751 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+ *
+ * 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.opencord.pppoeagent.impl;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.TimeUnit;
+import java.util.Dictionary;
+import java.util.Hashtable;
+import java.util.List;
+import java.util.LinkedList;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.time.Duration;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.onlab.packet.BasePacket;
+import org.onlab.packet.ChassisId;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.Ip4Address;
+import org.onlab.packet.MacAddress;
+import org.onlab.packet.PPPoED;
+import org.onlab.packet.VlanId;
+
+import org.onosproject.cluster.NodeId;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreServiceAdapter;
+import org.onosproject.core.DefaultApplicationId;
+import org.onosproject.event.DefaultEventSinkRegistry;
+import org.onosproject.event.Event;
+import org.onosproject.event.EventDeliveryService;
+import org.onosproject.event.EventSink;
+import org.onosproject.mastership.MastershipServiceAdapter;
+import org.onosproject.net.AnnotationKeys;
+import org.onosproject.net.Annotations;
+import org.onosproject.net.config.Config;
+import org.onosproject.net.config.NetworkConfigRegistryAdapter;
+import org.onosproject.net.ConnectPoint;
+import org.onosproject.net.DefaultAnnotations;
+import org.onosproject.net.DefaultDevice;
+import org.onosproject.net.Device;
+import org.onosproject.net.device.DeviceServiceAdapter;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.Element;
+import org.onosproject.net.packet.DefaultInboundPacket;
+import org.onosproject.net.packet.DefaultPacketContext;
+import org.onosproject.net.packet.InboundPacket;
+import org.onosproject.net.packet.OutboundPacket;
+import org.onosproject.net.packet.PacketContext;
+import org.onosproject.net.packet.PacketProcessor;
+import org.onosproject.net.packet.PacketServiceAdapter;
+import org.onosproject.net.provider.ProviderId;
+import org.onosproject.net.Port;
+import org.onosproject.net.PortNumber;
+import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
+import org.onosproject.store.cluster.messaging.ClusterMessageHandler;
+import org.onosproject.store.cluster.messaging.MessageSubject;
+
+import org.opencord.pppoeagent.PppoeAgentEvent;
+import org.opencord.sadis.BandwidthProfileInformation;
+import org.opencord.sadis.BaseInformationService;
+import org.opencord.sadis.SadisService;
+import org.opencord.sadis.SubscriberAndDeviceInformation;
+import org.opencord.sadis.UniTagInformation;
+
+import org.osgi.framework.Bundle;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.ServiceReference;
+import org.osgi.service.component.ComponentContext;
+import org.osgi.service.component.ComponentInstance;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/**
+ * Common methods for PpppoeAgent app tests.
+ */
+public class PppoeAgentTestBase {
+    static final int UPLINK_PORT = 5;
+    static final String OLT_DEV_ID = "of:00000000000000aa";
+    static final String OLT_SERIAL_NUMBER = "OLT123456789";
+
+    static final VlanId CLIENT_C_TAG = VlanId.vlanId((short) 999);
+    static final VlanId CLIENT_C_TAG_2 = VlanId.vlanId((short) 998);
+    static final VlanId CLIENT_S_TAG = VlanId.vlanId((short) 111);
+    static final String CLIENT_ID_1 = "SUBSCRIBER_ID_1";
+    static final short CLIENT_C_PBIT = 6;
+    static final String CLIENT_NAS_PORT_ID = "ONU123456789";
+    static final String CLIENT_REMOTE_ID = "remote0";
+
+    static final MacAddress CLIENT_MAC = MacAddress.valueOf("B8:26:D4:09:E5:D1");
+    static final MacAddress SERVER_MAC = MacAddress.valueOf("74:86:7A:FB:07:86");
+    static final MacAddress OLT_MAC_ADDRESS = MacAddress.valueOf("01:02:03:04:05:06");
+
+    static final ConnectPoint DEFAULT_CONNECT_POINT = ConnectPoint.deviceConnectPoint(OLT_DEV_ID + "/" + 1);
+    static final String CLIENT_CIRCUIT_ID = String.format("%s 0/%s:%s", OLT_SERIAL_NUMBER,
+                                                         (DEFAULT_CONNECT_POINT.port().toLong() >> 12) + 1,
+                                                          CLIENT_NAS_PORT_ID);
+
+    static final DeviceId DEVICE_ID = DeviceId.deviceId(OLT_DEV_ID);
+    static final String SCHEME_NAME = "pppoeagent";
+
+    static final ConnectPoint SERVER_CONNECT_POINT = ConnectPoint.deviceConnectPoint("of:00000000000000aa/5");
+
+    static final DefaultAnnotations DEVICE_ANNOTATIONS = DefaultAnnotations.builder()
+            .set(AnnotationKeys.PROTOCOL, SCHEME_NAME.toUpperCase()).build();
+
+    List<BasePacket> savedPackets = new LinkedList<>();
+    PacketProcessor packetProcessor;
+
+    /**
+     * Saves the given packet onto the saved packets list.
+     *
+     * @param packet packet to save
+     */
+    void savePacket(BasePacket packet) {
+        savedPackets.add(packet);
+    }
+
+    /**
+     * Gets and removes the packet in the 1st position of savedPackets list.
+     *
+     * @return the packet in 1st position of savedPackets list.
+     */
+    BasePacket getPacket() {
+        return savedPackets.size() > 0 ? savedPackets.remove(0) : null;
+    }
+
+    /**
+     * Gets the last generated event.
+     *
+     * @return the last generated pppoe agent event.
+     */
+    PppoeAgentEvent getEvent() {
+        List<Event> savedEvents = MockEventDispatcher.eventList;
+        return savedEvents.size() > 0 ? (PppoeAgentEvent) savedEvents.get(savedEvents.size() - 1) : null;
+    }
+
+    /**
+     * Sends an Ethernet packet to the process method of the Packet Processor.
+     *
+     * @param pkt Ethernet packet
+     * @param cp ConnectPoint cp
+     */
+    void sendPacket(Ethernet pkt, ConnectPoint cp) {
+        final ByteBuffer byteBuffer = ByteBuffer.wrap(pkt.serialize());
+        InboundPacket inPacket = new DefaultInboundPacket(cp, pkt, byteBuffer);
+
+        PacketContext context = new MockPacketContext(127L, inPacket, null, false);
+        packetProcessor.process(context);
+    }
+
+    /**
+     * Mocks core service adaptor that provides an appId.
+     */
+    class MockCoreServiceAdapter extends CoreServiceAdapter {
+
+        @Override
+        public ApplicationId registerApplication(String name) {
+            return new DefaultApplicationId(10, name);
+        }
+    }
+
+    /**
+     * Mocks device service to provide the of device object.
+     */
+    class MockDeviceService extends DeviceServiceAdapter {
+
+        private ProviderId providerId = new ProviderId("of", "foo");
+        private final Device device1 = new PppoeAgentTestBase.MockDevice(providerId, DEVICE_ID, Device.Type.SWITCH,
+                "foo.inc", "0", "0", OLT_SERIAL_NUMBER, new ChassisId(),
+                DEVICE_ANNOTATIONS);
+
+        @Override
+        public Device getDevice(DeviceId devId) {
+            return device1;
+        }
+
+        @Override
+        public Port getPort(ConnectPoint cp) {
+            return new PppoeAgentTestBase.MockPort(cp.port());
+        }
+
+        @Override
+        public Port getPort(DeviceId deviceId, PortNumber portNumber) {
+            return new PppoeAgentTestBase.MockPort(portNumber);
+        }
+
+        @Override
+        public boolean isAvailable(DeviceId d) {
+            return true;
+        }
+    }
+
+    /**
+     * Mocks device object.
+     */
+    class MockDevice extends DefaultDevice {
+
+        public MockDevice(ProviderId providerId, DeviceId id, Type type,
+                          String manufacturer, String hwVersion, String swVersion,
+                          String serialNumber, ChassisId chassisId, Annotations... annotations) {
+            super(providerId, id, type, manufacturer, hwVersion, swVersion, serialNumber,
+                    chassisId, annotations);
+        }
+    }
+
+    /**
+     * Mocks port object.
+     */
+    class  MockPort implements Port {
+        private PortNumber portNumber;
+        private boolean isEnabled = true;
+        MockPort(PortNumber portNumber) {
+            this.portNumber = portNumber;
+        }
+
+        MockPort(PortNumber portNumber, boolean isEnabled) {
+            this.portNumber = portNumber;
+            this.isEnabled = isEnabled;
+        }
+
+        @Override
+        public boolean isEnabled() {
+            return this.isEnabled;
+        }
+        @Override
+        public long portSpeed() {
+            return 1000;
+        }
+        @Override
+        public Element element() {
+            return null;
+        }
+        @Override
+        public PortNumber number() {
+            return this.portNumber;
+        }
+        @Override
+        public Annotations annotations() {
+            return new MockAnnotations();
+        }
+        @Override
+        public Type type() {
+            return Port.Type.FIBER;
+        }
+
+        private class MockAnnotations implements Annotations {
+
+            @Override
+            public String value(String val) {
+                return CLIENT_NAS_PORT_ID;
+            }
+            @Override
+            public Set<String> keys() {
+                return null;
+            }
+        }
+    }
+
+    /**
+     * Mocks mastership service to enable local-master operations.
+     */
+    class MockMastershipService extends MastershipServiceAdapter {
+        @Override
+        public boolean isLocalMaster(DeviceId d) {
+            return true;
+        }
+    }
+
+    /**
+     * Keeps a reference to the PacketProcessor and saves the OutboundPackets.
+     */
+    class MockPacketService extends PacketServiceAdapter {
+
+        @Override
+        public void addProcessor(PacketProcessor processor, int priority) {
+            packetProcessor = processor;
+        }
+
+        @Override
+        public void emit(OutboundPacket packet) {
+            try {
+                Ethernet eth = Ethernet.deserializer().deserialize(packet.data().array(),
+                        0, packet.data().array().length);
+                savePacket(eth);
+            } catch (Exception e) {
+                fail(e.getMessage());
+            }
+        }
+    }
+
+    /**
+     * Mocks Sadis service.
+     */
+    class MockSadisService implements SadisService {
+        @Override
+        public BaseInformationService<SubscriberAndDeviceInformation> getSubscriberInfoService() {
+            return new PppoeAgentTestBase.MockSubService();
+        }
+
+        @Override
+        public BaseInformationService<BandwidthProfileInformation> getBandwidthProfileService() {
+            return null;
+        }
+    }
+
+    /**
+     * Mocks Sadis content with a device and a subscriber entry.
+     */
+    class MockSubService implements BaseInformationService<SubscriberAndDeviceInformation> {
+
+        PppoeAgentTestBase.MockSubscriberAndDeviceInformation device =
+                new PppoeAgentTestBase.MockSubscriberAndDeviceInformation(OLT_DEV_ID, VlanId.NONE, VlanId.NONE,
+                                                                          VlanId.NONE, null, null,
+                                                                          OLT_MAC_ADDRESS,
+                                                                          Ip4Address.valueOf("10.10.10.10"),
+                                                                          UPLINK_PORT, null);
+
+        PppoeAgentTestBase.MockSubscriberAndDeviceInformation sub =
+                new PppoeAgentTestBase.MockSubscriberAndDeviceInformation(CLIENT_ID_1, CLIENT_C_TAG, CLIENT_C_TAG_2,
+                                                                          CLIENT_S_TAG, CLIENT_NAS_PORT_ID,
+                                                                          CLIENT_CIRCUIT_ID, null, null,
+                                                                -1, CLIENT_REMOTE_ID);
+
+        @Override
+        public SubscriberAndDeviceInformation get(String id) {
+            if (id.equals(OLT_SERIAL_NUMBER)) {
+                return device;
+            } else {
+                return  sub;
+            }
+        }
+
+        @Override
+        public void invalidateAll() {}
+        @Override
+        public void invalidateId(String id) {}
+        @Override
+        public SubscriberAndDeviceInformation getfromCache(String id) {
+            return null;
+        }
+    }
+
+    /**
+     * Mock Sadis object to populate service.
+     */
+    class MockSubscriberAndDeviceInformation extends SubscriberAndDeviceInformation {
+
+        MockSubscriberAndDeviceInformation(String id, VlanId cTag, VlanId cTag2,
+                                           VlanId sTag, String nasPortId,
+                                           String circuitId, MacAddress hardId,
+                                           Ip4Address ipAddress, int uplinkPort,
+                                           String remoteId) {
+            this.setHardwareIdentifier(hardId);
+            this.setId(id);
+            this.setIPAddress(ipAddress);
+            this.setNasPortId(nasPortId);
+            this.setCircuitId(circuitId);
+            this.setUplinkPort(uplinkPort);
+            this.setRemoteId(remoteId);
+
+            List<UniTagInformation> uniTagInformationList = new ArrayList<>();
+
+            UniTagInformation uniTagInformation = new UniTagInformation.Builder()
+                    .setPonCTag(cTag)
+                    .setPonSTag(sTag)
+                    .setUsPonCTagPriority(CLIENT_C_PBIT)
+                    .build();
+
+            UniTagInformation uniTagInformation2 = new UniTagInformation.Builder()
+                    .setPonCTag(cTag2)
+                    .setPonSTag(sTag)
+                    .setUsPonCTagPriority(CLIENT_C_PBIT)
+                    .build();
+
+            uniTagInformationList.add(uniTagInformation);
+            uniTagInformationList.add(uniTagInformation2);
+            this.setUniTagList(uniTagInformationList);
+        }
+    }
+
+    /**
+     * Mocks component context.
+     */
+    class MockComponentContext implements ComponentContext {
+        @Override
+        public Dictionary<String, Object> getProperties() {
+            Dictionary<String, Object> cfgDict = new Hashtable<String, Object>();
+            return cfgDict;
+        }
+
+        @Override
+        public Object locateService(String name) {
+            return null;
+        }
+
+        @Override
+        public Object locateService(String name, ServiceReference reference) {
+            return null;
+        }
+
+        @Override
+        public Object[] locateServices(String name) {
+            return null;
+        }
+
+        @Override
+        public BundleContext getBundleContext() {
+            return null;
+        }
+
+        @Override
+        public Bundle getUsingBundle() {
+            return null;
+        }
+
+        @Override
+        public ComponentInstance getComponentInstance() {
+            return null;
+        }
+
+        @Override
+        public void enableComponent(String name) {
+        }
+
+        @Override
+        public void disableComponent(String name) {
+        }
+
+        @Override
+        public ServiceReference getServiceReference() {
+            return null;
+        }
+    }
+
+    /**
+     * Mocks the network config registry.
+     */
+    @SuppressWarnings("unchecked")
+    class MockNetworkConfigRegistry
+            extends NetworkConfigRegistryAdapter {
+        @Override
+        public <S, C extends Config<S>> C getConfig(S subject, Class<C> configClass) {
+            PppoeAgentConfig pppoeAgentConfig = new MockPppoeAgentConfig();
+            return (C) pppoeAgentConfig;
+        }
+    }
+
+    /**
+     * Mocks the network config registry.
+     */
+    class MockPppoeAgentConfig extends PppoeAgentConfig {
+        @Override
+        public boolean getUseOltUplinkForServerPktInOut() {
+            return true;
+        }
+    }
+
+    /**
+     * Mocks the DefaultPacketContext.
+     */
+    final class MockPacketContext extends DefaultPacketContext {
+
+        private MockPacketContext(long time, InboundPacket inPkt,
+                                  OutboundPacket outPkt, boolean block) {
+            super(time, inPkt, outPkt, block);
+        }
+
+        @Override
+        public void send() {
+            // We don't send anything out.
+        }
+    }
+
+    /**
+     * Creates a mock for event delivery service.
+     */
+    static class MockEventDispatcher extends DefaultEventSinkRegistry
+            implements EventDeliveryService {
+        static List<Event> eventList = new LinkedList<>();
+        @Override
+        @SuppressWarnings("unchecked")
+        public synchronized void post(Event event) {
+            EventSink sink = getSink(event.getClass());
+            checkState(sink != null, "No sink for event %s", event);
+            sink.process(event);
+            eventList.add(event);
+        }
+
+        @Override
+        public void setDispatchTimeLimit(long millis) {
+        }
+
+        @Override
+        public long getDispatchTimeLimit() {
+            return 0;
+        }
+    }
+
+    /**
+     * Creates a mock object for a scheduled executor service.
+     */
+    class MockExecutor implements ScheduledExecutorService {
+        private ScheduledExecutorService executor;
+
+        MockExecutor(ScheduledExecutorService executor) {
+            this.executor = executor;
+        }
+
+        String lastMethodCalled = "";
+        long lastInitialDelay;
+        long lastDelay;
+        TimeUnit lastUnit;
+
+        public void assertLastMethodCalled(String method, long initialDelay, long delay, TimeUnit unit) {
+            assertEquals(method, lastMethodCalled);
+            assertEquals(initialDelay, lastInitialDelay);
+            assertEquals(delay, lastDelay);
+            assertEquals(unit, lastUnit);
+        }
+
+        @Override
+        public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+            lastMethodCalled = "scheduleRunnable";
+            lastDelay = delay;
+            lastUnit = unit;
+            return null;
+        }
+
+        @Override
+        public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
+            lastMethodCalled = "scheduleCallable";
+            lastDelay = delay;
+            lastUnit = unit;
+            return null;
+        }
+
+        @Override
+        public ScheduledFuture<?> scheduleAtFixedRate(
+                Runnable command, long initialDelay, long period, TimeUnit unit) {
+            lastMethodCalled = "scheduleAtFixedRate";
+            lastInitialDelay = initialDelay;
+            lastDelay = period;
+            lastUnit = unit;
+            return null;
+        }
+
+        @Override
+        public ScheduledFuture<?> scheduleWithFixedDelay(
+                Runnable command, long initialDelay, long delay, TimeUnit unit) {
+            lastMethodCalled = "scheduleWithFixedDelay";
+            lastInitialDelay = initialDelay;
+            lastDelay = delay;
+            lastUnit = unit;
+            command.run();
+            return null;
+        }
+
+        @Override
+        public boolean awaitTermination(long timeout, TimeUnit unit) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
+                throws InterruptedException {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public <T> List<Future<T>> invokeAll(
+                Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+                throws InterruptedException {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
+                throws ExecutionException, InterruptedException {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+                throws ExecutionException, InterruptedException, TimeoutException {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean isShutdown() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean isTerminated() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public void shutdown() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public List<Runnable> shutdownNow() {
+            return null;
+        }
+
+        @Override
+        public <T> Future<T> submit(Callable<T> task) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Future<?> submit(Runnable task) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public <T> Future<T> submit(Runnable task, T result) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public void execute(Runnable command) {
+            throw new UnsupportedOperationException();
+        }
+    }
+
+    /**
+     * Creates a PPPoED message encapsulated by an Ethernet object.
+     *
+     * @param type        PPoED message type.
+     * @param source      source address.
+     * @param destination destination address of the message.
+     * @param cVlan       inner vlan.
+     * @param sVlan       outer vlan.
+     * @param sessionId   session-id number.
+     *
+     * @return Ethernet packet with PPPoED payload.
+     */
+    Ethernet constructPppoedPacket(byte type, MacAddress source, MacAddress destination,
+                                   VlanId cVlan, VlanId sVlan, short sessionId) {
+        Ethernet pppoedPacket = new Ethernet();
+        pppoedPacket.setSourceMACAddress(source);
+        pppoedPacket.setDestinationMACAddress(destination);
+        pppoedPacket.setEtherType(Ethernet.TYPE_PPPOED);
+        pppoedPacket.setVlanID(cVlan.toShort());
+        pppoedPacket.setPriorityCode((byte) CLIENT_C_PBIT);
+        pppoedPacket.setQinQTPID(Ethernet.TYPE_VLAN);
+        pppoedPacket.setQinQVID(sVlan.toShort());
+
+        PPPoED pppoedLayer = new PPPoED();
+        pppoedLayer.setCode(type);
+        pppoedLayer.setSessionId(sessionId);
+        pppoedPacket.setPayload(pppoedLayer);
+
+        return pppoedPacket;
+    }
+
+    public class MockClusterCommunicationService<M> implements ClusterCommunicationService {
+        private Consumer handler;
+        @Override
+        public void addSubscriber(MessageSubject subject,
+                                  ClusterMessageHandler subscriber, ExecutorService executor) {
+        }
+        @Override
+        public <M> void broadcast(M message, MessageSubject subject, Function<M, byte[]> encoder) {
+        }
+        @Override
+        public <M> void broadcastIncludeSelf(M message, MessageSubject subject, Function<M, byte[]> encoder) {
+            handler.accept(message);
+        }
+        @Override
+        public <M> CompletableFuture<Void> unicast(M message, MessageSubject subject,
+                                                   Function<M, byte[]> encoder, NodeId toNodeId) {
+            return null;
+        }
+        @Override
+        public <M> void multicast(M message, MessageSubject subject,
+                                  Function<M, byte[]> encoder, Set<NodeId> nodeIds) {
+        }
+        @Override
+        public <M, R> CompletableFuture<R> sendAndReceive(M message, MessageSubject subject,
+                                                          Function<M, byte[]> encoder,
+                                                          Function<byte[], R> decoder, NodeId toNodeId) {
+            return null;
+        }
+        @Override
+        public <M, R> CompletableFuture<R> sendAndReceive(M message, MessageSubject subject,
+                                                          Function<M, byte[]> encoder, Function<byte[], R> decoder,
+                                                          NodeId toNodeId, Duration timeout) {
+            return null;
+        }
+        @Override
+        public <M, R> void addSubscriber(MessageSubject subject, Function<byte[], M> decoder,
+                                         Function<M, R> handler, Function<R, byte[]> encoder, Executor executor) {
+        }
+        @Override
+        public <M, R> void addSubscriber(MessageSubject subject, Function<byte[], M> decoder,
+                                         Function<M, CompletableFuture<R>> handler, Function<R, byte[]> encoder) {
+        }
+        @Override
+        public <M> void addSubscriber(MessageSubject subject, Function<byte[], M> decoder,
+                                      Consumer<M> handler, Executor executor) {
+            this.handler = handler;
+        }
+        @Override
+        public void removeSubscriber(MessageSubject subject) {
+        }
+    }
+
+    /**
+     * Creates a random MAC address.
+     *
+     * @return random MAC address object.
+     */
+    MacAddress randomizeMacAddress() {
+        byte[] mac = new byte[6];
+        Random r = new Random();
+        r.nextBytes(mac);
+        return MacAddress.valueOf(mac);
+    }
+}
\ No newline at end of file