diff --git a/dash-pipeline/Makefile b/dash-pipeline/Makefile
index daf3a7713..1b57b62c1 100644
--- a/dash-pipeline/Makefile
+++ b/dash-pipeline/Makefile
@@ -5,6 +5,7 @@ ifeq ($(HAVE_DPAPP),y)
 DPAPP_LINK = veth4
 DPAPP_LINK_PEER = veth5
 SWITCH_PORT_DPAPP = --interface 2@$(DPAPP_LINK)
+D_DPAPP_CONNTRACK = -DDPAPP_CONNTRACK
 endif
 
 mkfile_path := $(abspath $(lastword $(MAKEFILE_LIST)))
@@ -129,6 +130,7 @@ $(P4_OUTDIR)/dash_pipeline_p4rt.txt: $(P4_SRC)
 	-w / \
 	$(DOCKER_P4C_BMV2_IMG) p4c-bm2-ss \
 		-DTARGET_BMV2_V1MODEL \
+		$(D_DPAPP_CONNTRACK) \
 	    $(P4_MAIN) \
         -o $(P4_OUTDIR)/dash_pipeline.json \
 	    --p4runtime-files $(P4_OUTDIR)/dash_pipeline_p4rt.json,$(P4_OUTDIR)/dash_pipeline_p4rt.txt \
diff --git a/dash-pipeline/SAI/specs/dash_flow.yaml b/dash-pipeline/SAI/specs/dash_flow.yaml
index 866fcbdb5..dacb55bca 100644
--- a/dash-pipeline/SAI/specs/dash_flow.yaml
+++ b/dash-pipeline/SAI/specs/dash_flow.yaml
@@ -95,12 +95,6 @@ sai_apis:
       type: sai_uint16_t
       objects: null
       valid_only: null
-    - !!python/object:utils.sai_spec.sai_struct_entry.SaiStructEntry
-      name: ip_proto
-      description: Exact matched key ip_proto
-      type: sai_uint8_t
-      objects: null
-      valid_only: null
     - !!python/object:utils.sai_spec.sai_struct_entry.SaiStructEntry
       name: src_ip
       description: Exact matched key src_ip
@@ -125,6 +119,12 @@ sai_apis:
       type: sai_uint16_t
       objects: null
       valid_only: null
+    - !!python/object:utils.sai_spec.sai_struct_entry.SaiStructEntry
+      name: ip_proto
+      description: Exact matched key ip_proto
+      type: sai_uint8_t
+      objects: null
+      valid_only: null
   attributes:
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
     name: SAI_FLOW_ENTRY_ATTR_ACTION
@@ -140,24 +140,11 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_VERSION
-    description: Action parameter version
-    type: sai_uint32_t
-    attr_value_field: u32
-    default: '0'
-    isresourcetype: false
-    flags: CREATE_AND_SET
-    object_name: null
-    allow_null: false
-    valid_only: null
-    is_vlan: false
-    deprecated: false
-  - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_DASH_DIRECTION
-    description: Action parameter DASH direction
-    type: sai_dash_direction_t
-    attr_value_field: s32
-    default: SAI_DASH_DIRECTION_INVALID
+    name: SAI_FLOW_ENTRY_ATTR_IS_UNIDIRECTIONAL
+    description: Action parameter is unidirectional
+    type: bool
+    attr_value_field: booldata
+    default: 'false'
     isresourcetype: false
     flags: CREATE_AND_SET
     object_name: null
@@ -166,11 +153,11 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_DASH_FLOW_ACTION
-    description: Action parameter DASH flow action
-    type: sai_dash_flow_action_t
-    attr_value_field: s32
-    default: SAI_DASH_FLOW_ACTION_NONE
+    name: SAI_FLOW_ENTRY_ATTR_STATE
+    description: Action parameter state
+    type: sai_uint8_t
+    attr_value_field: u8
+    default: '0'
     isresourcetype: false
     flags: CREATE_AND_SET
     object_name: null
@@ -179,8 +166,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_METER_CLASS
-    description: Action parameter meter class
+    name: SAI_FLOW_ENTRY_ATTR_VERSION
+    description: Action parameter version
     type: sai_uint32_t
     attr_value_field: u32
     default: '0'
@@ -192,24 +179,11 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_IS_UNIDIRECTIONAL_FLOW
-    description: Action parameter is unidirectional flow
-    type: bool
-    attr_value_field: booldata
-    default: 'false'
-    isresourcetype: false
-    flags: CREATE_AND_SET
-    object_name: null
-    allow_null: false
-    valid_only: null
-    is_vlan: false
-    deprecated: false
-  - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_ENI_MAC
-    description: Action parameter reverse flow ENI MAC
-    type: sai_mac_t
-    attr_value_field: mac
-    default: vendor
+    name: SAI_FLOW_ENTRY_ATTR_DASH_DIRECTION
+    description: Action parameter DASH direction
+    type: sai_dash_direction_t
+    attr_value_field: s32
+    default: SAI_DASH_DIRECTION_INVALID
     isresourcetype: false
     flags: CREATE_AND_SET
     object_name: null
@@ -218,8 +192,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_VNET_ID
-    description: Action parameter reverse flow VNET id
+    name: SAI_FLOW_ENTRY_ATTR_TUNNEL_ID
+    description: Action parameter tunnel id
     type: sai_uint16_t
     attr_value_field: u16
     default: '0'
@@ -231,10 +205,10 @@ sai_apis:
     is_vlan: true
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_IP_PROTO
-    description: Action parameter reverse flow IP protocol
-    type: sai_uint8_t
-    attr_value_field: u8
+    name: SAI_FLOW_ENTRY_ATTR_ROUTING_ACTIONS
+    description: Action parameter routing actions
+    type: sai_uint32_t
+    attr_value_field: u32
     default: '0'
     isresourcetype: false
     flags: CREATE_AND_SET
@@ -244,11 +218,11 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_SRC_IP
-    description: Action parameter reverse flow src IP
-    type: sai_ip_address_t
-    attr_value_field: ipaddr
-    default: 0.0.0.0
+    name: SAI_FLOW_ENTRY_ATTR_METER_CLASS
+    description: Action parameter meter class
+    type: sai_uint32_t
+    attr_value_field: u32
+    default: '0'
     isresourcetype: false
     flags: CREATE_AND_SET
     object_name: null
@@ -257,11 +231,11 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_DST_IP
-    description: Action parameter reverse flow dst IP
-    type: sai_ip_address_t
-    attr_value_field: ipaddr
-    default: 0.0.0.0
+    name: SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_VNI
+    description: Action parameter encap data VNI
+    type: sai_uint32_t
+    attr_value_field: u32
+    default: '0'
     isresourcetype: false
     flags: CREATE_AND_SET
     object_name: null
@@ -270,34 +244,34 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_SRC_PORT
-    description: Action parameter reverse flow src port
-    type: sai_uint16_t
-    attr_value_field: u16
+    name: SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_DEST_VNET_VNI
+    description: Action parameter encap data destination VNET VNI
+    type: sai_uint32_t
+    attr_value_field: u32
     default: '0'
     isresourcetype: false
     flags: CREATE_AND_SET
     object_name: null
     allow_null: false
     valid_only: null
-    is_vlan: true
+    is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_DST_PORT
-    description: Action parameter reverse flow dst port
-    type: sai_uint16_t
-    attr_value_field: u16
+    name: SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_UNDERLAY_SIP
+    description: Action parameter encap data underlay sip
+    type: sai_uint32_t
+    attr_value_field: u32
     default: '0'
     isresourcetype: false
     flags: CREATE_AND_SET
     object_name: null
     allow_null: false
     valid_only: null
-    is_vlan: true
+    is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_UNDERLAY0_VNET_ID
-    description: Action parameter underlay0 VNET id
+    name: SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_UNDERLAY_DIP
+    description: Action parameter encap data underlay dip
     type: sai_uint32_t
     attr_value_field: u32
     default: '0'
@@ -309,11 +283,11 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_UNDERLAY0_SIP
-    description: Action parameter underlay0 sip
-    type: sai_uint32_t
-    attr_value_field: u32
-    default: '0'
+    name: SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_UNDERLAY_SMAC
+    description: Action parameter encap data underlay source MAC
+    type: sai_mac_t
+    attr_value_field: mac
+    default: vendor
     isresourcetype: false
     flags: CREATE_AND_SET
     object_name: null
