diff --git a/agent/Makefile.in b/agent/Makefile.in
index 25aefc5..c62baf8 100644
--- a/agent/Makefile.in
+++ b/agent/Makefile.in
@@ -43,7 +43,7 @@
 # This specifies the GIT tag in https://github.com/opencord/voltha-protos
 # repo that we need to refer to, to pick the right version of
 # openolt.proto and tech_profile.proto
-OPENOLT_PROTO_VER ?= v4.1.5
+OPENOLT_PROTO_VER ?= v4.2.0
 
 # Variables used for Inband build
 INBAND = "n"
diff --git a/agent/device/asfvolt16/vendor.h b/agent/device/asfvolt16/vendor.h
index ffbaec4..7cadd9b 100644
--- a/agent/device/asfvolt16/vendor.h
+++ b/agent/device/asfvolt16/vendor.h
@@ -27,17 +27,19 @@
 
 #define ONU_ID_START 1
 #define ONU_ID_END 32
-#define MAX_ONUS_PER_PON 32
+#define MAX_ONUS_PER_PON (ONU_ID_END - ONU_ID_START + 1)
 
+#define MAX_ALLOC_ID_PER_ONU 4
 #define ALLOC_ID_START 1024
-#define ALLOC_ID_END 16383
+#define ALLOC_ID_END (ALLOC_ID_START + MAX_ONUS_PER_PON * MAX_ALLOC_ID_PER_ONU)
 
+#define GEM_PORT_ID_PER_ALLOC_ID 8
 #define GEM_PORT_ID_START 1024
-#define GEM_PORT_ID_END 65535
+#define GEM_PORT_ID_END (GEM_PORT_ID_START + MAX_ONUS_PER_PON * MAX_ALLOC_ID_PER_ONU * GEM_PORT_ID_PER_ALLOC_ID)
 
 #define FLOW_ID_START 1
 #define FLOW_ID_END 65535
-#define MAX_FLOW_ID 65535
+#define MAX_FLOW_ID FLOW_ID_END
 #define INVALID_FLOW_ID 0
 
 #endif
diff --git a/agent/device/asgvolt64/vendor.h b/agent/device/asgvolt64/vendor.h
index 44dc9a2..3b5c844 100644
--- a/agent/device/asgvolt64/vendor.h
+++ b/agent/device/asgvolt64/vendor.h
@@ -27,13 +27,15 @@
 
 #define ONU_ID_START 1
 #define ONU_ID_END 32
-#define MAX_ONUS_PER_PON ONU_ID_END
+#define MAX_ONUS_PER_PON (ONU_ID_END - ONU_ID_START + 1)
 
+#define MAX_ALLOC_ID_PER_ONU 4
 #define ALLOC_ID_START 256
-#define ALLOC_ID_END 767
+#define ALLOC_ID_END (ALLOC_ID_START + MAX_ONUS_PER_PON * MAX_ALLOC_ID_PER_ONU)
 
+#define GEM_PORT_ID_PER_ALLOC_ID 8
 #define GEM_PORT_ID_START 256
-#define GEM_PORT_ID_END 4095
+#define GEM_PORT_ID_END (GEM_PORT_ID_START + MAX_ONUS_PER_PON * MAX_ALLOC_ID_PER_ONU * GEM_PORT_ID_PER_ALLOC_ID)
 
 #define FLOW_ID_START 1
 #define FLOW_ID_END 65535
diff --git a/agent/device/rlt-3200g-w/vendor.h b/agent/device/rlt-3200g-w/vendor.h
index 4c77e73..e77c183 100644
--- a/agent/device/rlt-3200g-w/vendor.h
+++ b/agent/device/rlt-3200g-w/vendor.h
@@ -27,13 +27,15 @@
 
 #define ONU_ID_START 1
 #define ONU_ID_END 32
-#define MAX_ONUS_PER_PON ONU_ID_END
+#define MAX_ONUS_PER_PON (ONU_ID_END - ONU_ID_START + 1)
 
