diff --git a/app/src/main/java/org/opencord/aaa/impl/AaaManager.java b/app/src/main/java/org/opencord/aaa/impl/AaaManager.java
index 8e1f5c1..f14cf66 100644
--- a/app/src/main/java/org/opencord/aaa/impl/AaaManager.java
+++ b/app/src/main/java/org/opencord/aaa/impl/AaaManager.java
@@ -67,6 +67,9 @@
 import org.onosproject.net.packet.PacketProcessor;
 import org.onosproject.net.packet.PacketService;
 import org.opencord.aaa.AaaConfig;
+import org.opencord.aaa.AaaMachineStatisticsEvent;
+import org.opencord.aaa.AaaMachineStatisticsService;
+import org.opencord.aaa.AaaSupplicantMachineStats;
 import org.opencord.aaa.AuthenticationEvent;
 import org.opencord.aaa.AuthenticationEventListener;
 import org.opencord.aaa.AuthenticationService;
@@ -128,6 +131,9 @@
     protected AuthenticationStatisticsService aaaStatisticsManager;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected AaaMachineStatisticsService aaaSupplicantStatsManager;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected ComponentConfigService cfgService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
@@ -475,9 +481,18 @@
             aaaStatisticsManager.getAaaStats().countDroppedResponsesRx();
             return;
         }
+
+        //instance of StateMachine using the sessionId for updating machine stats
+        StateMachine machineStats = StateMachine.lookupStateMachineBySessionId(stateMachine.sessionId());
+
         EAP eapPayload;
         Ethernet eth;
         checkReceivedPacketForValidValidator(radiusPacket, stateMachine.requestAuthenticator());
+
+        //increasing packets and octets received from server
+        machineStats.incrementTotalPacketsReceived();
+        machineStats.incrementTotalOctetReceived(radiusPacket.decapsulateMessage().getLength());
+
         if (outPacketSet.contains(radiusPacket.getIdentifier())) {
             aaaStatisticsManager.getAaaStats().increaseOrDecreasePendingRequests(false);
             outPacketSet.remove(new Byte(radiusPacket.getIdentifier()));
@@ -502,6 +517,9 @@
                 aaaStatisticsManager.getAaaStats().increaseChallengeResponsesRx();
                 outPacketSupp.add(eapPayload.getIdentifier());
                 aaaStatisticsManager.getAaaStats().incrementPendingResSupp();
+                //increasing packets send to server
+                machineStats.incrementTotalPacketsSent();
+                machineStats.incrementTotalOctetSent(eapPayload.getLength());
                 break;
             case RADIUS.RADIUS_CODE_ACCESS_ACCEPT:
                 log.debug("RADIUS packet: RADIUS_CODE_ACCESS_ACCEPT");
@@ -521,6 +539,9 @@
 
                 stateMachine.authorizeAccess();
                 aaaStatisticsManager.getAaaStats().increaseAcceptResponsesRx();
+                //increasing packets send to server
+                machineStats.incrementTotalPacketsSent();
+                machineStats.incrementTotalOctetSent(eapPayload.getLength());
                 break;
             case RADIUS.RADIUS_CODE_ACCESS_REJECT:
                 log.debug("RADIUS packet: RADIUS_CODE_ACCESS_REJECT");
@@ -548,10 +569,20 @@
 
                 stateMachine.denyAccess();
                 aaaStatisticsManager.getAaaStats().increaseRejectResponsesRx();
+                //increasing packets send to server
+                machineStats.incrementTotalPacketsSent();
+                machineStats.incrementTotalOctetSent(eapPayload.getLength());
+                //pushing machine stats to kafka
+                AaaSupplicantMachineStats machineObj = aaaSupplicantStatsManager.getSupplicantStats(machineStats);
+                aaaSupplicantStatsManager.getMachineStatsDelegate()
+                        .notify(new AaaMachineStatisticsEvent(AaaMachineStatisticsEvent.Type.STATS_UPDATE, machineObj));
                 break;
             default:
                 log.warn("Unknown RADIUS message received with code: {}", radiusPacket.getCode());
                 aaaStatisticsManager.getAaaStats().increaseUnknownTypeRx();
+                //increasing packets received to server
+                machineStats.incrementTotalPacketsReceived();
+                machineStats.incrementTotalOctetReceived(radiusPacket.decapsulateMessage().getLength());
         }
         aaaStatisticsManager.getAaaStats().countDroppedResponsesRx();
     }
