diff --git a/tests/bgp/test_ipv6_bgp_scale.py b/tests/bgp/test_ipv6_bgp_scale.py
index 70b45a99dd2..b966d541e6e 100644
--- a/tests/bgp/test_ipv6_bgp_scale.py
+++ b/tests/bgp/test_ipv6_bgp_scale.py
@@ -415,6 +415,26 @@ def check_bgp_routes_converged(duthost, expected_routes, shutdown_connections=No
         pytest.fail(f"BGP routes aren't stable in {timeout} seconds")
 
 
+@pytest.fixture(scope="function")
+def clean_ptf_dataplane(ptfadapter):
+    """
+    Drain queued packets and clear mask counters before and after each test.
+    The idea is that each test should start with clean dataplane state without
+    having to restart ptfadapter fixture for each test.
+    Takes in the function scope so that each parametrized test case also gets a clean dataplane.
+    """
+    dp = ptfadapter.dataplane
+
+    def _perform_cleanup_on_dp():
+        dp.drain()
+        dp.clear_masks()
+    # Before test run DP cleanup
+    _perform_cleanup_on_dp()
+    yield
+    # After test run DP cleanup
+    _perform_cleanup_on_dp()
+
+
 def compress_expected_routes(expected_routes):
     json_str = json.dumps(expected_routes)
     compressed = gzip.compress(json_str.encode('utf-8'))
@@ -503,7 +523,7 @@ def _select_targets_to_flap(bgp_peers_info, all_flap, flapping_count):
     return flapping_neighbors, injection_neighbor, flapping_ports, injection_port
 
 
-def flapper(duthost, pdp, bgp_peers_info, transient_setup, flapping_count, connection_type, action):
+def flapper(duthost, ptfadapter, bgp_peers_info, transient_setup, flapping_count, connection_type, action):
     """
     Orchestrates interface/BGP session flapping and recovery on the DUT, generating test traffic to assess both
     control and data plane convergence behavior. This function is designed for use in test scenarios
@@ -525,6 +545,9 @@ def flapper(duthost, pdp, bgp_peers_info, transient_setup, flapping_count, conne
     global global_icmp_type, current_test, test_results
     current_test = f"flapper_{action}_{connection_type}_count_{flapping_count}"
     global_icmp_type += 1
+    pdp = ptfadapter.dataplane
+    pdp.clear_masks()
+    pdp.set_qlen(PACKET_QUEUE_LENGTH)
     exp_mask = setup_packet_mask_counters(pdp, global_icmp_type)
     all_flap = (flapping_count == 'all')
 
@@ -785,7 +808,9 @@ def test_bgp_admin_flap(
     duthost,
     ptfadapter,
     bgp_peers_info,
-    flapping_neighbor_count
+    clean_ptf_dataplane,
+    flapping_neighbor_count,
+    setup_routes_before_test
 ):
     """
     Validates that both control plane and data plane remain functional with acceptable downtime when BGP sessions are
@@ -798,12 +823,11 @@ def test_bgp_admin_flap(
     Expected result:
         Dataplane downtime is less than MAX_BGP_SESSION_DOWNTIME or MAX_DOWNTIME_UNISOLATION for all ports.
     """
-    pdp = ptfadapter.dataplane
-    pdp.set_qlen(PACKET_QUEUE_LENGTH)
     # Measure shutdown convergence
-    transient_setup = flapper(duthost, pdp, bgp_peers_info, None, flapping_neighbor_count, 'bgp_sessions', 'shutdown')
+    transient_setup = flapper(duthost, ptfadapter, bgp_peers_info, None, flapping_neighbor_count,
+                              'bgp_sessions', 'shutdown')
     # Measure startup convergence
-    flapper(duthost, pdp, None, transient_setup, flapping_neighbor_count, 'bgp_sessions', 'startup')
+    flapper(duthost, ptfadapter, None, transient_setup, flapping_neighbor_count, 'bgp_sessions', 'startup')
 
 
 @pytest.mark.parametrize("flapping_port_count", [1, 10, 20, 'all'])
@@ -812,6 +836,7 @@ def test_sessions_flapping(
     duthost,
     ptfadapter,
     bgp_peers_info,
+    clean_ptf_dataplane,
     flapping_port_count,
     setup_routes_before_test
 ):
@@ -826,10 +851,7 @@ def test_sessions_flapping(
     Expected result:
         Dataplane downtime is less than MAX_DOWNTIME_PORT_FLAPPING or MAX_DOWNTIME_UNISOLATION for all ports.
     '''
-    pdp = ptfadapter.dataplane
-    pdp.set_qlen(PACKET_QUEUE_LENGTH)
-
     # Measure shutdown convergence
-    transient_setup = flapper(duthost, pdp, bgp_peers_info, None, flapping_port_count, 'ports', 'shutdown')
+    transient_setup = flapper(duthost, ptfadapter, bgp_peers_info, None, flapping_port_count, 'ports', 'shutdown')
     # Measure startup convergence
-    flapper(duthost, pdp, None, transient_setup, flapping_port_count, 'ports', 'startup')
+    flapper(duthost, ptfadapter, None, transient_setup, flapping_port_count, 'ports', 'startup')
diff --git a/tests/common/plugins/ptfadapter/ptfadapter.py b/tests/common/plugins/ptfadapter/ptfadapter.py
index 47bbffbbafb..6b29eb89431 100644
--- a/tests/common/plugins/ptfadapter/ptfadapter.py
+++ b/tests/common/plugins/ptfadapter/ptfadapter.py
@@ -129,6 +129,39 @@ def _init_ptf_dataplane(self, ptf_config=None):
                 packet
             )
         self.dataplane = ptf.dataplane_instance
+        self._attach_cleanup_helpers()
+
+    def _attach_cleanup_helpers(self):
+        dp = self.dataplane
+
+        def drain(max_per_port=800):
+            """
+            Best-effort non-blocking drain of residual queued packets per port.
+            Prevents backlog from prior test affecting pps/downtime.
+            """
+            try:
+                for (dev, port) in list(getattr(dp, "ports", {}).keys()):
+                    drained = 0
+                    while drained < max_per_port:
+                        pkt = dp.poll(device_number=dev, port_number=port, timeout=0)
+                        if pkt is None:
+                            break
+                        drained += 1
+            except Exception:
+                pass
+
+        def clear_masks():
+            """
+            Remove any previously registered Mask counters to avoid cumulative match overhead.
+            """
+            try:
+                dp.mask_rx_cnt.clear()
+                dp.mask_tx_cnt.clear()
+                dp.masked_packets.clear()
+            except Exception:
+                pass
+        dp.drain = drain
+        dp.clear_masks = clear_masks
 
     def kill(self):
         """ Close dataplane socket and kill data plane thread """