+#define MAX_ALLOC_ID_PER_ONU 4
 #define ALLOC_ID_START 256
-#define ALLOC_ID_END 767
+#define ALLOC_ID_END (ALLOC_ID_START + MAX_ONUS_PER_PON * MAX_ALLOC_ID_PER_ONU)
 
+#define GEM_PORT_ID_PER_ALLOC_ID 8
 #define GEM_PORT_ID_START 256
-#define GEM_PORT_ID_END 4095
+#define GEM_PORT_ID_END (GEM_PORT_ID_START + MAX_ONUS_PER_PON * MAX_ALLOC_ID_PER_ONU * GEM_PORT_ID_PER_ALLOC_ID)
 
 #define FLOW_ID_START 1
 #define FLOW_ID_END 65535
diff --git a/agent/src/core_api_handler.cc b/agent/src/core_api_handler.cc
index 0e195d9..1579a4f 100644
--- a/agent/src/core_api_handler.cc
+++ b/agent/src/core_api_handler.cc
@@ -270,7 +270,7 @@
             pool->set_type(::openolt::DeviceInfo::DeviceResourceRanges::Pool::ALLOC_ID);
             pool->set_sharing(::openolt::DeviceInfo::DeviceResourceRanges::Pool::DEDICATED_PER_INTF);
             pool->set_start(ALLOC_ID_START);
-            pool->set_end(ALLOC_ID_START);
+            pool->set_end(ALLOC_ID_END);
 
             pool = range->add_pools();
             pool->set_type(::openolt::DeviceInfo::DeviceResourceRanges::Pool::GEMPORT_ID);
@@ -365,6 +365,9 @@
         bcmos_fastlock_init(&gem_cfg_wait_lock, 0);
         bcmos_fastlock_init(&onu_deactivate_wait_lock, 0);
         bcmos_fastlock_init(&acl_packet_trap_handler_lock, 0);
+        bcmos_fastlock_init(&symmetric_datapath_flow_id_lock, 0);
+        bcmos_fastlock_init(&pon_gem_to_onu_uni_map_lock, 0);
+
 
         OPENOLT_LOG(INFO, openolt_log_id, "Enable OLT - %s-%s\n", VENDOR_ID, MODEL_ID);
 