@@ -684,6 +715,7 @@
                 stateMachine = new StateMachine(sessionId);
             } else {
                 log.debug("Using existing state-machine for sessionId: {}", sessionId);
+                stateMachine.setEapolTypeVal(eapol.getEapolType());
             }
 
             switch (eapol.getEapolType()) {
@@ -713,6 +745,15 @@
                     break;
                 case EAPOL.EAPOL_LOGOFF:
                     log.debug("EAP packet: EAPOL_LOGOFF");
+                    //posting the machine stat data for current supplicant device.
+                    if (stateMachine.getSessionTerminateReason() == null ||
+                            stateMachine.getSessionTerminateReason().equals("")) {
+                        stateMachine.setSessionTerminateReason(
+                                StateMachine.SessionTerminationReasons.SUPPLICANT_LOGOFF.getReason());
+                    }
+                    AaaSupplicantMachineStats obj = aaaSupplicantStatsManager.getSupplicantStats(stateMachine);
+                    aaaSupplicantStatsManager.getMachineStatsDelegate()
+                            .notify(new AaaMachineStatisticsEvent(AaaMachineStatisticsEvent.Type.STATS_UPDATE, obj));
                     if (stateMachine.state() == StateMachine.STATE_AUTHORIZED) {
                         stateMachine.logoff();
                         aaaStatisticsManager.getAaaStats().incrementEapolLogoffRx();
@@ -750,6 +791,8 @@
                             // change the state to "PENDING"
                             if (stateMachine.state() == StateMachine.STATE_PENDING) {
                                 aaaStatisticsManager.getAaaStats().increaseRequestReTx();
+                                stateMachine.incrementTotalPacketsSent();
+                                stateMachine.incrementTotalOctetSent(eapol.getPacketLength());
                             }
                             stateMachine.requestAccess();
                             break;
@@ -917,6 +960,16 @@
                     PortNumber portNumber = event.port().number();
                     String sessionId = devId.toString() + portNumber.toString();
 
+                    StateMachine stateMachine = StateMachine.lookupStateMachineBySessionId(sessionId);
+                    if (stateMachine != null) {
+                        stateMachine.setSessionTerminateReason(
+                                StateMachine.SessionTerminationReasons.PORT_REMOVED.getReason());
+                    }
+                    //pushing captured machine stats to kafka
+                    AaaSupplicantMachineStats obj = aaaSupplicantStatsManager.getSupplicantStats(stateMachine);
+                    aaaSupplicantStatsManager.getMachineStatsDelegate()
+                            .notify(new AaaMachineStatisticsEvent(AaaMachineStatisticsEvent.Type.STATS_UPDATE, obj));
+
                     Map<String, StateMachine> sessionIdMap = StateMachine.sessionIdMap();
                     StateMachine removed = sessionIdMap.remove(sessionId);
                     if (removed != null) {
diff --git a/app/src/main/java/org/opencord/aaa/impl/AaaSupplicantMachineStatsManager.java b/app/src/main/java/org/opencord/aaa/impl/AaaSupplicantMachineStatsManager.java
new file mode 100644
index 0000000..62eb5bf
--- /dev/null
+++ b/app/src/main/java/org/opencord/aaa/impl/AaaSupplicantMachineStatsManager.java
@@ -0,0 +1,124 @@
+/*
+ * Copyright 2018-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.aaa.impl;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Service;
+import org.onosproject.event.AbstractListenerManager;
+import org.opencord.aaa.AaaMachineStatisticsDelegate;
+import org.opencord.aaa.AaaMachineStatisticsEvent;
+import org.opencord.aaa.AaaMachineStatisticsEventListener;
+import org.opencord.aaa.AaaMachineStatisticsService;
+import org.opencord.aaa.AaaSupplicantMachineStats;
+import org.slf4j.Logger;
+
+@Service
+@Component(immediate = true)
+public class AaaSupplicantMachineStatsManager
+        extends AbstractListenerManager<AaaMachineStatisticsEvent, AaaMachineStatisticsEventListener>
+        implements AaaMachineStatisticsService {
+
+    private final Logger log = getLogger(getClass());
+
+    private AaaMachineStatisticsDelegate machineStatDelegate;
+
+    @Activate
+    public void activate() {
+        log.info("Activate aaaStatisticsManager");
+        machineStatDelegate = new InternalMachineStatDelegate();
+        eventDispatcher.addSink(AaaMachineStatisticsEvent.class, listenerRegistry);
+    }
+
+    @Deactivate
+    public void deactivate() {
+        eventDispatcher.removeSink(AaaMachineStatisticsEvent.class);
+    }
+
+    @Override
+    public AaaSupplicantMachineStats getSupplicantStats(Object obj) {
+        StateMachine stateMachine = null;
+        AaaSupplicantMachineStats stats = new AaaSupplicantMachineStats();
+        try {
+            stateMachine = (StateMachine) obj;
+        } catch (ClassCastException e) {
+            log.debug("casting exception detected for StateMachine.");
+            return null;
+        }
+        log.debug("capturing supplicant machine stat from authentication session");
+        stats.setTotalPacketsSent(stateMachine.totalPacketsSent());
+        stats.setTotalPacketsRecieved(stateMachine.totalPacketsReceived());
+        stats.setTotalFramesSent(stateMachine.totalPacketsSent());
+        stats.setTotalFramesReceived(stateMachine.totalPacketsReceived());
+        stats.setSrcMacAddress(stateMachine.supplicantAddress() == null ? ""
+                : stateMachine.supplicantAddress().toString());
+        stats.setSessionName(stateMachine.username() == null ? ""
+                : new String(stateMachine.username()));
+        stats.setSessionId(stateMachine.sessionId());
+        stats.setSessionDuration(System.currentTimeMillis() - stateMachine.sessionStartTime());
+        stats.setEapolType(stateMachine.eapolType());
+        stats.setSessionTerminateReason(stateMachine.getSessionTerminateReason());
+
+        log.trace("EapolType" + " - " + stats.getEapolType());
+        log.trace("SessionDuration" + " - " + stats.getSessionDuration());
+        log.trace("SessionId" + " - " + stats.getSessionId());
+        log.trace("SessionName" + " - " + stats.getSessionName());
+        log.trace("SessionTerminateReason" + " - " + stats.getSessionTerminateReason());
+        log.trace("SrcMacAddress" + " - " + stats.getSrcMacAddress());
+        log.trace("TotalFramesReceived" + " - " + stats.getTotalFramesReceived());
+        log.trace("TotalFramesSent" + " - " + stats.getTotalFramesSent());
+        log.trace("TotalOctetRecieved" + " - " + stats.getTotalOctetRecieved());
+        log.trace("TotalOctetSent" + " - " + stats.getTotalOctetSent());
+        log.trace("TotalPacketsSent" + " - " + stats.getTotalPacketsSent());
+        log.trace("TotalOctetRecieved" + " - " + stats.getTotalOctetRecieved());
+        return stats;
+    }
+
+    @Override
+    public void logAaaSupplicantMachineStats(AaaSupplicantMachineStats obj) {
+        log.trace("EapolType" + " - " + obj.getEapolType());
+        log.trace("SessionDuration" + " - " + obj.getSessionDuration());
+        log.trace("SessionId" + " - " + obj.getSessionId());
+        log.trace("SessionName" + " - " + obj.getSessionName());
+        log.trace("SessionTerminateReason" + " - " + obj.getSessionTerminateReason());
+        log.trace("SrcMacAddress" + " - " + obj.getSrcMacAddress());
+        log.trace("TotalFramesReceived" + " - " + obj.getTotalFramesReceived());
+        log.trace("TotalFramesSent" + " - " + obj.getTotalFramesSent());
+        log.trace("TotalOctetRecieved" + " - " + obj.getTotalOctetRecieved());
+        log.trace("TotalOctetSent" + " - " + obj.getTotalOctetSent());
+        log.trace("TotalPacketsSent" + " - " + obj.getTotalPacketsSent());
+        log.trace("TotalOctetRecieved" + " - " + obj.getTotalOctetRecieved());
+    }
+
+    @Override
+    public AaaMachineStatisticsDelegate getMachineStatsDelegate() {
+        return machineStatDelegate;
+    }
+
+    private class InternalMachineStatDelegate implements AaaMachineStatisticsDelegate {
+        @Override
+        public void notify(AaaMachineStatisticsEvent aaaMachineStatisticsEvent) {
+            log.debug("Supplicant Statistics event {} for {}", aaaMachineStatisticsEvent.type(),
+                    aaaMachineStatisticsEvent.subject());
+            post(aaaMachineStatisticsEvent);
+        }
+    }
+
+}
diff --git a/app/src/main/java/org/opencord/aaa/impl/StateMachine.java b/app/src/main/java/org/opencord/aaa/impl/StateMachine.java
index 7635d86..16543df 100644
--- a/app/src/main/java/org/opencord/aaa/impl/StateMachine.java
+++ b/app/src/main/java/org/opencord/aaa/impl/StateMachine.java
@@ -67,6 +67,46 @@
     private MacAddress supplicantAddress;
     private short vlanId;
     private byte priorityCode;
+    private long sessionStartTime;
+    private String eapolTypeVal;
+
+    public enum EapolType {
+        EAPOL_PACKET("EAPOL_PACKET"),
+        EAPOL_START("EAPOL_START"),
+        EAPOL_LOGOFF("EAPOL_LOGOFF"),
+        EAPOL_KEY("EAPOL_KEY"),
+        EAPOL_ASF("EAPOL_ASF");
+
+        private final String eaptype;
+
+        private EapolType(String value) {
+            this.eaptype = value;
+        }
+    };
+
+    private String sessionTerminateReason;
+
+    public enum SessionTerminationReasons {
+        SUPPLICANT_LOGOFF("SUPPLICANT_LOGOFF"),
+        TIME_OUT("TIME_OUT"),
+        PORT_REMOVED("PORT_REMOVED");
+
+        private final String reason;
+
+        private SessionTerminationReasons(String value) {
+            this.reason = value;
+        }
+
+        public String getReason() {
+            return this.reason;
+        }
+    };
+
+    // Supplicant packet count
+    private int totalPacketsSent;
+    private int totalPacketsReceived;
+    private int totalOctetSent;
+    private int totalOctetReceived;
 
     // Boolean flag indicating whether response is pending from AAA Server.
     // Used for counting timeout happening for AAA Sessions due to no response.
@@ -317,6 +357,86 @@
     }
 
     /**
+     * Gets the session start time.
+     *
+     * @return session start time
+     */
+    public long sessionStartTime() {
+        return sessionStartTime;
+    }
+
+    /**
+     * Sets the session start time.
+     *
+     * @param sessionStartTime new session start time
+     */
+    public void setSessionStartTime(long sessionStartTime) {
+        this.sessionStartTime = sessionStartTime;
+    }
+
+    /**
+     * returns eapol Type.
+     *
+     * @return eapolTypeVal.
+     */
+    public String eapolType() {
+        return this.eapolTypeVal;
+    }
+
+    /**
+     * Sets eapol Type name from eapol value.
+     *
+     * @param value eapol type as byte.
+     */
+    public void setEapolTypeVal(byte value) {
+        switch (value) {
+            case (byte) 0: this.eapolTypeVal = EapolType.EAPOL_PACKET.eaptype;
+                break;
+            case (byte) 1: this.eapolTypeVal = EapolType.EAPOL_START.eaptype;
+                break;
+            case (byte) 2: this.eapolTypeVal = EapolType.EAPOL_LOGOFF.eaptype;
+                break;
+            case (byte) 3: this.eapolTypeVal = EapolType.EAPOL_KEY.eaptype;
+                break;
+            case (byte) 4: this.eapolTypeVal = EapolType.EAPOL_ASF.eaptype;
+                break;
+            default : this.eapolTypeVal = "INVALID TYPE";
+        }
+    }
+
+    public String getSessionTerminateReason() {
+        return sessionTerminateReason;
+    }
+
+    public void setSessionTerminateReason(String sessionTerminateReason) {
+        this.sessionTerminateReason = sessionTerminateReason;
+    }
+
+    public int totalPacketsReceived() {
+        return this.totalPacketsReceived;
+    }
+
+    public void incrementTotalPacketsReceived() {
+        this.totalPacketsReceived = this.totalPacketsReceived + 1;
+    }
+
+    public int totalPacketsSent() {
+        return this.totalPacketsSent;
+    }
+
+    public void incrementTotalPacketsSent() {
+        this.totalPacketsSent = this.totalPacketsSent + 1;
+    }
+
+    public void incrementTotalOctetReceived(short packetLen) {
+        this.totalOctetReceived = this.totalOctetReceived + packetLen;
+    }
+
+    public void incrementTotalOctetSent(short packetLen) {
+        this.totalOctetSent = this.totalOctetSent + packetLen;
+    }
+
+    /**
      * Gets the client id that is requesting for access.
      *
      * @return The client id.