test/tablets: Check that after RF change data is replicated properly

There's a test that checks system.tablets contents to see that after
changing ks replication factor via ALTER KEYSPACE the tablet map is
updated properly. This patch extends this test that also validates that
mutations themselves are replicated according to the desired replication
factor.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>

Closes scylladb/scylladb#18644

test/topology_custom/test_tablets.py

@@ -8,11 +8,15 @@ from cassandra.query import SimpleStatement, ConsistencyLevel
 from test.pylib.manager_client import ManagerClient
 from test.pylib.rest_client import HTTPError
 from test.pylib.tablets import get_all_tablet_replicas
+from test.pylib.util import read_barrier
+from test.topology.util import wait_for_cql_and_get_hosts
+import time
 import pytest
 import logging
 import asyncio
 import re
 import requests
+import random
 
 logger = logging.getLogger(__name__)
 
@@ -143,6 +147,20 @@ async def test_tablet_rf_change(manager: ManagerClient, direction):
     logger.info(f"Checking {rf_to} re-allocated replicas")
     await check_allocated_replica(rf_to)
 
+    fragments = { pk: [] for pk in random.sample(range(128), 17) }
+    for s in servers:
+        host_id = await manager.get_host_id(s.server_id)
+        host = await wait_for_cql_and_get_hosts(cql, [s], time.time() + 30)
+        await read_barrier(manager.get_cql(), host[0]) # scylladb/scylladb#18199
+        for k in fragments:
+            res = await cql.run_async(f"SELECT partition_region FROM MUTATION_FRAGMENTS(test.test) WHERE pk={k}", host=host[0])
+            for fragment in res:
+                if fragment.partition_region == 0: # partition start
+                    fragments[k].append(host_id)
+    logger.info("Checking fragments")
+    for k in fragments:
+        assert len(fragments[k]) == rf_to, f"Found mutations for {k} key on {fragments[k]} hosts, but expected only {rf_to} of them"
+
 
 # Reproducer for https://github.com/scylladb/scylladb/issues/18110
 # Check that an existing cached read, will be cleaned up when the tablet it reads