@@ -1523,12 +1526,13 @@
 
 Status FlowAddWrapper_(const ::openolt::Flow* request) {
 
+    Status st = Status::OK;
     int32_t access_intf_id = request->access_intf_id();
     int32_t onu_id = request->onu_id();
     int32_t uni_id = request->uni_id();
     uint32_t port_no = request->port_no();
     uint64_t voltha_flow_id = request->flow_id();
-    uint64_t symmetric_voltha_flow_id = request->symmetric_flow_id();
+    uint64_t symmetric_voltha_flow_id = 0;
     const std::string flow_type = request->flow_type();
     int32_t alloc_id = request->alloc_id();
     int32_t network_intf_id = request->network_intf_id();
@@ -1544,6 +1548,25 @@
     const google::protobuf::Map<unsigned int, bool> &gemport_to_aes = request->gemport_to_aes();
     uint16_t flow_id;
     bool enable_encryption;
+    // When following conditions are ALL met, it qualifies as datapath flow.
+    // 1. valid access_intf_id, onu_id, uni_id
+    // 2. Valid tech_profile_id
+    // 3. flow_type that is not MULTICAST
+    // 4. Not a trap-to-host flow.
+    bool datapathFlow = access_intf_id >= 0 && onu_id >= 0 && uni_id >= 0 && tech_profile_id > 0
+                        && flow_type != multicast && !action.cmd().trap_to_host();
+
+    if (datapathFlow) {
+        const std::string inverse_flow_type = flow_type.compare(upstream) == 0 ? downstream : upstream;
+        symmetric_datapath_flow_id_map_key key(access_intf_id, onu_id, uni_id, tech_profile_id, inverse_flow_type);
+        // Find the onu-uni mapping for the pon-gem key
+        bcmos_fastlock_lock(&symmetric_datapath_flow_id_lock);
+        auto it = symmetric_datapath_flow_id_map.find(key);
+        bcmos_fastlock_unlock(&symmetric_datapath_flow_id_lock, 0);
+        if (it != symmetric_datapath_flow_id_map.end()) {
+            symmetric_voltha_flow_id = it->second;
+        }
+    }
 
     // The intf_id variable defaults to access(PON) interface ID.
     // For trap-from-nni flow where access interface ID is not valid , change it to NNI interface ID
@@ -1576,25 +1599,27 @@
         }
 
         if (!replicate_flow) {  // No flow replication
-                flow_id = dev_fl_symm_params[0].flow_id;
-                gemport_id = dev_fl_symm_params[0].gemport_id; // overwrite the gemport with symmetric flow gemport
-                                                               // Should be same as what is coming in this request.
-                enable_encryption = get_aes_flag_for_gem_port(gemport_to_aes, gemport_id);
-                ::openolt::Classifier cl = ::openolt::Classifier(classifier);
-                cl.set_o_pbits(dev_fl_symm_params[0].pbit);
-                Status st = FlowAdd_(access_intf_id, onu_id, uni_id, port_no, flow_id,
-                                    flow_type, alloc_id, network_intf_id, gemport_id, cl,
-                                    action, priority, cookie, group_id, tech_profile_id, enable_encryption);
-                if (st.error_code() == grpc::StatusCode::OK) {
-                    device_flow dev_fl;
-                    dev_fl.is_flow_replicated = false;
-                    dev_fl.symmetric_voltha_flow_id = symmetric_voltha_flow_id;
-                    dev_fl.voltha_flow_id = voltha_flow_id;
-                    memcpy(dev_fl.params, dev_fl_symm_params, sizeof(device_flow_params));
-                    // update voltha flow to cache
-                    update_voltha_flow_to_cache(voltha_flow_id, dev_fl);
-                }
+            flow_id = dev_fl_symm_params[0].flow_id;
+            gemport_id = dev_fl_symm_params[0].gemport_id; // overwrite the gemport with symmetric flow gemport
+                                                           // Should be same as what is coming in this request.
+            enable_encryption = get_aes_flag_for_gem_port(gemport_to_aes, gemport_id);
+            ::openolt::Classifier cl = ::openolt::Classifier(classifier);
+            cl.set_o_pbits(dev_fl_symm_params[0].pbit);
+            st = FlowAdd_(access_intf_id, onu_id, uni_id, port_no, flow_id,
+                                flow_type, alloc_id, network_intf_id, gemport_id, cl,
+                                action, priority, cookie, group_id, tech_profile_id, enable_encryption);
+            if (st.error_code() != grpc::StatusCode::OK && st.error_code() != grpc::StatusCode::ALREADY_EXISTS) {
+                OPENOLT_LOG(ERROR, openolt_log_id, "failed to install device flow=%u for voltha flow=%lu", flow_id, voltha_flow_id);
                 return st;
+            }
+
+            device_flow dev_fl;
+            dev_fl.is_flow_replicated = false;
+            dev_fl.symmetric_voltha_flow_id = symmetric_voltha_flow_id;
+            dev_fl.voltha_flow_id = voltha_flow_id;
+            memcpy(dev_fl.params, dev_fl_symm_params, sizeof(device_flow_params));
+            // update voltha flow to cache
+            update_voltha_flow_to_cache(voltha_flow_id, dev_fl);
         } else { // Flow to be replicated
             OPENOLT_LOG(INFO, openolt_log_id,"symmetric flow and replication is needed\n");
             for (uint8_t i=0; i<NUMBER_OF_REPLICATED_FLOWS; i++) {
@@ -1603,7 +1628,7 @@
                 gemport_id = dev_fl_symm_params[i].gemport_id;
                 enable_encryption = get_aes_flag_for_gem_port(gemport_to_aes, gemport_id);
                 cl.set_o_pbits(dev_fl_symm_params[i].pbit);
-                Status st = FlowAdd_(access_intf_id, onu_id, uni_id, port_no, flow_id,
+                st = FlowAdd_(access_intf_id, onu_id, uni_id, port_no, flow_id,
                                     flow_type, alloc_id, network_intf_id, gemport_id, cl,
                                     action, priority, cookie, group_id, tech_profile_id, enable_encryption);
                 if (st.error_code() != grpc::StatusCode::OK && st.error_code() != grpc::StatusCode::ALREADY_EXISTS) {
@@ -1635,7 +1660,7 @@
                 return ::Status(grpc::StatusCode::RESOURCE_EXHAUSTED, "flow-ids-exhausted");
             }
             enable_encryption = get_aes_flag_for_gem_port(gemport_to_aes, gemport_id);
-            Status st = FlowAdd_(access_intf_id, onu_id, uni_id, port_no, flow_id,
+            st = FlowAdd_(access_intf_id, onu_id, uni_id, port_no, flow_id,
                                 flow_type, alloc_id, network_intf_id, gemport_id, classifier,
                                 action, priority, cookie, group_id, tech_profile_id, enable_encryption);
             if (st.error_code() == grpc::StatusCode::OK) {
@@ -1651,8 +1676,8 @@
             } else {
                 // Free the flow id on failure
                 free_flow_id(flow_id);
+                return st;
             }
-            return st;
         } else { // Flow to be replicated
             OPENOLT_LOG(INFO, openolt_log_id,"not a symmetric flow and replication is needed\n");
             if (pbit_to_gemport.size() != NUMBER_OF_PBITS) {
@@ -1676,7 +1701,7 @@
                     gemport_id = dev_fl.params[cnt].gemport_id;
                     enable_encryption = get_aes_flag_for_gem_port(gemport_to_aes, gemport_id);
                     cl.set_o_pbits(dev_fl.params[cnt].pbit);
-                    Status st = FlowAdd_(access_intf_id, onu_id, uni_id, port_no, flow_id,
+                    st = FlowAdd_(access_intf_id, onu_id, uni_id, port_no, flow_id,
                                         flow_type, alloc_id, network_intf_id, gemport_id, cl,
                                         action, priority, cookie, group_id, tech_profile_id, enable_encryption);
                     if (st.error_code() != grpc::StatusCode::OK) {
@@ -1703,7 +1728,15 @@
         }
     }
 
-    return Status::OK;
+    if (datapathFlow) {
+        // Create the pon-gem to onu-uni mapping
+        symmetric_datapath_flow_id_map_key key(access_intf_id, onu_id, uni_id, tech_profile_id, flow_type);
+        bcmos_fastlock_lock(&symmetric_datapath_flow_id_lock);
+        symmetric_datapath_flow_id_map[key] = voltha_flow_id;
+        bcmos_fastlock_unlock(&symmetric_datapath_flow_id_lock, 0);
+    }
+
+    return st;
 }
 
 
@@ -2084,6 +2117,10 @@
 }
 
 Status FlowRemoveWrapper_(const ::openolt::Flow* request) {
+    int32_t access_intf_id = request->access_intf_id();
+    int32_t onu_id = request->onu_id();
+    int32_t uni_id = request->uni_id();
+    uint32_t tech_profile_id = request->tech_profile_id();
     const std::string flow_type = request->flow_type();
     uint64_t voltha_flow_id = request->flow_id();
     Status st;
@@ -2116,6 +2153,13 @@
     }
     // remove the flow from cache on voltha flow removal
     remove_voltha_flow_from_cache(voltha_flow_id);
+
+    symmetric_datapath_flow_id_map_key key(access_intf_id, onu_id, uni_id, tech_profile_id, flow_type);
+    // Remove onu-uni mapping for the pon-gem key
+    bcmos_fastlock_lock(&symmetric_datapath_flow_id_lock);
+    symmetric_datapath_flow_id_map.erase(key);
+    bcmos_fastlock_unlock(&symmetric_datapath_flow_id_lock, 0);
+
     return Status::OK;
 }
 
@@ -2769,6 +2813,14 @@
             OPENOLT_LOG(ERROR, openolt_log_id, "failed to created gemport=%d, access_intf=%d, onu_id=%d\n", gemport_id, access_intf_id, onu_id);
             return BCM_ERR_INTERNAL;
         }
+        if (direction == upstream) {
+            // Create the pon-gem to onu-uni mapping
+            pon_gem pg(access_intf_id, gemport_id);
+            onu_uni ou(onu_id, uni_id);
+            bcmos_fastlock_lock(&pon_gem_to_onu_uni_map_lock);
+            pon_gem_to_onu_uni_map[pg] = ou;
+            bcmos_fastlock_unlock(&pon_gem_to_onu_uni_map_lock, 0);
+        }
     }
 
     OPENOLT_LOG(INFO, openolt_log_id, "Created tm_queue, direction %s, id %d, sched_id %d, tm_q_set_id %d, \
@@ -2852,6 +2904,13 @@
             OPENOLT_LOG(ERROR, openolt_log_id, "failed to remove gemport=%d, access_intf=%d, onu_id=%d\n", gemport_id, access_intf_id, onu_id);
             return BCM_ERR_INTERNAL;
         }
