test: add test to check if repair handles no_such_keyspace

locator/abstract_replication_strategy.cc

@@ -657,6 +657,8 @@ future<> global_vnode_effective_replication_map::get_keyspace_erms(sharded<repli
         // all under the lock.
         auto lk = co_await db.get_shared_token_metadata().get_lock();
         auto erm = db.find_keyspace(keyspace_name).get_vnode_effective_replication_map();
+        utils::get_local_injector().inject("get_keyspace_erms_throw_no_such_keyspace",
+                [&keyspace_name] { throw data_dictionary::no_such_keyspace{keyspace_name}; });
         auto ring_version = erm->get_token_metadata().get_ring_version();
         _erms[0] = make_foreign(std::move(erm));
         co_await coroutine::parallel_for_each(std::views::iota(1u, smp::count), [this, &sharded_db, keyspace_name, ring_version] (unsigned shard) -> future<> {

test/topology_custom/test_repair.py

@@ -259,3 +259,19 @@ async def test_repair_abort(manager):
     await manager.api.client.get_json(f"/task_manager/wait_task/{id}", host=servers[0].ip_addr)
     statuses = await manager.api.client.get_json(f"/task_manager/task_status_recursive/{id}", host=servers[0].ip_addr)
     assert all([status["state"] == "failed" for status in statuses])
+
+@pytest.mark.asyncio
+@skip_mode('release', 'error injections are not supported in release mode')
+async def test_keyspace_drop_during_data_sync_repair(manager):
+    cfg = {
+        'enable_tablets': False,
+        'error_injections_at_startup': ['get_keyspace_erms_throw_no_such_keyspace']
+    }
+    await manager.server_add(config=cfg)
+
+    cql = manager.get_cql()
+
+    cql.execute("CREATE KEYSPACE ks WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor': 2}")
+    cql.execute("CREATE TABLE ks.tbl (pk int, ck int, PRIMARY KEY (pk, ck)) WITH tombstone_gc = {'mode': 'repair'}")
+
+    await manager.server_add(config=cfg)