diff --git a/voltha/adapters/openolt/openolt_device.py b/voltha/adapters/openolt/openolt_device.py
index d3c55f9..2bc036a 100644
--- a/voltha/adapters/openolt/openolt_device.py
+++ b/voltha/adapters/openolt/openolt_device.py
@@ -25,6 +25,7 @@
 from voltha.adapters.openolt.openolt_utils import OpenoltUtils
 from voltha.adapters.openolt.openolt_grpc import OpenoltGrpc
 from voltha.adapters.openolt.openolt_indications import OpenoltIndications
+from voltha.adapters.openolt.openolt_packet import OpenoltPacket
 from voltha.adapters.openolt.openolt_kafka_admin import KAdmin
 
 
@@ -111,6 +112,8 @@
 
     def do_state_init(self, event):
         self.log.debug('init')
+        self._packet = OpenoltPacket(self)
+        self._packet.start()
         self._indications = OpenoltIndications(self)
         self._indications.start()
 
diff --git a/voltha/adapters/openolt/openolt_grpc.py b/voltha/adapters/openolt/openolt_grpc.py
index 54c0500..a7cfd07 100644
--- a/voltha/adapters/openolt/openolt_grpc.py
+++ b/voltha/adapters/openolt/openolt_grpc.py
@@ -20,8 +20,7 @@
 import time
 from simplejson import dumps
 from twisted.internet import reactor
-from google.protobuf.json_format import MessageToJson
-from voltha.northbound.kafka.kafka_proxy import get_kafka_proxy
+from voltha.northbound.kafka.kafka_proxy import kafka_send_pb
 from voltha.adapters.openolt.protos import openolt_pb2_grpc, openolt_pb2
 
 
@@ -53,22 +52,6 @@
 
     def indications_thread(self):
 
-        def forward_indication(topic, msg):
-            try:
-                self.log.debug('forward indication', topic=topic, msg=msg)
-                kafka_proxy = get_kafka_proxy()
-                if kafka_proxy and not kafka_proxy.is_faulty():
-                    self.log.debug('kafka-proxy-available')
-                    kafka_proxy.send_message(
-                        topic,
-                        dumps(MessageToJson(
-                            msg,
-                            including_default_value_fields=True)))
-                else:
-                    self.log.error('kafka-proxy-unavailable')
-            except Exception, e:
-                self.log.exception('failed-sending-message', e=e)
-
         self.log.debug('openolt grpc connecting to olt')
 
         self.stub = openolt_pb2_grpc.OpenoltStub(self.channel)
@@ -126,4 +109,4 @@
                                       indications=ind)
                         continue
 
-                forward_indication("openolt.ind", ind)
+                kafka_send_pb("openolt.ind", ind)
diff --git a/voltha/adapters/openolt/openolt_packet.py b/voltha/adapters/openolt/openolt_packet.py
new file mode 100644
index 0000000..cfd838c
--- /dev/null
+++ b/voltha/adapters/openolt/openolt_packet.py
@@ -0,0 +1,134 @@
+#
+# Copyright 2019 the original author or authors.
+#
+# 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.
+#
+
+import threading
+from google.protobuf.json_format import Parse
+from simplejson import loads
+import structlog
+from scapy.layers.l2 import Ether, Dot1Q
+import binascii
+
+from common.frameio.frameio import hexify
+from voltha.protos.openflow_13_pb2 import PacketOut
+from voltha.adapters.openolt.openolt_kafka_consumer import KConsumer
+from voltha.core.flow_decomposer import OUTPUT
+from voltha.protos.device_pb2 import Port
+from voltha.adapters.openolt.protos import openolt_pb2
+
+
+class OpenoltPacket(object):
+    def __init__(self, device):
+        self.log = structlog.get_logger()
+        self.device = device
+        self.packet_thread_handle = threading.Thread(
+            target=self.packet_thread)
+        self.packet_thread_handle.setDaemon(True)
+
+    def start(self):
+        self.packet_thread_handle.start()
+
+    def stop(self):
+        pass
+
+    def packet_thread(self):
+        self.log.debug('openolt packet-out thread starting')
+        KConsumer(self.packet_process, 'voltha.pktout')
+
+    def packet_process(self, topic, msg):
+
+        def get_port_out(opo):
+            for action in opo.actions:
+                if action.type == OUTPUT:
+                    return action.output.port
+
+        pb = Parse(loads(msg), PacketOut(), ignore_unknown_fields=True)
+
+        logical_device_id = pb.id
+        ofp_packet_out = pb.packet_out
+
+        self.log.debug("received packet-out form kafka",
+                       logical_device_id=logical_device_id,
+                       ofp_packet_out=ofp_packet_out)
+
+        egress_port = get_port_out(ofp_packet_out)
+        msg = ofp_packet_out.data
+
+        self.log.debug('rcv-packet-out', logical_device_id=logical_device_id,
+                       egress_port=egress_port,
+                       # adapter_name=self.adapter_name,
+                       data=hexify(msg))
+
+        pkt = Ether(msg)
+        self.log.debug('packet out', egress_port=egress_port,
+                       packet=str(pkt).encode("HEX"))
+
+        # Find port type
+        egress_port_type = self.device.platform \
+            .intf_id_to_port_type_name(egress_port)
+
+        if egress_port_type == Port.ETHERNET_UNI:
+
+            if pkt.haslayer(Dot1Q):
+                outer_shim = pkt.getlayer(Dot1Q)
+                if isinstance(outer_shim.payload, Dot1Q):
+                    # If double tag, remove the outer tag
+                    payload = (
+                            Ether(src=pkt.src, dst=pkt.dst,
+                                  type=outer_shim.type) /
+                            outer_shim.payload
+                    )
+                else:
+                    payload = pkt
+            else:
+                payload = pkt
+
+            send_pkt = binascii.unhexlify(str(payload).encode("HEX"))
+
+            self.log.debug(
+                'sending-packet-to-ONU', egress_port=egress_port,
+                intf_id=self.device.platform.intf_id_from_uni_port_num(
+                    egress_port),
+                onu_id=self.device.platform.onu_id_from_port_num(egress_port),
+                uni_id=self.device.platform.uni_id_from_port_num(egress_port),
+                port_no=egress_port,
+                packet=str(payload).encode("HEX"))
+
+            onu_pkt = openolt_pb2.OnuPacket(
+                intf_id=self.device.platform.intf_id_from_uni_port_num(
+                    egress_port),
+                onu_id=self.device.platform.onu_id_from_port_num(egress_port),
+                port_no=egress_port,
+                pkt=send_pkt)
+
+            self.device._grpc.stub.OnuPacketOut(onu_pkt)
+
+        elif egress_port_type == Port.ETHERNET_NNI:
+            self.log.debug('sending-packet-to-uplink', egress_port=egress_port,
+                           packet=str(pkt).encode("HEX"))
+
+            send_pkt = binascii.unhexlify(str(pkt).encode("HEX"))
+
+            uplink_pkt = openolt_pb2.UplinkPacket(
+                intf_id=self.device.platform.intf_id_from_nni_port_num(
+                    egress_port),
+                pkt=send_pkt)
+
+            self.device._grpc.stub.UplinkPacketOut(uplink_pkt)
+
+        else:
+            self.log.warn('Packet-out-to-this-interface-type-not-implemented',
+                          egress_port=egress_port,
+                          port_type=egress_port_type)
diff --git a/voltha/core/local_handler.py b/voltha/core/local_handler.py
index 39f550e..6daef6e 100644
--- a/voltha/core/local_handler.py
+++ b/voltha/core/local_handler.py
@@ -41,6 +41,7 @@
 from voltha.protos.omci_alarm_db_pb2 import AlarmDeviceData
 from requests.api import request
 from common.utils.asleep import asleep