+        if (direction == upstream) {
+            // Remove the pon-gem to onu-uni mapping
+            pon_gem pg(access_intf_id, gemport_id);
+            bcmos_fastlock_lock(&pon_gem_to_onu_uni_map_lock);
+            pon_gem_to_onu_uni_map.erase(pg);
+            bcmos_fastlock_unlock(&pon_gem_to_onu_uni_map_lock, 0);
+        }
     }
 
     if (direction == downstream) {
diff --git a/agent/src/core_data.cc b/agent/src/core_data.cc
index 65fc62c..bd83e81 100644
--- a/agent/src/core_data.cc
+++ b/agent/src/core_data.cc
@@ -211,6 +211,14 @@
 std::map<uint64_t, device_flow> voltha_flow_to_device_flow;
 bcmos_fastlock voltha_flow_to_device_flow_lock;
 
+// Map of symmetric_datapath_flow_id_map_key to voltha-flow-id
+std::map<symmetric_datapath_flow_id_map_key, uint64_t> symmetric_datapath_flow_id_map;
+bcmos_fastlock symmetric_datapath_flow_id_lock;
+
+// Map of {pon-port-id, gem-port-id} to {onu-id, uni-id}
+std::map<pon_gem, onu_uni> pon_gem_to_onu_uni_map;
+bcmos_fastlock pon_gem_to_onu_uni_map_lock;
+
 // Lock to protect critical section around handling data associated with ACL trap packet handling
 bcmos_fastlock acl_packet_trap_handler_lock;
 
diff --git a/agent/src/core_data.h b/agent/src/core_data.h
index 57b2fe2..576c248 100644
--- a/agent/src/core_data.h
+++ b/agent/src/core_data.h
@@ -205,6 +205,13 @@
     double rx_power_mean_dbm;
 } onu_rssi_complete_result;
 