@@ -322,11 +296,11 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_UNDERLAY0_DIP
-    description: Action parameter underlay0 dip
-    type: sai_uint32_t
-    attr_value_field: u32
-    default: '0'
+    name: SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_UNDERLAY_DMAC
+    description: Action parameter encap data underlay destination MAC
+    type: sai_mac_t
+    attr_value_field: mac
+    default: vendor
     isresourcetype: false
     flags: CREATE_AND_SET
     object_name: null
@@ -335,8 +309,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_UNDERLAY0_DASH_ENCAPSULATION
-    description: Action parameter underlay0 DASH encapsulation
+    name: SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_DASH_ENCAPSULATION
+    description: Action parameter encap data DASH encapsulation
     type: sai_dash_encapsulation_t
     attr_value_field: s32
     default: SAI_DASH_ENCAPSULATION_INVALID
@@ -348,8 +322,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_UNDERLAY1_VNET_ID
-    description: Action parameter underlay1 VNET id
+    name: SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_VNI
+    description: Action parameter tunnel data VNI
     type: sai_uint32_t
     attr_value_field: u32
     default: '0'
@@ -361,8 +335,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_UNDERLAY1_SIP
-    description: Action parameter underlay1 sip
+    name: SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_DEST_VNET_VNI
+    description: Action parameter tunnel data destination VNET VNI
     type: sai_uint32_t
     attr_value_field: u32
     default: '0'
@@ -374,8 +348,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_UNDERLAY1_DIP
-    description: Action parameter underlay1 dip
+    name: SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_UNDERLAY_SIP
+    description: Action parameter tunnel data underlay sip
     type: sai_uint32_t
     attr_value_field: u32
     default: '0'
@@ -387,8 +361,21 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_UNDERLAY1_SMAC
-    description: Action parameter underlay1 source MAC
+    name: SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_UNDERLAY_DIP
+    description: Action parameter tunnel data underlay dip
+    type: sai_uint32_t
+    attr_value_field: u32
+    default: '0'
+    isresourcetype: false
+    flags: CREATE_AND_SET
+    object_name: null
+    allow_null: false
+    valid_only: null
+    is_vlan: false
+    deprecated: false
+  - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
+    name: SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_UNDERLAY_SMAC
+    description: Action parameter tunnel data underlay source MAC
     type: sai_mac_t
     attr_value_field: mac
     default: vendor
@@ -400,8 +387,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_UNDERLAY1_DMAC
-    description: Action parameter underlay1 destination MAC
+    name: SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_UNDERLAY_DMAC
+    description: Action parameter tunnel data underlay destination MAC
     type: sai_mac_t
     attr_value_field: mac
     default: vendor
@@ -413,8 +400,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_UNDERLAY1_DASH_ENCAPSULATION
-    description: Action parameter underlay1 DASH encapsulation
+    name: SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_DASH_ENCAPSULATION
+    description: Action parameter tunnel data DASH encapsulation
     type: sai_dash_encapsulation_t
     attr_value_field: s32
     default: SAI_DASH_ENCAPSULATION_INVALID
@@ -426,8 +413,21 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_DST_MAC
-    description: Action parameter dst MAC
+    name: SAI_FLOW_ENTRY_ATTR_OVERLAY_DATA_IS_IPV6
+    description: Action parameter overlay data is ipv6
+    type: bool
+    attr_value_field: booldata
+    default: 'false'
+    isresourcetype: false
+    flags: CREATE_AND_SET
+    object_name: null
+    allow_null: false
+    valid_only: null
+    is_vlan: false
+    deprecated: false
+  - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
+    name: SAI_FLOW_ENTRY_ATTR_OVERLAY_DATA_DST_MAC
+    description: Action parameter overlay data dst MAC
     type: sai_mac_t
     attr_value_field: mac
     default: vendor
@@ -439,8 +439,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_SIP
-    description: Action parameter sip
+    name: SAI_FLOW_ENTRY_ATTR_OVERLAY_DATA_SIP
+    description: Action parameter overlay data sip
     type: sai_ip_address_t
     attr_value_field: ipaddr
     default: 0.0.0.0
@@ -452,8 +452,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_DIP
-    description: Action parameter dip
+    name: SAI_FLOW_ENTRY_ATTR_OVERLAY_DATA_DIP
+    description: Action parameter overlay data dip
     type: sai_ip_address_t
     attr_value_field: ipaddr
     default: 0.0.0.0
@@ -465,8 +465,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_SIP_MASK
-    description: Action parameter sip mask
+    name: SAI_FLOW_ENTRY_ATTR_OVERLAY_DATA_SIP_MASK
+    description: Action parameter overlay data sip mask
     type: sai_ip_address_t
     attr_value_field: ipaddr
     default: 0.0.0.0
@@ -478,8 +478,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_ATTR_DIP_MASK
-    description: Action parameter dip mask
+    name: SAI_FLOW_ENTRY_ATTR_OVERLAY_DATA_DIP_MASK
+    description: Action parameter overlay data dip mask
     type: sai_ip_address_t
     attr_value_field: ipaddr
     default: 0.0.0.0
@@ -539,33 +539,33 @@ sai_apis:
           name: SAI_FLOW_ENTRY_ACTION_SET_FLOW_ENTRY_ATTR
           id: 26241621
           attr_param_id:
-            SAI_FLOW_ENTRY_ATTR_VERSION: 1
-            SAI_FLOW_ENTRY_ATTR_DASH_DIRECTION: 2
-            SAI_FLOW_ENTRY_ATTR_DASH_FLOW_ACTION: 3
-            SAI_FLOW_ENTRY_ATTR_METER_CLASS: 4
-            SAI_FLOW_ENTRY_ATTR_IS_UNIDIRECTIONAL_FLOW: 5
-            SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_ENI_MAC: 6
-            SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_VNET_ID: 7
-            SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_IP_PROTO: 8
-            SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_SRC_IP: 9
-            SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_DST_IP: 10
-            SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_SRC_PORT: 11
-            SAI_FLOW_ENTRY_ATTR_REVERSE_FLOW_DST_PORT: 12
-            SAI_FLOW_ENTRY_ATTR_UNDERLAY0_VNET_ID: 13
-            SAI_FLOW_ENTRY_ATTR_UNDERLAY0_SIP: 14
-            SAI_FLOW_ENTRY_ATTR_UNDERLAY0_DIP: 15
-            SAI_FLOW_ENTRY_ATTR_UNDERLAY0_DASH_ENCAPSULATION: 16
-            SAI_FLOW_ENTRY_ATTR_UNDERLAY1_VNET_ID: 17
-            SAI_FLOW_ENTRY_ATTR_UNDERLAY1_SIP: 18
-            SAI_FLOW_ENTRY_ATTR_UNDERLAY1_DIP: 19
-            SAI_FLOW_ENTRY_ATTR_UNDERLAY1_SMAC: 20
-            SAI_FLOW_ENTRY_ATTR_UNDERLAY1_DMAC: 21
-            SAI_FLOW_ENTRY_ATTR_UNDERLAY1_DASH_ENCAPSULATION: 22
-            SAI_FLOW_ENTRY_ATTR_DST_MAC: 23
-            SAI_FLOW_ENTRY_ATTR_SIP: 24
-            SAI_FLOW_ENTRY_ATTR_DIP: 25
-            SAI_FLOW_ENTRY_ATTR_SIP_MASK: 26
-            SAI_FLOW_ENTRY_ATTR_DIP_MASK: 27
+            SAI_FLOW_ENTRY_ATTR_IS_UNIDIRECTIONAL: 1
+            SAI_FLOW_ENTRY_ATTR_STATE: 2
+            SAI_FLOW_ENTRY_ATTR_VERSION: 3
+            SAI_FLOW_ENTRY_ATTR_DASH_DIRECTION: 4
+            SAI_FLOW_ENTRY_ATTR_TUNNEL_ID: 5
+            SAI_FLOW_ENTRY_ATTR_ROUTING_ACTIONS: 6
+            SAI_FLOW_ENTRY_ATTR_METER_CLASS: 7
+            SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_VNI: 8
+            SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_DEST_VNET_VNI: 9
+            SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_UNDERLAY_SIP: 10
+            SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_UNDERLAY_DIP: 11
+            SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_UNDERLAY_SMAC: 12
+            SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_UNDERLAY_DMAC: 13
+            SAI_FLOW_ENTRY_ATTR_ENCAP_DATA_DASH_ENCAPSULATION: 14
+            SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_VNI: 15
+            SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_DEST_VNET_VNI: 16
+            SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_UNDERLAY_SIP: 17
+            SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_UNDERLAY_DIP: 18
+            SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_UNDERLAY_SMAC: 19
+            SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_UNDERLAY_DMAC: 20
+            SAI_FLOW_ENTRY_ATTR_TUNNEL_DATA_DASH_ENCAPSULATION: 21
+            SAI_FLOW_ENTRY_ATTR_OVERLAY_DATA_IS_IPV6: 22
+            SAI_FLOW_ENTRY_ATTR_OVERLAY_DATA_DST_MAC: 23
+            SAI_FLOW_ENTRY_ATTR_OVERLAY_DATA_SIP: 24
+            SAI_FLOW_ENTRY_ATTR_OVERLAY_DATA_DIP: 25
+            SAI_FLOW_ENTRY_ATTR_OVERLAY_DATA_SIP_MASK: 26
+            SAI_FLOW_ENTRY_ATTR_OVERLAY_DATA_DIP_MASK: 27
             SAI_FLOW_ENTRY_ATTR_VENDOR_METADATA: 28
             SAI_FLOW_ENTRY_ATTR_FLOW_DATA_PB: 29
 - !!python/object:utils.sai_spec.sai_api.SaiApi