+from voltha.northbound.kafka.kafka_proxy import kafka_send_pb
 
 log = structlog.get_logger()
 
@@ -1116,8 +1117,17 @@
             agent = self.core.get_logical_device_agent(packet_out.id)
             agent.packet_out(packet_out.packet_out)
 
-        for request in request_iterator:
-            forward_packet_out(packet_out=request)
+        for req in request_iterator:
+            device_agent = self.core.get_logical_device_agent(req.id)
+            adapter_name = device_agent.device_adapter_agent.name
+
+            if adapter_name == 'openolt':
+                log.debug('fast path pkt-out to kafka')
+                # topic = 'openolt.pktout:{}'.format(req.id)
+                topic = 'voltha.pktout'
+                kafka_send_pb(topic, req)
+            else:
+                forward_packet_out(packet_out=req)
 
         log.debug('stop-stream-packets-out')
 
diff --git a/voltha/northbound/kafka/kafka_proxy.py b/voltha/northbound/kafka/kafka_proxy.py
index 0fba534..08d42f1 100644
--- a/voltha/northbound/kafka/kafka_proxy.py
+++ b/voltha/northbound/kafka/kafka_proxy.py
@@ -15,18 +15,20 @@
 #
 
 
-from confluent_kafka import Producer as _kafkaProducer
+import threading
 from structlog import get_logger
 from twisted.internet import reactor
 from twisted.internet.defer import inlineCallbacks, returnValue
 from twisted.internet.threads import deferToThread
 from zope.interface import implementer
+from google.protobuf.json_format import MessageToJson
+from simplejson import dumps
+from confluent_kafka import Producer as _kafkaProducer
+from confluent_kafka import Consumer, KafkaError
 
 from common.utils.consulhelpers import get_endpoint_from_consul
 from voltha.northbound.kafka.event_bus_publisher import EventBusPublisher
 from voltha.registry import IComponent
-from confluent_kafka import Consumer, KafkaError
-import threading
 
 log = get_logger()
 
@@ -336,3 +338,19 @@
 # Common method to get the singleton instance of the kafka proxy class
 def get_kafka_proxy():
     return KafkaProxy._kafka_instance
+
+def kafka_send_pb(topic, msg):
+    try:
+        log.debug('send protobuf to kafka', topic=topic, msg=msg)
+        kafka_proxy = get_kafka_proxy()
+        if kafka_proxy and not kafka_proxy.is_faulty():
+            log.debug('kafka-proxy-available')
+            kafka_proxy.send_message(
+                topic,
+                dumps(MessageToJson(
+                    msg,
+                    including_default_value_fields=True)))
+        else:
+            log.error('kafka-proxy-unavailable')
+    except Exception, e:
+        log.exception('failed-sending-protobuf', e=e)