+// Key for map used for tracking symmetric datapath flows ==> <pon, onu, uni, tp_id, flow_type>
+typedef std::tuple<int32_t, int32_t, int32_t, uint32_t, std::string> symmetric_datapath_flow_id_map_key;
+
+// Key and value for the pon_gem_to_onu_uni_map
+typedef std::tuple<uint32_t, uint32_t> pon_gem;
+typedef std::tuple<uint32_t, uint32_t> onu_uni;
+
 // *******************************************************//
 // Extern Variable/Constant declarations used by the core //
 // *******************************************************//
@@ -378,6 +385,12 @@
 extern std::map<uint64_t, device_flow> voltha_flow_to_device_flow;
 extern bcmos_fastlock voltha_flow_to_device_flow_lock;
 
+extern std::map<symmetric_datapath_flow_id_map_key, uint64_t> symmetric_datapath_flow_id_map;
+extern bcmos_fastlock symmetric_datapath_flow_id_lock;
+
+extern std::map<pon_gem, onu_uni> pon_gem_to_onu_uni_map;
+extern bcmos_fastlock pon_gem_to_onu_uni_map_lock;
+
 // Lock to protect critical section around handling data associated with ACL trap packet handling
 extern bcmos_fastlock acl_packet_trap_handler_lock;
 