@@ -684,8 +684,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_BULK_GET_SESSION_ATTR_BULK_GET_SESSION_SERVER_IP
-    description: Action parameter bulk get session server IP
+    name: SAI_FLOW_ENTRY_BULK_GET_SESSION_ATTR_BULK_GET_SESSION_IP
+    description: Action parameter bulk get session IP
     type: sai_ip_address_t
     attr_value_field: ipaddr
     default: 0.0.0.0
@@ -697,8 +697,8 @@ sai_apis:
     is_vlan: false
     deprecated: false
   - !!python/object:utils.sai_spec.sai_attribute.SaiAttribute
-    name: SAI_FLOW_ENTRY_BULK_GET_SESSION_ATTR_BULK_GET_SESSION_SERVER_PORT
-    description: Action parameter bulk get session server port
+    name: SAI_FLOW_ENTRY_BULK_GET_SESSION_ATTR_BULK_GET_SESSION_PORT
+    description: Action parameter bulk get session port
     type: sai_uint16_t
     attr_value_field: u16
     default: '0'
diff --git a/dash-pipeline/SAI/specs/sai_spec.yaml b/dash-pipeline/SAI/specs/sai_spec.yaml
index ed86286e3..33800030b 100644
--- a/dash-pipeline/SAI/specs/sai_spec.yaml
+++ b/dash-pipeline/SAI/specs/sai_spec.yaml
@@ -377,6 +377,30 @@ enums:
     name: SWITCHING_TO_STANDALONE
     description: ''
     value: '12'
+- !!python/object:utils.sai_spec.sai_enum.SaiEnum
+  name: sai_dash_flow_state_t
+  description: ''
+  members:
+  - !!python/object:utils.sai_spec.sai_enum_member.SaiEnumMember
+    name: FLOW_MISS
+    description: ''
+    value: '0'
+  - !!python/object:utils.sai_spec.sai_enum_member.SaiEnumMember
+    name: FLOW_CREATED
+    description: ''
+    value: '1'
+  - !!python/object:utils.sai_spec.sai_enum_member.SaiEnumMember
+    name: FLOW_SYNCED
+    description: ''
+    value: '2'
+  - !!python/object:utils.sai_spec.sai_enum_member.SaiEnumMember
+    name: FLOW_PENDING_DELETE
+    description: ''
+    value: '3'
+  - !!python/object:utils.sai_spec.sai_enum_member.SaiEnumMember
+    name: FLOW_PENDING_RESIMULATION
+    description: ''
+    value: '4'
 - !!python/object:utils.sai_spec.sai_enum.SaiEnum
   name: sai_dash_eni_mac_override_type_t
   description: ''
diff --git a/dash-pipeline/SAI/utils/sai_spec/sai_spec_utils.py b/dash-pipeline/SAI/utils/sai_spec/sai_spec_utils.py
index daf0d465b..791a4896e 100644
--- a/dash-pipeline/SAI/utils/sai_spec/sai_spec_utils.py
+++ b/dash-pipeline/SAI/utils/sai_spec/sai_spec_utils.py
@@ -24,6 +24,7 @@ def load_word_fixers() -> None:
         "pb": "protocol buffer",
         "proto": "protocol",
         "smac": "source MAC",
+        "dest": "destination",
         "dpu": "DPU"
     }
 
diff --git a/dash-pipeline/bmv2/dash_headers.p4 b/dash-pipeline/bmv2/dash_headers.p4
index 60fc1d0f5..b2238e57e 100644
--- a/dash-pipeline/bmv2/dash_headers.p4
+++ b/dash-pipeline/bmv2/dash_headers.p4
@@ -6,6 +6,65 @@ typedef bit<32>  IPv4Address;
 typedef bit<128> IPv6Address;
 typedef bit<128> IPv4ORv6Address;
 
