Merge remote-tracking branch 'refs/remotes/origin/anaik-mutation-tracking-metrics' into anaik-mutation-tracking-metrics

# Conflicts:
#	test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingMetricsTest.java

Author: aparnanaik0522

test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingMetricsTest.java

@@ -149,12 +149,16 @@ public void testBroadcastOffsetsDiscoveredMetric() throws Throwable
             cluster.get(1).runOnInstance(() -> MutationTrackingService.instance.broadcastOffsetsForTesting());
 
             // Wait for broadcasts to propagate to node 3
+            long[] previousCount = {0};
             Awaitility.await()
                     .atMost(Duration.ofSeconds(5))
                     .pollInterval(Duration.ofMillis(100))
                     .until(() -> {
-                        long node3Delta = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance.broadcastOffsetsDiscovered.getCount()) - initialNode3Count;
-                        return node3Delta > 0;
+                        long currentCount = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance.broadcastOffsetsDiscovered.getCount());
+                        boolean hasDiscoveredOffsets = currentCount > initialNode3Count;
+                        boolean isStable = hasDiscoveredOffsets && currentCount == previousCount[0];
+                        previousCount[0] = currentCount;
+                        return isStable;
                     });
 
             // Get the count after first broadcast