diff --git a/agent/src/indications.cc b/agent/src/indications.cc
index 0407260..5261222 100644
--- a/agent/src/indications.cc
+++ b/agent/src/indications.cc
@@ -514,14 +514,28 @@
                     pkt_ind->set_intf_id((bcmolt_interface_id)pkt_data->interface_ref.intf_id);
                     pkt_ind->set_pkt(pkt_data->buffer.arr, pkt_data->buffer.len);
                     pkt_ind->set_gemport_id(pkt_data->svc_port_id);
+		    if (pkt_data->svc_port_id != BCMOLT_SERVICE_PORT_ID_INVALID) { // case of packet-in from the PON interface
+                        pon_gem pg((uint32_t)pkt_data->interface_ref.intf_id, pkt_data->svc_port_id);
+                        // Find to onu-uni mapping for the pon-gem key
+                        bcmos_fastlock_lock(&pon_gem_to_onu_uni_map_lock);
+                        auto it = pon_gem_to_onu_uni_map.find(pg);
+                        bcmos_fastlock_unlock(&pon_gem_to_onu_uni_map_lock, 0);
+                        if (it == pon_gem_to_onu_uni_map.end()) {
+                            bcmolt_msg_free(msg);
+                            OPENOLT_LOG(ERROR, openolt_log_id, "onu-uni reference not found for packet-in on gemport=%d, pon_intf_id=%d", pkt_data->svc_port_id,  pkt_data->interface_ref.intf_id);
+                            return;
+                        }
+                        pkt_ind->set_onu_id(std::get<0>(it->second));
+                        pkt_ind->set_uni_id(std::get<1>(it->second));
+                    }
                     ind.set_allocated_pkt_ind(pkt_ind);
 
                     if (pkt_data->interface_ref.intf_type == BCMOLT_INTERFACE_TYPE_PON) {
-                        OPENOLT_LOG(INFO, openolt_log_id, "packet indication, ingress intf_type %s, ingress intf_id %d, gem_port %d\n",
-                            pkt_ind->intf_type().c_str(), pkt_ind->intf_id(), pkt_data->svc_port_id);
+                        OPENOLT_LOG(INFO, openolt_log_id,"packet indication, ingress intf_type %s, ingress intf_id %d, gem_port %d, onu_id=%d, uni_id=%d\n",
+                            pkt_ind->intf_type().c_str(), pkt_ind->intf_id(), pkt_ind->gemport_id(), pkt_ind->onu_id(), pkt_ind->uni_id());
                     } else if (pkt_data->interface_ref.intf_type == BCMOLT_INTERFACE_TYPE_NNI ) {
-                        OPENOLT_LOG(INFO, openolt_log_id, "packet indication, ingress intf_type %s, ingress intf_id %d, gem_port %d \n",
-                            pkt_ind->intf_type().c_str(), pkt_ind->intf_id(), pkt_data->svc_port_id);
+                        OPENOLT_LOG(INFO, openolt_log_id, "packet indication, ingress intf_type %s, ingress intf_id %d\n",
+                            pkt_ind->intf_type().c_str(), pkt_ind->intf_id());
                     }
                 }
             }
diff --git a/agent/test/Makefile b/agent/test/Makefile
index 54bb780..a6e983d 100644
--- a/agent/test/Makefile
+++ b/agent/test/Makefile
@@ -21,7 +21,7 @@
 TOP_DIR=`pwd`
 OPENOLTDEVICE ?= asfvolt16
 
-OPENOLT_PROTO_VER ?= v4.1.5
+OPENOLT_PROTO_VER ?= v4.2.0
 
 ########################################################################
 ##