+enum bit<16> dash_direction_t {
+    INVALID = 0,
+    OUTBOUND = 1,
+    INBOUND = 2
+};
+
+typedef bit<32> dash_meter_class_t;
+
+enum bit<8> dash_packet_source_t {
+    EXTERNAL = 0,           // Packets from external sources.
+    PIPELINE = 1,           // Packets from P4 pipeline.
+    DPAPP = 2,              // Packets from data plane app.
+    PEER = 3                // Packets from the paired DPU.
+};
+
+enum bit<4> dash_packet_type_t {
+    REGULAR = 0,            // Regular packets from external sources.
+    FLOW_SYNC_REQ = 1,      // Flow sync request packet.
+    FLOW_SYNC_ACK = 2,      // Flow sync ack packet.
+    DP_PROBE_REQ = 3,       // Data plane probe packet.
+    DP_PROBE_ACK = 4        // Data plane probe ack packet.
+};
+
+// Packet subtype for one kind of packet type
+enum bit<4> dash_packet_subtype_t {
+    NONE = 0,        // no op
+    FLOW_CREATE = 1, // New flow creation.
+    FLOW_UPDATE = 2, // Flow resimulation or any other reason causing existing flow to be updated.
+    FLOW_DELETE = 3  // Flow deletion.
+};
+
+enum bit<16> dash_encapsulation_t {
+    INVALID = 0,
+    VXLAN = 1,
+    NVGRE = 2
+}
+
+header encap_data_t {
+    bit<24> vni;
+    bit<24> dest_vnet_vni;
+    IPv4Address underlay_sip;
+    IPv4Address underlay_dip;
+    EthernetAddress underlay_smac;
+    EthernetAddress underlay_dmac;
+    dash_encapsulation_t dash_encapsulation;
+}
+const bit<16> ENCAP_DATA_HDR_SIZE=(24+24+32+32+48+48+16)/8;
+
+header overlay_rewrite_data_t {
+    EthernetAddress dmac;
+    IPv4ORv6Address sip;
+    IPv4ORv6Address dip;
+    IPv6Address sip_mask;
+    IPv6Address dip_mask;
+    bit<7> reserved;
+    bool is_ipv6;
+}
+const bit<16> OVERLAY_REWRITE_DATA_HDR_SIZE=(48+128+128+128+128+8)/8;
+
 header ethernet_t {
     EthernetAddress dst_addr;
     EthernetAddress src_addr;
@@ -93,7 +152,103 @@ header ipv6_t {
 
 const bit<16> IPV6_HDR_SIZE=320/8;
 
+
+#ifdef DPAPP_CONNTRACK
+enum bit<16> dash_flow_enabled_key_t {
+    ENI_ADDR = (1 << 0),
+    VNI = (1 << 1),
+    PROTOCOL = (1 << 2),
+    SRC_IP = (1 << 3),
+    DST_IP = (1 << 4),
+    SRC_PORT = (1 << 5),
+    DST_PORT = (1 << 6)
+}
+
+enum bit<16> dash_flow_entry_bulk_get_session_mode_t {
+    SAI_DASH_FLOW_ENTRY_BULK_GET_SESSION_MODE_GRPC = 0,
+    SAI_DASH_FLOW_ENTRY_BULK_GET_SESSION_MODE_VENDOR = 1,
+    SAI_DASH_FLOW_ENTRY_BULK_GET_SESSION_MODE_EVENT = 2,
+    SAI_DASH_FLOW_ENTRY_BULK_GET_SESSION_MODE_EVENT_WITHOUT_FLOW_STATE = 3
+}
+
+enum bit<16> dash_flow_entry_bulk_get_session_filter_key_t {
+    INVAILD = 0,
+    FLOW_TABLE_ID = 1,
+    ENI_ADDR = 2,
+    IP_PROTOCOL = 3,
+    SRC_IP_ADDR = 4,
+    DST_IP_ADDR = 5,
+    SRC_L4_PORT = 6,
+    DST_L4_PORT = 7,
+    KEY_VERSION = 8
+}
+
+enum bit<8> dash_flow_entry_bulk_get_session_op_key_t {
+    FILTER_OP_INVALID = 0,
+    FILTER_OP_EQUAL_TO = 1,
+    FILTER_OP_GREATER_THAN = 2,
+    FILTER_OP_GREATER_THAN_OR_EQUAL_TO = 3,
+    FILTER_OP_LESS_THAN = 4,
+    FILTER_OP_LESS_THAN_OR_EQUAL_TO = 5
+}
+
+struct flow_table_data_t {
+    bit<16> id;
+    bit<32> max_flow_count;
+    bit<16> flow_enabled_key;
+    bit<32> flow_ttl_in_milliseconds;
+}
+
+header flow_key_t {
+    EthernetAddress eni_mac;
+    bit<16> vnet_id;
+    IPv4ORv6Address src_ip;
+    IPv4ORv6Address dst_ip;
+    bit<16> src_port;
+    bit<16> dst_port;
+    bit<8> ip_proto;
+    bit<7> reserved;
+    bit<1> is_ip_v6;
+}
+const bit<16> FLOW_KEY_HDR_SIZE=(48+16+128+128+16+16+8+8)/8;
+
+header flow_data_t {
+    bit<7> reserved;
+    bit<1> is_unidirectional;
+    bit<32> version;
+    dash_direction_t direction;
+    bit<16> tunnel_id;
+    bit<32> routing_actions;
+    dash_meter_class_t meter_class;
+}
+const bit<16> FLOW_DATA_HDR_SIZE=(8+32+16+16+32+32)/8;
+
+// dash packet metadata
+header dash_packet_meta_t {
+    dash_packet_source_t packet_source;
+    dash_packet_type_t packet_type;
+    dash_packet_subtype_t packet_subtype;
+    bit<16>     length;
+}
+const bit<16> PACKET_META_HDR_SIZE=(8+4+4+16)/8;
+
+#define DASH_ETHTYPE 0x876d
+#define DPAPP_MAC 0x02fe23f0e413
+
+#endif // DPAPP_CONNTRACK
+
 struct headers_t {
+#ifdef DPAPP_CONNTRACK
+    /* packet metadata headers */
+    ethernet_t   dp_ethernet;
+    dash_packet_meta_t    packet_meta;
+    flow_key_t   flow_key;
+    flow_data_t  flow_data; // flow common data
+    overlay_rewrite_data_t flow_overlay_data;
+    encap_data_t flow_encap_data;
+    encap_data_t flow_tunnel_data;
+#endif // DPAPP_CONNTRACK
+
     /* Underlay 1 headers */
     ethernet_t    u1_ethernet;
     ipv4_t        u1_ipv4;
@@ -122,10 +277,4 @@ struct headers_t {
     tcp_t         customer_tcp;
 }
 
-enum bit<16> dash_encapsulation_t {
-    INVALID = 0,
-    VXLAN = 1,
-    NVGRE = 2
-}
-
 #endif /* _SIRIUS_HEADERS_P4_ */
diff --git a/dash-pipeline/bmv2/dash_metadata.p4 b/dash-pipeline/bmv2/dash_metadata.p4
index 48f01561c..fbe329cb0 100644
--- a/dash-pipeline/bmv2/dash_metadata.p4
+++ b/dash-pipeline/bmv2/dash_metadata.p4
@@ -14,26 +14,6 @@ enum bit<32> dash_routing_actions_t {
     NAT_PORT = (1 << 4)
 };
 
-enum bit<16> dash_direction_t {
-    INVALID = 0,
-    OUTBOUND = 1,
-    INBOUND = 2
-};
-
-enum bit<8> dash_packet_source_t {
-    EXTERNAL = 0,           // Packets from external sources.
-    DPAPP = 1,              // Packets from data plane app.
-    PEER = 2                // Packets from the paired DPU.
-};
-
-enum bit<8> dash_packet_type_t {
-    REGULAR = 0,            // Regular packets from external sources.
-    FLOW_SYNC_REQ = 1,      // Flow sync request packet.
-    FLOW_SYNC_ACK = 2,      // Flow sync ack packet.
-    DP_PROBE_REQ = 3,       // Data plane probe packet.
-    DP_PROBE_ACK = 4        // Data plane probe ack packet.
-};
-
 // Pipeline stages:
 enum bit<16> dash_pipeline_stage_t {
     INVALID = 0,
@@ -51,74 +31,6 @@ enum bit<16> dash_pipeline_stage_t {
     ROUTING_ACTION_APPLY = 300
 };
 
-enum bit<16> dash_flow_enabled_key_t {
-    ENI_ADDR = (1 << 0),
-    VNI = (1 << 1),
-    PROTOCOL = (1 << 2),
-    SRC_IP = (1 << 3),
-    DST_IP = (1 << 4),
-    SRC_PORT = (1 << 5),
-    DST_PORT = (1 << 6)
-}
-
-struct flow_table_data_t {
-    bit<16> id;
-    bit<32> max_flow_count;
-    dash_flow_enabled_key_t flow_enabled_key;
-    bit<32> flow_ttl_in_milliseconds;
-}
-
-enum bit<32> dash_flow_action_t {
-    NONE = 0
-}
-
-struct flow_key_t {
-    EthernetAddress eni_mac;
-    bit<8> ip_proto;
-    bit<16> vnet_id;
-    IPv4ORv6Address src_ip;
-    IPv4ORv6Address dst_ip;
-    bit<16> src_port;
-    bit<16> dst_port;
-    bool is_ipv6;
-}
-
-struct flow_data_t {
-    bit<32> version;
-    dash_direction_t dash_direction;
-    dash_flow_action_t actions;
-}
-
-enum bit<16> dash_flow_entry_bulk_get_session_mode_t {
-    SAI_DASH_FLOW_ENTRY_BULK_GET_SESSION_MODE_GRPC = 0,
-    SAI_DASH_FLOW_ENTRY_BULK_GET_SESSION_MODE_VENDOR = 1,
-    SAI_DASH_FLOW_ENTRY_BULK_GET_SESSION_MODE_EVENT = 2,
-    SAI_DASH_FLOW_ENTRY_BULK_GET_SESSION_MODE_EVENT_WITHOUT_FLOW_STATE = 3
-}
-
-enum bit<16> dash_flow_entry_bulk_get_session_filter_key_t
-{
-    INVAILD = 0,
-    FLOW_TABLE_ID = 1,
-    ENI_ADDR = 2,
-    IP_PROTOCOL = 3,
-    SRC_IP_ADDR = 4,
-    DST_IP_ADDR = 5,
-    SRC_L4_PORT = 6,
-    DST_L4_PORT = 7,
-    KEY_VERSION = 8
-}
-
-enum bit<8> dash_flow_entry_bulk_get_session_op_key_t
-{
-    FILTER_OP_INVALID = 0,
-    FILTER_OP_EQUAL_TO = 1,
-    FILTER_OP_GREATER_THAN = 2,
-    FILTER_OP_GREATER_THAN_OR_EQUAL_TO = 3,
-    FILTER_OP_LESS_THAN = 4,
-    FILTER_OP_LESS_THAN_OR_EQUAL_TO = 5
-}
-
 enum bit<8> dash_eni_mac_override_type_t {
     NONE = 0,
     SRC_MAC = 1,
@@ -133,14 +45,6 @@ enum bit<8> dash_eni_mac_type_t {
 struct conntrack_data_t {
     bool allow_in;
     bool allow_out;
-    flow_table_data_t flow_table;
-    EthernetAddress eni_mac;
-    flow_data_t flow_data;
-    flow_key_t flow_key;
-    flow_key_t reverse_flow_key;
-    bit<1> is_unidirectional_flow;
-    bit<16> bulk_get_session_id;
-    bit<16> bulk_get_session_filter_id;
 }
 
 enum bit<16> dash_tunnel_dscp_mode_t {
@@ -173,24 +77,6 @@ struct meter_context_t {
     IPv4ORv6Address meter_policy_lookup_ip;
 }
 
-struct encap_data_t {
-    bit<24> vni;
-    IPv4Address underlay_sip;
-    IPv4Address underlay_dip;
-    dash_encapsulation_t dash_encapsulation;
-    EthernetAddress underlay_smac;
-    EthernetAddress underlay_dmac;
-}
-
-struct overlay_rewrite_data_t {
-    bool is_ipv6;
-    EthernetAddress dmac;
-    IPv4ORv6Address sip;
-    IPv4ORv6Address dip;
-    IPv6Address sip_mask;
-    IPv6Address dip_mask;
-}
-
 // HA roles
 enum bit<8> dash_ha_role_t {
     DEAD = 0,
@@ -229,8 +115,8 @@ enum bit<8> dash_ha_state_t {
     SWITCHING_TO_STANDALONE = 12
 };
 
-// Flow sync state
-enum bit<8> dash_ha_flow_sync_state_t {
+// Flow state
+enum bit<8> dash_flow_state_t {
     FLOW_MISS = 0,                  // Flow not created yet
     FLOW_CREATED = 1,               // Flow is created but not synched or waiting for ack
     FLOW_SYNCED = 2,                // Flow has been synched to its peer
@@ -238,13 +124,6 @@ enum bit<8> dash_ha_flow_sync_state_t {
     FLOW_PENDING_RESIMULATION = 4   // Flow is marked as pending resimulation
 };
 
-// HA flow sync operations
-enum bit<8> dash_ha_flow_sync_op_t {
-    FLOW_CREATE = 0, // New flow creation.
-    FLOW_UPDATE = 1, // Flow resimulation or any other reason causing existing flow to be updated.
-    FLOW_DELETE = 2  // Flow deletion.
-};
-
 struct ha_data_t {
     // HA scope settings
     bit<16> ha_scope_id;
@@ -261,7 +140,7 @@ struct ha_data_t {
     bit<16> dp_channel_src_port_max;
 
     // HA packet/flow state
-    dash_ha_flow_sync_state_t flow_sync_state;
+    dash_flow_state_t flow_sync_state;
 }
 
 struct metadata_t {
@@ -303,13 +182,18 @@ struct metadata_t {
 
     // Flow data
     conntrack_data_t conntrack_data;
+#ifdef DPAPP_CONNTRACK
+    flow_table_data_t flow_table;
+    dash_flow_state_t flow_state;
+    bit<16> bulk_get_session_id;
+    bit<16> bulk_get_session_filter_id;
+#endif // DPAPP_CONNTRACK
 
     // Stage transition control
     dash_pipeline_stage_t target_stage;
 
     // Actions
     bit<32> routing_actions;
-    bit<32> flow_actions;
 
     // Action data
     bool dropped;
diff --git a/dash-pipeline/bmv2/dash_parser.p4 b/dash-pipeline/bmv2/dash_parser.p4
index d6eed01fc..3ff5652bf 100644
--- a/dash-pipeline/bmv2/dash_parser.p4
+++ b/dash-pipeline/bmv2/dash_parser.p4
@@ -29,14 +29,35 @@ parser dash_parser(
     )
 {
     state start {
+#ifdef DPAPP_CONNTRACK
+        // By default, packet is REGULAR from EXTERNAL
+        hd.packet_meta.setValid();
+        hd.packet_meta.packet_source = dash_packet_source_t.EXTERNAL;
+        hd.packet_meta.packet_type = dash_packet_type_t.REGULAR;
+        hd.packet_meta.packet_subtype = dash_packet_subtype_t.NONE;
+        hd.packet_meta.length = PACKET_META_HDR_SIZE;
+#endif // DPAPP_CONNTRACK
+
         packet.extract(hd.u0_ethernet);
         transition select(hd.u0_ethernet.ether_type) {
             IPV4_ETHTYPE:  parse_u0_ipv4;
             IPV6_ETHTYPE:  parse_u0_ipv6;
+#ifdef DPAPP_CONNTRACK
+            DASH_ETHTYPE:  parse_dash;
+#endif // DPAPP_CONNTRACK
             default: accept;
         }
     }
 
+#ifdef DPAPP_CONNTRACK
+    state parse_dash {
+        packet.extract(hd.packet_meta);
+        // TODO extract flow_key, etc by checking packet type/subtype
+        packet.extract(hd.flow_key);
+        transition parse_customer_ethernet;
+    }
+#endif // DPAPP_CONNTRACK
+
     state parse_u0_ipv4 {
         packet.extract(hd.u0_ipv4);
         verify(hd.u0_ipv4.version == 4w4, error.IPv4IncorrectVersion);
@@ -138,7 +159,17 @@ control dash_deparser(
     )
 {
     apply {
-	packet.emit(hdr.u0_ethernet);
+#ifdef DPAPP_CONNTRACK
+        packet.emit(hdr.dp_ethernet);
+        packet.emit(hdr.packet_meta);
+        packet.emit(hdr.flow_key);
+        packet.emit(hdr.flow_data);
+        packet.emit(hdr.flow_overlay_data);
+        packet.emit(hdr.flow_encap_data);
+        packet.emit(hdr.flow_tunnel_data);
+#endif // DPAPP_CONNTRACK
+
+        packet.emit(hdr.u0_ethernet);
         packet.emit(hdr.u0_ipv4);
         packet.emit(hdr.u0_ipv4options);
         packet.emit(hdr.u0_ipv6);
diff --git a/dash-pipeline/bmv2/dash_pipeline.p4 b/dash-pipeline/bmv2/dash_pipeline.p4
index 0113ec195..19f110514 100644
--- a/dash-pipeline/bmv2/dash_pipeline.p4
+++ b/dash-pipeline/bmv2/dash_pipeline.p4
@@ -8,8 +8,11 @@
 #include "dash_tunnel.p4"
 #include "dash_outbound.p4"
 #include "dash_inbound.p4"
-#include "dash_conntrack.p4"
+#ifdef DPAPP_CONNTRACK
 #include "stages/conntrack_lookup.p4"
+#else
+#include "dash_conntrack.p4"
+#endif
 #include "stages/direction_lookup.p4"
 #include "stages/eni_lookup.p4"
 #include "stages/ha.p4"
@@ -250,6 +253,9 @@ control dash_ingress(
     }
 
     apply {
+        meta.encap_data.setValid();
+        meta.tunnel_data.setValid();
+        meta.overlay_data.setValid();
 
 #ifdef TARGET_DPDK_PNA
 #ifdef DPDK_PNA_SEND_TO_PORT_FIX_MERGED
@@ -265,6 +271,11 @@ control dash_ingress(
 #endif  // DPDK_PNA_SEND_TO_PORT_FIX_MERGED
 #endif // TARGET_DPDK_PNA
 
+#ifdef DPAPP_CONNTRACK
+        // If packet is from DPAPP, just do conntrack_lookup
+        if (hdr.packet_meta.packet_source != dash_packet_source_t.DPAPP) {
+#endif // DPAPP_CONNTRACK
+
         if (meta.is_fast_path_icmp_flow_redirection_packet) {
             UPDATE_COUNTER(port_lb_fast_path_icmp_in, 0);
         }
@@ -337,16 +348,24 @@ control dash_ingress(
         if (meta.eni_data.admin_state == 0) {
             deny();
         }
-
-        conntrack_lookup_stage.apply(hdr, meta);
-
+        
         UPDATE_ENI_COUNTER(eni_rx);
         if (meta.is_fast_path_icmp_flow_redirection_packet) {
             UPDATE_ENI_COUNTER(eni_lb_fast_path_icmp_in);
         }
 
+
+#ifdef DPAPP_CONNTRACK
+        } // from dpapp
+        conntrack_lookup_stage.apply(hdr, meta);
+#endif // DPAPP_CONNTRACK
+
         ha_stage.apply(hdr, meta);
 
+#ifdef DPAPP_CONNTRACK
+        if (meta.flow_state == dash_flow_state_t.FLOW_MISS) {
+#endif // DPAPP_CONNTRACK
+
         acl_group.apply();
 
         if (meta.direction == dash_direction_t.OUTBOUND) {
@@ -361,6 +380,15 @@ control dash_ingress(
 
         tunnel_stage.apply(hdr, meta);
 
+#ifdef DPAPP_CONNTRACK
+        if (!meta.dropped) {
+            conntrack_set_flow_data.apply(hdr, meta);
+            standard_metadata.egress_spec = 2; // vpp port
+            return;
+        }
+        }
+#endif // DPAPP_CONNTRACK
+
         routing_action_apply.apply(hdr, meta);
 
         tunnel_stage_encap.apply(hdr, meta);
@@ -396,6 +424,19 @@ control dash_ingress(
                 UPDATE_ENI_COUNTER(eni_inbound_tx);
             }
         }
+
+#ifdef DPAPP_CONNTRACK
+        // Drop dash header in fast path
+        if (meta.flow_state == dash_flow_state_t.FLOW_CREATED) {
+            hdr.dp_ethernet.setInvalid();
+            hdr.packet_meta.setInvalid();
+            hdr.flow_key.setInvalid();
+            hdr.flow_data.setInvalid();
+            hdr.flow_overlay_data.setInvalid();
+            hdr.flow_encap_data.setInvalid();
+            hdr.flow_tunnel_data.setInvalid();
+        }
+#endif // DPAPP_CONNTRACK
     }
 }
 
diff --git a/dash-pipeline/bmv2/stages/conntrack_lookup.p4 b/dash-pipeline/bmv2/stages/conntrack_lookup.p4
index 99e1c011c..4907c73c9 100644
--- a/dash-pipeline/bmv2/stages/conntrack_lookup.p4
+++ b/dash-pipeline/bmv2/stages/conntrack_lookup.p4
@@ -1,8 +1,53 @@
 #ifndef _DASH_STAGE_CONNTRACK_LOOKUP_P4
 #define _DASH_STAGE_CONNTRACK_LOOKUP_P4
 
+#ifdef DPAPP_CONNTRACK
 #include "../dash_metadata.p4"
 
+control conntrack_set_flow_data(inout headers_t hdr, inout metadata_t meta)
+{
+    apply {
+        bit<16> length = 0;
+
+        hdr.flow_data.setValid();
+        hdr.flow_data.is_unidirectional = 0;
+        hdr.flow_data.version = 0;
+        hdr.flow_data.direction = meta.direction;
+        hdr.flow_data.tunnel_id = meta.dash_tunnel_id;
+        hdr.flow_data.routing_actions = meta.routing_actions;
+        hdr.flow_data.meter_class = meta.meter_class;
+        length = length + FLOW_DATA_HDR_SIZE;
+
+        if (meta.routing_actions & dash_routing_actions_t.STATIC_ENCAP != 0) {
+            hdr.flow_encap_data= meta.encap_data;
+            length = length + ENCAP_DATA_HDR_SIZE;
+        }
+
+        if (meta.dash_tunnel_id != 0) {
+            hdr.flow_tunnel_data= meta.tunnel_data;
+            length = length + ENCAP_DATA_HDR_SIZE;
+        }
+
+        if (meta.routing_actions != 0) {
+            hdr.flow_overlay_data= meta.overlay_data;
+            length = length + OVERLAY_REWRITE_DATA_HDR_SIZE;
+        }
+
+        length = length + FLOW_KEY_HDR_SIZE;
+
+        hdr.packet_meta.setValid();
+        hdr.packet_meta.packet_source = dash_packet_source_t.PIPELINE;
+        hdr.packet_meta.packet_type = dash_packet_type_t.REGULAR;
+        hdr.packet_meta.packet_subtype = dash_packet_subtype_t.FLOW_CREATE;
+        hdr.packet_meta.length = length + PACKET_META_HDR_SIZE;
+
+        hdr.dp_ethernet.setValid();
+        hdr.dp_ethernet.dst_addr = DPAPP_MAC;
+        hdr.dp_ethernet.src_addr = meta.encap_data.underlay_smac;
+        hdr.dp_ethernet.ether_type = DASH_ETHTYPE;
+    }
+}
+
 control conntrack_lookup_stage(inout headers_t hdr, inout metadata_t meta) {
     //
     // Flow table:
@@ -12,15 +57,15 @@ control conntrack_lookup_stage(inout headers_t hdr, inout metadata_t meta) {
         @SaiVal[type="sai_dash_flow_enabled_key_t"] dash_flow_enabled_key_t dash_flow_enabled_key,
         bit<32> flow_ttl_in_milliseconds)
     {
-        meta.conntrack_data.flow_table.max_flow_count = max_flow_count;
-        meta.conntrack_data.flow_table.flow_enabled_key = dash_flow_enabled_key;
-        meta.conntrack_data.flow_table.flow_ttl_in_milliseconds = flow_ttl_in_milliseconds;
+        meta.flow_table.max_flow_count = max_flow_count;
+        meta.flow_table.flow_enabled_key = dash_flow_enabled_key;
+        meta.flow_table.flow_ttl_in_milliseconds = flow_ttl_in_milliseconds;
     }
 
     @SaiTable[name = "flow_table", api = "dash_flow", order = 0, isobject="true"]
     table flow_table {
         key = {
-            meta.conntrack_data.flow_table.id : exact;
+            meta.flow_table.id : exact;
         }
 
         actions = {
@@ -33,108 +78,115 @@ control conntrack_lookup_stage(inout headers_t hdr, inout metadata_t meta) {
     //
     action set_flow_entry_attr(
         /* Flow basic metadata */
+        bit<1> is_unidirectional,
+        dash_flow_state_t state,
         bit<32> version,
         @SaiVal[type="sai_dash_direction_t"] dash_direction_t dash_direction,
-        @SaiVal[type="sai_dash_flow_action_t"] dash_flow_action_t dash_flow_action,
+        bit<16> tunnel_id,
+        bit<32> routing_actions,
         bit<32> meter_class,
-        bit<1> is_unidirectional_flow,
-
-        /* Reverse flow key */ 
-        EthernetAddress reverse_flow_eni_mac,
-        bit<16> reverse_flow_vnet_id,
-        bit<8> reverse_flow_ip_proto,
-        IPv4ORv6Address reverse_flow_src_ip,
-        IPv4ORv6Address reverse_flow_dst_ip, 
-        bit<16> reverse_flow_src_port,
-        bit<16> reverse_flow_dst_port,
 
         /* Flow encap related attributes */
-        bit<24> underlay0_vnet_id,
-        IPv4Address underlay0_sip, 
-        IPv4Address underlay0_dip,
-        @SaiVal[type="sai_dash_encapsulation_t"] dash_encapsulation_t underlay0_dash_encapsulation,
-        
-        bit<24> underlay1_vnet_id,
-        IPv4Address underlay1_sip, 
-        IPv4Address underlay1_dip,
-        EthernetAddress underlay1_smac,
-        EthernetAddress underlay1_dmac,
-        @SaiVal[type="sai_dash_encapsulation_t"] dash_encapsulation_t underlay1_dash_encapsulation,
-        
+        bit<24> encap_data_vni,
+        bit<24> encap_data_dest_vnet_vni,
+        IPv4Address encap_data_underlay_sip,
+        IPv4Address encap_data_underlay_dip,
+        EthernetAddress encap_data_underlay_smac,
+        EthernetAddress encap_data_underlay_dmac,
+        @SaiVal[type="sai_dash_encapsulation_t"] dash_encapsulation_t encap_data_dash_encapsulation,
+
+        /* Flow tunnel related attributes */
+        bit<24> tunnel_data_vni,
+        bit<24> tunnel_data_dest_vnet_vni,
+        IPv4Address tunnel_data_underlay_sip,
+        IPv4Address tunnel_data_underlay_dip,
+        EthernetAddress tunnel_data_underlay_smac,
+        EthernetAddress tunnel_data_underlay_dmac,
+        @SaiVal[type="sai_dash_encapsulation_t"] dash_encapsulation_t tunnel_data_dash_encapsulation,
+
         /* Flow overlay rewrite related attributes */
-        EthernetAddress dst_mac,
-        IPv4ORv6Address sip,
-        IPv4ORv6Address dip,
-        IPv6Address sip_mask,
-        IPv6Address dip_mask,
-      
-        /* Extra flow metadata */ 
+        bit<1> overlay_data_is_ipv6,
+        EthernetAddress overlay_data_dst_mac,
+        IPv4ORv6Address overlay_data_sip,
+        IPv4ORv6Address overlay_data_dip,
+        IPv6Address overlay_data_sip_mask,
+        IPv6Address overlay_data_dip_mask,
+
+        /* Extra flow metadata, unused */
         @SaiVal[type="sai_u8_list_t"] bit<16> vendor_metadata,
         @SaiVal[type="sai_u8_list_t"] bit<16> flow_data_pb)
     {
-        /* Set Flow basic metadata */
-        meta.conntrack_data.flow_data.version = version;
-        meta.conntrack_data.flow_data.dash_direction = dash_direction;
-        meta.conntrack_data.flow_data.actions = dash_flow_action;
-        meta.meter_class = meter_class;
-        meta.conntrack_data.is_unidirectional_flow = is_unidirectional_flow;
-        
-        /* Set reverse flow key */
-        meta.conntrack_data.reverse_flow_key.eni_mac = reverse_flow_eni_mac;
-        meta.conntrack_data.reverse_flow_key.vnet_id = reverse_flow_vnet_id;
-        meta.conntrack_data.reverse_flow_key.ip_proto = reverse_flow_ip_proto;
-        meta.conntrack_data.reverse_flow_key.src_ip = reverse_flow_src_ip;
-        meta.conntrack_data.reverse_flow_key.dst_ip = reverse_flow_dst_ip;
-        meta.conntrack_data.reverse_flow_key.src_port = reverse_flow_src_port;
-        meta.conntrack_data.reverse_flow_key.dst_port = reverse_flow_dst_port;
+        meta.flow_state = state;
+        // TODO check FLOW_SYNCED in ha
+        // Do state FSM later
+        if (state != dash_flow_state_t.FLOW_CREATED) {
+            return;
+        }
 
+        /* Set basic metadata */
+        meta.direction = dash_direction;
+        meta.dash_tunnel_id = tunnel_id;
+        meta.routing_actions = routing_actions;
+        meta.meter_class = meter_class;
 
         /* Set encapsulation metadata */
-        meta.encap_data.vni = underlay0_vnet_id;
-        meta.encap_data.underlay_sip = underlay0_sip;
-        meta.encap_data.underlay_dip = underlay0_dip;
-        meta.encap_data.dash_encapsulation = underlay0_dash_encapsulation;
+        meta.encap_data.vni = encap_data_vni;
+        meta.encap_data.dest_vnet_vni = encap_data_dest_vnet_vni;
+        meta.encap_data.underlay_sip = encap_data_underlay_sip;
+        meta.encap_data.underlay_dip = encap_data_underlay_dip;
+        meta.encap_data.underlay_smac = encap_data_underlay_smac;
+        meta.encap_data.underlay_dmac = encap_data_underlay_dmac;
+        meta.encap_data.dash_encapsulation = encap_data_dash_encapsulation;
 
-        meta.tunnel_data.vni = underlay1_vnet_id;
-        meta.tunnel_data.underlay_sip = underlay1_sip;
-        meta.tunnel_data.underlay_dip = underlay1_dip;
-        meta.tunnel_data.dash_encapsulation = underlay1_dash_encapsulation;
-        meta.tunnel_data.underlay_smac = underlay1_smac;
-        meta.tunnel_data.underlay_dmac = underlay1_dmac;
 
+        /* Set tunnel metadata */
+        meta.tunnel_data.vni = tunnel_data_vni;
+        meta.tunnel_data.dest_vnet_vni = tunnel_data_dest_vnet_vni;
+        meta.tunnel_data.underlay_sip = tunnel_data_underlay_sip;
+        meta.tunnel_data.underlay_dip = tunnel_data_underlay_dip;
+        meta.tunnel_data.underlay_smac = tunnel_data_underlay_smac;
+        meta.tunnel_data.underlay_dmac = tunnel_data_underlay_dmac;
+        meta.tunnel_data.dash_encapsulation = tunnel_data_dash_encapsulation;
 
         /* Set overlay rewrite metadata */
-        meta.overlay_data.dmac = dst_mac;
-        meta.overlay_data.sip = sip;
-        meta.overlay_data.dip = dip;
-        meta.overlay_data.sip_mask = sip_mask;
-        meta.overlay_data.dip_mask = dip_mask;
+        meta.overlay_data.is_ipv6 = (bool)overlay_data_is_ipv6;
+        meta.overlay_data.dmac = overlay_data_dst_mac;
+        meta.overlay_data.sip = overlay_data_sip;
+        meta.overlay_data.dip = overlay_data_dip;
+        meta.overlay_data.sip_mask = overlay_data_sip_mask;
+        meta.overlay_data.dip_mask = overlay_data_dip_mask;
+    }
+
+    action flow_miss() {
+        meta.flow_state = dash_flow_state_t.FLOW_MISS;
     }
 
     @SaiTable[name = "flow", api = "dash_flow", order = 1, enable_bulk_get_api = "true", enable_bulk_get_server = "true"]
     table flow_entry {
         key = {
-            meta.conntrack_data.flow_key.eni_mac : exact;
-            meta.conntrack_data.flow_key.vnet_id : exact;
-            meta.conntrack_data.flow_key.ip_proto : exact;
-            meta.conntrack_data.flow_key.src_ip : exact;
-            meta.conntrack_data.flow_key.dst_ip : exact;
-            meta.conntrack_data.flow_key.src_port : exact;
-            meta.conntrack_data.flow_key.dst_port : exact;
-            meta.conntrack_data.flow_key.is_ipv6 : exact @SaiVal[name = "src_ip_is_v6"];
+            hdr.flow_key.eni_mac : exact;
+            hdr.flow_key.vnet_id : exact;
+            hdr.flow_key.src_ip : exact;
+            hdr.flow_key.dst_ip : exact;
+            hdr.flow_key.src_port : exact;
+            hdr.flow_key.dst_port : exact;
+            hdr.flow_key.ip_proto : exact;
+            hdr.flow_key.is_ip_v6 : exact @SaiVal[name = "src_ip_is_v6"];
         }
 
         actions = {
             set_flow_entry_attr;
+            @defaultonly flow_miss;
         }
+        const default_action = flow_miss;
     }
-    
+
     //
     // Flow bulk get session filter:
     // For API generation only and has no effect on the dataplane
     //
     action set_flow_entry_bulk_get_session_filter_attr(
-        @SaiVal[type="sai_dash_flow_entry_bulk_get_session_filter_key_t"] dash_flow_entry_bulk_get_session_filter_key_t dash_flow_entry_bulk_get_session_filter_key, 
+        @SaiVal[type="sai_dash_flow_entry_bulk_get_session_filter_key_t"] dash_flow_entry_bulk_get_session_filter_key_t dash_flow_entry_bulk_get_session_filter_key,
         @SaiVal[type="sai_dash_flow_entry_bulk_get_session_op_key_t"] dash_flow_entry_bulk_get_session_op_key_t  dash_flow_entry_bulk_get_session_op_key,
         bit<64> int_value,
         IPv4ORv6Address ip_value,
@@ -148,14 +200,14 @@ control conntrack_lookup_stage(inout headers_t hdr, inout metadata_t meta) {
     //
     action set_flow_entry_bulk_get_session_attr(
         /* Mode and limitation */
-        @SaiVal[type="sai_dash_flow_entry_bulk_get_session_mode_t"] dash_flow_entry_bulk_get_session_mode_t dash_flow_entry_bulk_get_session_mode, 
+        @SaiVal[type="sai_dash_flow_entry_bulk_get_session_mode_t"] dash_flow_entry_bulk_get_session_mode_t dash_flow_entry_bulk_get_session_mode,
         bit<32> bulk_get_entry_limitation,
-        
+
         /* GRPC Session server IP and port */
-        IPv4ORv6Address bulk_get_session_server_ip,
-        bit<16> bulk_get_session_server_port,
+        IPv4ORv6Address bulk_get_session_ip,
+        bit<16> bulk_get_session_port,
 
-        /* Session filters */ 
+        /* Session filters */
         @SaiVal[type="sai_object_id_t"] bit<16> first_flow_entry_bulk_get_session_filter_id,
         @SaiVal[type="sai_object_id_t"] bit<16> second_flow_entry_bulk_get_session_filter_id,
         @SaiVal[type="sai_object_id_t"] bit<16> third_flow_entry_bulk_get_session_filter_id,
@@ -164,12 +216,11 @@ control conntrack_lookup_stage(inout headers_t hdr, inout metadata_t meta) {
     {
     }
 
-     
 
     @SaiTable[name = "flow_entry_bulk_get_session_filter", api = "dash_flow", order = 2, isobject="true"]
     table flow_entry_bulk_get_session_filter {
         key = {
-            meta.conntrack_data.bulk_get_session_filter_id: exact @SaiVal[name = "bulk_get_session_filter_id", type="sai_object_id_t"];
+            meta.bulk_get_session_filter_id: exact @SaiVal[name = "bulk_get_session_filter_id", type="sai_object_id_t"];
         }
 
         actions = {
@@ -180,7 +231,7 @@ control conntrack_lookup_stage(inout headers_t hdr, inout metadata_t meta) {
    @SaiTable[name = "flow_entry_bulk_get_session", api = "dash_flow", order = 3, isobject="true"]
     table flow_entry_bulk_get_session {
         key = {
-            meta.conntrack_data.bulk_get_session_id: exact @SaiVal[name = "bulk_get_session_id", type="sai_object_id_t"];
+            meta.bulk_get_session_id: exact @SaiVal[name = "bulk_get_session_id", type="sai_object_id_t"];
         }
 
         actions = {
@@ -188,33 +239,48 @@ control conntrack_lookup_stage(inout headers_t hdr, inout metadata_t meta) {
         }
     }
 
-    apply {
-        flow_table.apply();
+    action set_flow_key()
+    {
+        hdr.flow_key.setValid();
+        hdr.flow_key.is_ip_v6 = meta.is_overlay_ip_v6;
+        // TODO remove hardcode flow_enabled_key later
+        meta.flow_table.flow_enabled_key = dash_flow_enabled_key_t.ENI_ADDR |
+                                           dash_flow_enabled_key_t.VNI |
+                                           dash_flow_enabled_key_t.PROTOCOL |
+                                           dash_flow_enabled_key_t.SRC_IP |
+                                           dash_flow_enabled_key_t.DST_IP |
+                                           dash_flow_enabled_key_t.SRC_PORT |
+                                           dash_flow_enabled_key_t.DST_PORT;
 
-        if (meta.conntrack_data.flow_table.flow_enabled_key & dash_flow_enabled_key_t.ENI_ADDR != 0) {
-            meta.conntrack_data.flow_key.eni_mac = meta.eni_addr;
+        if (meta.flow_table.flow_enabled_key & dash_flow_enabled_key_t.ENI_ADDR != 0) {
+            hdr.flow_key.eni_mac = meta.eni_addr;
         }
-
-        if (meta.conntrack_data.flow_table.flow_enabled_key & dash_flow_enabled_key_t.VNI != 0) {
-             meta.conntrack_data.flow_key.vnet_id = meta.vnet_id;
-        } 
-        
-        if (meta.conntrack_data.flow_table.flow_enabled_key & dash_flow_enabled_key_t.PROTOCOL != 0) {
-            meta.conntrack_data.flow_key.ip_proto = meta.ip_protocol;
+        if (meta.flow_table.flow_enabled_key & dash_flow_enabled_key_t.VNI != 0) {
+            hdr.flow_key.vnet_id = meta.vnet_id;
+        }
+        if (meta.flow_table.flow_enabled_key & dash_flow_enabled_key_t.PROTOCOL != 0) {
+            hdr.flow_key.ip_proto = meta.ip_protocol;
+        }
+        if (meta.flow_table.flow_enabled_key & dash_flow_enabled_key_t.SRC_IP != 0) {
+            hdr.flow_key.src_ip = meta.src_ip_addr;
         }
-        if (meta.conntrack_data.flow_table.flow_enabled_key & dash_flow_enabled_key_t.SRC_IP != 0) {
-            meta.conntrack_data.flow_key.src_ip = meta.src_ip_addr;
+        if (meta.flow_table.flow_enabled_key & dash_flow_enabled_key_t.DST_IP != 0) {
+            hdr.flow_key.dst_ip = meta.dst_ip_addr;
         }
-        if (meta.conntrack_data.flow_table.flow_enabled_key & dash_flow_enabled_key_t.DST_IP != 0) {
-            meta.conntrack_data.flow_key.dst_ip = meta.dst_ip_addr;
+
+        if (meta.flow_table.flow_enabled_key & dash_flow_enabled_key_t.SRC_PORT != 0) {
+            hdr.flow_key.src_port = meta.src_l4_port;
         }
 
-        if (meta.conntrack_data.flow_table.flow_enabled_key & dash_flow_enabled_key_t.SRC_PORT != 0) {
-            meta.conntrack_data.flow_key.src_port = meta.src_l4_port;
+        if (meta.flow_table.flow_enabled_key & dash_flow_enabled_key_t.DST_PORT != 0) {
+            hdr.flow_key.dst_port = meta.dst_l4_port;
         }
+    }
 
-        if (meta.conntrack_data.flow_table.flow_enabled_key & dash_flow_enabled_key_t.DST_PORT != 0) {
-            meta.conntrack_data.flow_key.dst_port = meta.dst_l4_port;
+    apply {
+        if (!hdr.flow_key.isValid()) {
+            flow_table.apply();
+            set_flow_key();
         }
 
         flow_entry.apply();
@@ -222,5 +288,5 @@ control conntrack_lookup_stage(inout headers_t hdr, inout metadata_t meta) {
         flow_entry_bulk_get_session.apply();
     }
 }
-
+#endif // DPAPP_CONNTRACK
 #endif /* _DASH_STAGE_CONNTRACK_LOOKUP_P4 */
diff --git a/dash-pipeline/bmv2/underlay.p4 b/dash-pipeline/bmv2/underlay.p4
index a55e71dcd..ba8811d7a 100644
--- a/dash-pipeline/bmv2/underlay.p4
+++ b/dash-pipeline/bmv2/underlay.p4
@@ -50,6 +50,12 @@ control underlay(
 
     action def_act() {
 #ifdef TARGET_BMV2_V1MODEL
+#ifdef DPAPP_CONNTRACK
+        if (hdr.packet_meta.packet_source == dash_packet_source_t.DPAPP) {
+            standard_metadata.egress_spec = 0;
+        }
+        else
+#endif // DPAPP_CONNTRACK
         standard_metadata.egress_spec = standard_metadata.ingress_port;
 #endif // TARGET_BMV2_V1MODEL
 
@@ -82,6 +88,8 @@ control underlay(
             /* Send packet on same port it arrived (echo) by default */
             @defaultonly def_act;
         }
+
+        const default_action = def_act;
     }
 
     apply {