From e393697a2d2bf796f02bb4c70b784e0fa7ff78df Mon Sep 17 00:00:00 2001 From: Yaniv Kaul Date: Tue, 11 Aug 2026 13:54:37 +0300 Subject: [PATCH 1/2] Maintain token-aware routing when schema metadata is disabled With schema_metadata_enabled=False the driver skipped all schema refresh, so the token map never had keyspace replica information and token-aware routing was silently given away. Fetch keyspace replication strategies from the lightweight system_schema.keyspaces table on connect and on keyspace schema change events, keep them in Metadata._keyspace_replication_strategies, and have TokenMap fall back to them when keyspace metadata is unavailable. Dropped table events are also processed so tablet metadata stays up to date. Schema keyspace events are still ignored as before. --- CHANGELOG.rst | 4 + cassandra/cluster.py | 64 ++++++++- cassandra/metadata.py | 48 ++++++- tests/integration/standard/test_metadata.py | 34 +++++ tests/unit/test_control_connection.py | 138 +++++++++++++++++++ tests/unit/test_metadata.py | 141 +++++++++++++++++++- 6 files changed, 419 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 2a02f1ac54..d1ab4f1a8a 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -9,6 +9,10 @@ Features Others ------ +* Token-aware request routing is now maintained when ``schema_metadata_enabled`` is + disabled: the driver fetches the keyspace replication strategies from the lightweight + ``system_schema.keyspaces`` table (on connect and on keyspace schema change events), + and drops tablet metadata for dropped tables and keyspaces. * ``PreparedStatement.result_metadata`` and ``PreparedStatement.result_metadata_id`` are now read-only. They are replaced together by ``PreparedStatement.update_result_metadata()``, so a request can never observe a metadata diff --git a/cassandra/cluster.py b/cassandra/cluster.py index bcc7852c33..61a6599c88 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -69,7 +69,7 @@ RESULT_KIND_SET_KEYSPACE, RESULT_KIND_ROWS, RESULT_KIND_SCHEMA_CHANGE, ProtocolHandler, RESULT_KIND_VOID, ProtocolException) -from cassandra.metadata import Metadata, Token, protect_name, murmur3, _NodeInfo +from cassandra.metadata import Metadata, Token, protect_name, murmur3, _NodeInfo, ReplicationStrategy from cassandra.policies import (TokenAwarePolicy, DCAwareRoundRobinPolicy, SimpleConvictionPolicy, ExponentialReconnectionPolicy, HostDistance, RetryPolicy, IdentityTranslator, NoSpeculativeExecutionPlan, @@ -1153,8 +1153,9 @@ def schema_metadata_enabled(self): Flag indicating whether internal schema metadata is updated. When disabled, the driver does not populate Cluster.metadata.keyspaces on connect, or on schema change events. This - can be used to speed initial connection, and reduce load on client and server during operation. Turning this off - gives away token aware request routing, and programmatic inspection of the metadata model. + can be used to speed initial connection, and reduce load on client and server during operation. Token aware request + routing is still maintained using a lightweight query of keyspace replication strategies, but programmatic inspection + of the metadata model is not available. """ return self.control_connection._schema_meta_enabled @@ -3843,6 +3844,8 @@ class ControlConnection(object): _SELECT_SCHEMA_PEERS_TEMPLATE = "SELECT peer, host_id, {nt_col_name}, schema_version FROM system.peers" _SELECT_SCHEMA_LOCAL = "SELECT schema_version FROM system.local WHERE key='local'" + _SELECT_KEYSPACES_REPLICATION = "SELECT keyspace_name, replication FROM system_schema.keyspaces" + _SELECT_PEERS_V2 = "SELECT * FROM system.peers_v2" _SELECT_PEERS_NO_TOKENS_V2 = "SELECT host_id, peer, peer_port, data_center, rack, native_address, native_port, release_version, schema_version FROM system.peers_v2" _SELECT_SCHEMA_PEERS_V2 = "SELECT host_id, peer, peer_port, native_address, native_port, schema_version FROM system.peers_v2" @@ -4132,18 +4135,65 @@ def refresh_schema(self, force=False, **kwargs): self._signal_error() return False + def _refresh_replication_strategies(self, connection): + """ + Fetch the keyspace replication strategies from the lightweight + ``system_schema.keyspaces`` table. Used to support token-aware routing + when schema metadata is disabled. + """ + if not self._token_meta_enabled: + return + cl = ConsistencyLevel.ONE + query = QueryMessage( + query=maybe_add_timeout_to_query( + self._SELECT_KEYSPACES_REPLICATION, self._metadata_request_timeout), + consistency_level=cl) + try: + result = connection.wait_for_response(query, timeout=self._timeout) + rows = dict_factory(result.column_names, result.parsed_rows) + except Exception as exc: + # not supported by very old servers; degrade to no token-aware routing + log.warning("[control connection] Failed to fetch keyspace replication strategies, " + "token-aware routing may be degraded: %s", exc) + return + + strategies = {} + for row in rows: + try: + replication = dict(row["replication"]) + strategy_class = replication.pop("class") + except (TypeError, KeyError): + log.warning("[control connection] Skipping keyspace %s with unparseable " + "replication settings", row.get("keyspace_name")) + continue + strategy = ReplicationStrategy.create(strategy_class, replication) + if strategy: + strategies[row["keyspace_name"]] = strategy + self._cluster.metadata._update_replication_strategies(strategies) + def _refresh_schema(self, connection, preloaded_results=None, schema_agreement_wait=None, force=False, **kwargs): if self._cluster.is_shutdown: return False - agreed = self._wait_for_schema_agreement(connection=connection, - preloaded_results=preloaded_results, - wait_time=schema_agreement_wait) - if not self._schema_meta_enabled and not force: + target_type = kwargs.get("target_type") + change_type = kwargs.get("change_type") + if not target_type or target_type.lower() == "keyspace": + # keep token-aware routing functional without fetching the full schema + self._refresh_replication_strategies(connection) + elif target_type.lower() == "table" and change_type == "DROPPED": + # keep tablet metadata up to date for dropped tables + keyspace = kwargs.get("keyspace") + table = kwargs.get("table") + if keyspace and table: + self._cluster.metadata._table_removed(keyspace, table) log.debug("[control connection] Skipping schema refresh because schema metadata is disabled") return False + agreed = self._wait_for_schema_agreement(connection=connection, + preloaded_results=preloaded_results, + wait_time=schema_agreement_wait) + if not agreed: log.debug("Skipping schema refresh due to lack of schema agreement") return False diff --git a/cassandra/metadata.py b/cassandra/metadata.py index 5669e6a80e..deefc36421 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -129,6 +129,7 @@ def __init__(self): self._host_id_by_endpoint = {} self._hosts_lock = RLock() self._tablets = Tablets({}) + self._keyspace_replication_strategies = {} def export_schema_as_string(self): """ @@ -273,16 +274,52 @@ def _table_removed(self, keyspace, table): def _keyspace_added(self, ksname): if self.token_map: self.token_map.rebuild_keyspace(ksname, build_if_absent=False) + ks_meta = self.keyspaces.get(ksname) + if ks_meta and ks_meta.replication_strategy: + self._keyspace_replication_strategies[ksname] = ks_meta.replication_strategy def _keyspace_updated(self, ksname): if self.token_map: self.token_map.rebuild_keyspace(ksname, build_if_absent=False) self._tablets.drop_tablets(ksname) + ks_meta = self.keyspaces.get(ksname) + if ks_meta and ks_meta.replication_strategy: + self._keyspace_replication_strategies[ksname] = ks_meta.replication_strategy def _keyspace_removed(self, ksname): if self.token_map: self.token_map.remove_keyspace(ksname) self._tablets.drop_tablets(ksname) + self._keyspace_replication_strategies.pop(ksname, None) + + def _update_replication_strategies(self, replication_strategies): + """ + Update the keyspace replication strategies used for token-aware routing. + + Used to maintain the token map when schema metadata is disabled and + full keyspace metadata is not available. + """ + removed = set(self._keyspace_replication_strategies) - set(replication_strategies) + for keyspace in removed: + self._keyspace_replication_strategies.pop(keyspace, None) + self.keyspaces.pop(keyspace, None) # stale metadata if it was previously enabled + self._tablets.drop_tablets(keyspace) + if self.token_map: + self.token_map.remove_keyspace(keyspace) + changed = [(keyspace, strategy) for keyspace, strategy in replication_strategies.items() + if self._keyspace_replication_strategies.get(keyspace) != strategy] + self._keyspace_replication_strategies.update(replication_strategies) + token_map = self.token_map + for keyspace, strategy in changed: + ks_meta = self.keyspaces.get(keyspace) + if ks_meta is not None and ks_meta.replication_strategy != strategy: + # keep the metadata in sync so the token map rebuild uses the fresh strategy + ks_meta.replication_strategy = strategy + # a replication change can alter the tablet layout, so stale tablets are dropped + self._tablets.drop_tablets(keyspace) + if token_map: + token_map.remove_keyspace(keyspace) + token_map.rebuild_keyspace(keyspace, build_if_absent=True) def rebuild_token_map(self, partitioner, token_map): """ @@ -1847,10 +1884,17 @@ def rebuild_keyspace(self, keyspace, build_if_absent=False): try: current = self.tokens_to_hosts_by_ks.get(keyspace, None) if (build_if_absent and current is None) or (not build_if_absent and current is not None): + strategy = None ks_meta = self._metadata.keyspaces.get(keyspace) if ks_meta: - replica_map = self.replica_map_for_keyspace(self._metadata.keyspaces[keyspace]) - self.tokens_to_hosts_by_ks[keyspace] = replica_map + strategy = ks_meta.replication_strategy + if strategy is None: + strategy = self._metadata._keyspace_replication_strategies.get(keyspace) + if strategy: + replica_map = strategy.make_token_replica_map(self.token_to_host_owner, self.ring) + else: + replica_map = None + self.tokens_to_hosts_by_ks[keyspace] = replica_map except Exception: # should not happen normally, but we don't want to blow up queries because of unexpected meta state # bypass until new map is generated diff --git a/tests/integration/standard/test_metadata.py b/tests/integration/standard/test_metadata.py index 562f457a32..42a10f1dd9 100644 --- a/tests/integration/standard/test_metadata.py +++ b/tests/integration/standard/test_metadata.py @@ -30,6 +30,7 @@ RegisteredTableExtension, _RegisteredExtensionType, get_schema_parser, group_keys_by_replica, NO_VALID_REPLICA) from cassandra.protocol import QueryMessage, ProtocolHandler +from cassandra.query import SimpleStatement from tests.integration import (get_cluster, use_singledc, PROTOCOL_VERSION, execute_until_pass, BasicSegregatedKeyspaceUnitTestCase, BasicSharedKeyspaceUnitTestCase, @@ -170,6 +171,39 @@ def test_schema_metadata_disable(self): no_schema.shutdown() no_token.shutdown() + def test_schema_metadata_disable_token_aware_routing(self): + """ + Token-aware routing is maintained when schema metadata is disabled. + + @since 3.30 + @expected_result The token map is populated from keyspace replication strategies without full schema metadata + + @test_category metadata + """ + no_schema = TestCluster(schema_metadata_enabled=False) + try: + no_schema_session = no_schema.connect() + assert len(no_schema.metadata.keyspaces) == 0 + + # replication strategies are fetched without the full schema + assert no_schema.metadata._keyspace_replication_strategies.get(self.ks_name) is not None + + # token-aware routing works via the replication strategies + token_map = no_schema.metadata.token_map + replicas = token_map.get_replicas(self.ks_name, token_map.token_class.from_string("0")) + assert replicas + + # a token-aware query is routed to a replica of the keyspace + statement = SimpleStatement( + "SELECT * FROM system.local WHERE key='local'", + keyspace=self.ks_name, routing_key=b"routing-key") + rs = no_schema_session.execute(statement) + assert rs.one() is not None + replicas = set(no_schema.metadata.get_replicas(self.ks_name, statement.routing_key)) + assert rs.response_future._current_host in replicas + finally: + no_schema.shutdown() + def make_create_statement(self, partition_cols, clustering_cols=None, other_cols=None): clustering_cols = clustering_cols or [] other_cols = other_cols or [] diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py index fd62323f33..4a18118830 100644 --- a/tests/unit/test_control_connection.py +++ b/tests/unit/test_control_connection.py @@ -24,6 +24,8 @@ from cassandra.connection import EndPoint, DefaultEndPoint, DefaultEndPointFactory from cassandra.policies import (SimpleConvictionPolicy, RoundRobinPolicy, ConstantReconnectionPolicy, IdentityTranslator) +from cassandra.metadata import SimpleStrategy, LocalStrategy, Metadata +from cassandra.tablets import Tablet PEER_IP = "foobar" @@ -491,6 +493,142 @@ def test_handle_topology_change(self): self.control_connection._handle_topology_change(event) self.cluster.scheduler.schedule_unique.assert_called_once_with(ANY, self.control_connection._refresh_nodes_if_not_up, None) + def test_refresh_schema_disabled_fetches_replication_strategies(self): + """ + When schema metadata is disabled, the full schema refresh is skipped + but keyspace replication strategies are fetched for token-aware routing. + """ + self.control_connection._schema_meta_enabled = False + result = ResultMessage(kind=RESULT_KIND_ROWS) + result.column_names = ["keyspace_name", "replication"] + result.parsed_rows = [ + ["ks", {"class": "org.apache.cassandra.locator.SimpleStrategy", "replication_factor": "1"}], + ["system", {"class": "org.apache.cassandra.locator.LocalStrategy", "replication_factor": "1"}], + ] + self.connection.wait_for_response = Mock(return_value=result) + self.cluster.metadata._update_replication_strategies = Mock() + + self.control_connection._refresh_schema(self.connection) + + assert self.connection.wait_for_response.call_count == 1 + strategies = self.cluster.metadata._update_replication_strategies.call_args[0][0] + assert isinstance(strategies["ks"], SimpleStrategy) + assert strategies["ks"].replication_factor == 1 + assert isinstance(strategies["system"], LocalStrategy) + + def test_refresh_schema_disabled_keyspace_event_fetches_replication_strategies(self): + """ + Keyspace schema change events are processed when schema metadata is + disabled to keep token-aware routing up to date. + """ + self.control_connection._schema_meta_enabled = False + result = ResultMessage(kind=RESULT_KIND_ROWS) + result.column_names = ["keyspace_name", "replication"] + result.parsed_rows = [ + ["ks", {"class": "org.apache.cassandra.locator.SimpleStrategy", "replication_factor": "1"}], + ] + self.connection.wait_for_response = Mock(return_value=result) + self.cluster.metadata._update_replication_strategies = Mock() + + self.control_connection._refresh_schema( + self.connection, target_type="KEYSPACE", change_type="CREATED", keyspace="ks") + + self.cluster.metadata._update_replication_strategies.assert_called_once() + strategies = self.cluster.metadata._update_replication_strategies.call_args[0][0] + assert set(strategies) == {"ks"} + + def test_refresh_schema_disabled_table_dropped_invalidates_tablets(self): + """ + Dropped table events are processed when schema metadata is disabled so + tablet metadata stays up to date. + """ + self.control_connection._schema_meta_enabled = False + metadata = Metadata() + metadata._tablets.add_tablet("ks", "tb", Tablet(0, 100, [("host1", 0)])) + self.cluster.metadata = metadata + + self.control_connection._refresh_schema( + self.connection, target_type="TABLE", change_type="DROPPED", keyspace="ks", table="tb") + + assert metadata._tablets.table_has_tablets("ks", "tb") is False + + def test_refresh_schema_disabled_table_dropped_without_names(self): + """ + A dropped table event without keyspace/table names is ignored without + raising, keeping the control connection healthy. + """ + self.control_connection._schema_meta_enabled = False + self.cluster.metadata._table_removed = Mock() + + self.control_connection._refresh_schema( + self.connection, target_type="TABLE", change_type="DROPPED") + + self.cluster.metadata._table_removed.assert_not_called() + + def test_refresh_schema_disabled_ignores_other_events(self): + """ + Non-keyspace, non-dropped-table schema events are ignored when schema + metadata is disabled. + """ + self.control_connection._schema_meta_enabled = False + self.cluster.metadata._update_replication_strategies = Mock() + self.cluster.metadata._table_removed = Mock() + + self.control_connection._refresh_schema( + self.connection, target_type="TABLE", change_type="CREATED", keyspace="ks", table="tb") + + self.cluster.metadata._update_replication_strategies.assert_not_called() + self.cluster.metadata._table_removed.assert_not_called() + + def test_refresh_schema_disabled_skips_strategies_without_token_metadata(self): + """ + Replication strategies are not fetched when token metadata is disabled. + """ + self.control_connection._schema_meta_enabled = False + self.control_connection._token_meta_enabled = False + self.connection.wait_for_response = Mock() + self.cluster.metadata._update_replication_strategies = Mock() + + self.control_connection._refresh_schema(self.connection) + + self.connection.wait_for_response.assert_not_called() + self.cluster.metadata._update_replication_strategies.assert_not_called() + + def test_refresh_schema_disabled_degrades_gracefully_on_fetch_error(self): + """ + A failure fetching replication strategies is not propagated, so the + control connection stays healthy on servers without system_schema.keyspaces. + """ + self.control_connection._schema_meta_enabled = False + self.connection.wait_for_response = Mock(side_effect=OperationTimedOut()) + self.cluster.metadata._update_replication_strategies = Mock() + + assert self.control_connection._refresh_schema(self.connection) is False + + self.cluster.metadata._update_replication_strategies.assert_not_called() + + def test_refresh_schema_disabled_skips_malformed_strategy_row(self): + """ + A malformed row is skipped without discarding the strategies parsed + from the remaining valid rows. + """ + self.control_connection._schema_meta_enabled = False + result = ResultMessage(kind=RESULT_KIND_ROWS) + result.column_names = ["keyspace_name", "replication"] + result.parsed_rows = [ + ["bad", None], + ["good", {"class": "org.apache.cassandra.locator.SimpleStrategy", "replication_factor": "1"}], + ] + self.connection.wait_for_response = Mock(return_value=result) + self.cluster.metadata._update_replication_strategies = Mock() + + self.control_connection._refresh_schema(self.connection) + + self.cluster.metadata._update_replication_strategies.assert_called_once() + strategies = self.cluster.metadata._update_replication_strategies.call_args[0][0] + assert set(strategies) == {"good"} + assert isinstance(strategies["good"], SimpleStrategy) + def test_handle_status_change(self): event = { 'change_type': 'UP', diff --git a/tests/unit/test_metadata.py b/tests/unit/test_metadata.py index a058f73c61..571aa29537 100644 --- a/tests/unit/test_metadata.py +++ b/tests/unit/test_metadata.py @@ -15,7 +15,7 @@ from binascii import unhexlify import logging -from unittest.mock import Mock +from unittest.mock import Mock, patch import os import timeit import uuid @@ -443,6 +443,145 @@ def test_bytes_tokens(self): self._get_replicas(BytesToken) +class TokenMapReplicationStrategiesTest(unittest.TestCase): + """ + TokenMap should use Metadata._keyspace_replication_strategies to compute + replicas when keyspace metadata is unavailable (schema_metadata_enabled=False). + """ + + def _build_metadata(self): + tokens = [Murmur3Token(i) for i in range(0, 500, 100)] + hosts = [Host("ip%d" % i, SimpleConvictionPolicy, datacenter="dc1", rack="rack1", host_id=uuid.uuid4()) + for i in range(len(tokens))] + metadata = Metadata() + metadata.rebuild_token_map( + "org.apache.cassandra.dht.Murmur3Partitioner", + {host: [str(token.value)] for host, token in zip(hosts, tokens)}) + return metadata, tokens, hosts + + def test_get_replicas_without_keyspace_metadata(self): + """Replica lookup works without keyspace metadata when the replication strategy is known.""" + metadata, tokens, hosts = self._build_metadata() + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + + replicas = metadata.token_map.get_replicas("ks", tokens[0]) + assert set(replicas) == {hosts[0]} + assert metadata.keyspaces == {} + + def test_update_replication_strategies_removes_dropped_keyspaces(self): + """Removing a keyspace drops its token map entry and tablets.""" + metadata, _, _ = self._build_metadata() + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + metadata._tablets.add_tablet("ks", "tb", Tablet(0, 100, [("host1", 0)])) + assert "ks" in metadata._keyspace_replication_strategies + assert "ks" in metadata.token_map.tokens_to_hosts_by_ks + + metadata._update_replication_strategies({}) + + assert metadata._keyspace_replication_strategies == {} + assert metadata.token_map.tokens_to_hosts_by_ks == {} + assert metadata._tablets.table_has_tablets("ks", "tb") is False + + def test_keyspace_added_syncs_replication_strategies(self): + """Schema refresh keeps _keyspace_replication_strategies in sync with keyspace metadata.""" + metadata, _, _ = self._build_metadata() + keyspace = KeyspaceMetadata("ks", True, "NetworkTopologyStrategy", {"dc1": "1"}) + metadata.keyspaces["ks"] = keyspace + metadata._keyspace_added("ks") + + assert metadata._keyspace_replication_strategies["ks"] is keyspace.replication_strategy + + def test_keyspace_removed_syncs_replication_strategies(self): + """Removing a keyspace clears its replication strategy and tablets.""" + metadata, _, _ = self._build_metadata() + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + metadata._tablets.add_tablet("ks", "tb", Tablet(0, 100, [("host1", 0)])) + + metadata._keyspace_removed("ks") + + assert metadata._keyspace_replication_strategies == {} + assert metadata._tablets.table_has_tablets("ks", "tb") is False + + def test_rebuild_token_map_keeps_routing_after_topology_change(self): + """A rebuilt token map still routes when schema metadata is disabled (rebuilt lazily).""" + metadata, tokens, hosts = self._build_metadata() + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + + metadata.rebuild_token_map( + "org.apache.cassandra.dht.Murmur3Partitioner", + {host: [str(token.value)] for host, token in zip(hosts, tokens)}) + + assert set(metadata.token_map.get_replicas("ks", tokens[0])) == {hosts[0]} + + def test_update_replication_strategies_change_drops_tablets_and_rebuilds(self): + """A changed replication strategy drops tablets and rebuilds the token map.""" + metadata, tokens, hosts = self._build_metadata() + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + metadata._tablets.add_tablet("ks", "tb", Tablet(0, 100, [("host1", 0)])) + + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "2"})}) + + assert metadata._tablets.table_has_tablets("ks", "tb") is False + assert set(metadata.token_map.get_replicas("ks", tokens[0])) == {hosts[0], hosts[1]} + + def test_update_replication_strategies_skips_unchanged(self): + """An unchanged replication strategy does not rebuild the token map or drop tablets.""" + metadata, _, _ = self._build_metadata() + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + metadata._tablets.add_tablet("ks", "tb", Tablet(0, 100, [("host1", 0)])) + before = dict(metadata.token_map.tokens_to_hosts_by_ks["ks"]) + + with patch.object(TokenMap, "rebuild_keyspace") as rebuild_keyspace: + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + + rebuild_keyspace.assert_not_called() + assert metadata._tablets.table_has_tablets("ks", "tb") is True + assert metadata.token_map.tokens_to_hosts_by_ks["ks"] == before + + def test_update_replication_strategies_change_syncs_keyspace_metadata(self): + """A strategy change updates stale keyspace metadata so the token map uses the fresh strategy.""" + metadata, tokens, hosts = self._build_metadata() + keyspace = KeyspaceMetadata("ks", True, "NetworkTopologyStrategy", {"dc1": "1"}) + metadata.keyspaces["ks"] = keyspace + metadata._keyspace_added("ks") + assert set(metadata.token_map.get_replicas("ks", tokens[0])) == {hosts[0]} + + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.NetworkTopologyStrategy", {"dc1": "2"})}) + + assert keyspace.replication_strategy is metadata._keyspace_replication_strategies["ks"] + assert set(metadata.token_map.get_replicas("ks", tokens[0])) == {hosts[0], hosts[1]} + + def test_update_replication_strategies_removed_syncs_keyspace_metadata(self): + """A dropped keyspace is removed from keyspace metadata when it was previously enabled.""" + metadata, _, _ = self._build_metadata() + metadata.keyspaces["ks"] = KeyspaceMetadata("ks", True, "NetworkTopologyStrategy", {"dc1": "1"}) + metadata._keyspace_added("ks") + + metadata._update_replication_strategies({}) + + assert metadata.keyspaces == {} + assert metadata._keyspace_replication_strategies == {} + assert metadata.token_map.tokens_to_hosts_by_ks == {} + + class Murmur3TokensTest(unittest.TestCase): def test_murmur3_init(self): From 430445fcc0dd8bdd8c86c2cb053a623120db8268 Mon Sep 17 00:00:00 2001 From: Yaniv Kaul Date: Tue, 11 Aug 2026 23:26:25 +0300 Subject: [PATCH 2/2] Address review comments on token-aware routing without schema metadata - Wait for schema agreement before the disabled-metadata branch so the CL.ONE replication read cannot return pre-agreement data - Filter the keyspace read to the changed keyspace on keyspace events, merging the result into the known strategies (a missing row drops it) - Deprecate the now-unused TokenMap.replica_map_for_keyspace - Assert on the first attempted host in the integration test, exercise the real _aggregate_results in the consistency-mode unit tests, and add regression tests for the agreement-first behavior Co-Authored-By: Claude Fable 5 --- cassandra/cluster.py | 40 +++++++---- cassandra/metadata.py | 3 + tests/integration/standard/test_metadata.py | 3 +- tests/unit/test_control_connection.py | 80 ++++++++++++++++++++- tests/unit/test_metadata.py | 14 ++-- 5 files changed, 118 insertions(+), 22 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 61a6599c88..0b17f2bd67 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -4135,18 +4135,21 @@ def refresh_schema(self, force=False, **kwargs): self._signal_error() return False - def _refresh_replication_strategies(self, connection): + def _refresh_replication_strategies(self, connection, keyspace=None): """ Fetch the keyspace replication strategies from the lightweight ``system_schema.keyspaces`` table. Used to support token-aware routing - when schema metadata is disabled. + when schema metadata is disabled. When ``keyspace`` is given, only that + keyspace is re-read and merged into the current strategies. """ if not self._token_meta_enabled: return + select = self._SELECT_KEYSPACES_REPLICATION + if keyspace is not None: + select += bind_params(" WHERE keyspace_name = %s", (keyspace,), Encoder()) cl = ConsistencyLevel.ONE query = QueryMessage( - query=maybe_add_timeout_to_query( - self._SELECT_KEYSPACES_REPLICATION, self._metadata_request_timeout), + query=maybe_add_timeout_to_query(select, self._metadata_request_timeout), consistency_level=cl) try: result = connection.wait_for_response(query, timeout=self._timeout) @@ -4169,18 +4172,37 @@ def _refresh_replication_strategies(self, connection): strategy = ReplicationStrategy.create(strategy_class, replication) if strategy: strategies[row["keyspace_name"]] = strategy + + if keyspace is not None: + # merge the single re-read keyspace into the current map; a missing + # row means the keyspace was dropped + merged = dict(self._cluster.metadata._keyspace_replication_strategies) + merged.pop(keyspace, None) + merged.update(strategies) + strategies = merged self._cluster.metadata._update_replication_strategies(strategies) def _refresh_schema(self, connection, preloaded_results=None, schema_agreement_wait=None, force=False, **kwargs): if self._cluster.is_shutdown: return False + agreed = self._wait_for_schema_agreement(connection=connection, + preloaded_results=preloaded_results, + wait_time=schema_agreement_wait) + + if not agreed: + log.debug("Skipping schema refresh due to lack of schema agreement") + return False + if not self._schema_meta_enabled and not force: target_type = kwargs.get("target_type") change_type = kwargs.get("change_type") - if not target_type or target_type.lower() == "keyspace": + if not target_type: # keep token-aware routing functional without fetching the full schema self._refresh_replication_strategies(connection) + elif target_type.lower() == "keyspace": + # a single keyspace changed; no need to re-read the whole table + self._refresh_replication_strategies(connection, keyspace=kwargs.get("keyspace")) elif target_type.lower() == "table" and change_type == "DROPPED": # keep tablet metadata up to date for dropped tables keyspace = kwargs.get("keyspace") @@ -4190,14 +4212,6 @@ def _refresh_schema(self, connection, preloaded_results=None, schema_agreement_w log.debug("[control connection] Skipping schema refresh because schema metadata is disabled") return False - agreed = self._wait_for_schema_agreement(connection=connection, - preloaded_results=preloaded_results, - wait_time=schema_agreement_wait) - - if not agreed: - log.debug("Skipping schema refresh due to lack of schema agreement") - return False - self._cluster.metadata.refresh( connection, self._timeout, diff --git a/cassandra/metadata.py b/cassandra/metadata.py index deefc36421..23205f375c 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -24,6 +24,7 @@ import re import sys from threading import RLock +from warnings import warn import struct import random import itertools @@ -1902,6 +1903,8 @@ def rebuild_keyspace(self, keyspace, build_if_absent=False): log.exception("Failed creating a token map for keyspace '%s' with %s. PLEASE REPORT THIS: https://datastax-oss.atlassian.net/projects/PYTHON", keyspace, self.token_to_host_owner) def replica_map_for_keyspace(self, ks_metadata): + warn("TokenMap.replica_map_for_keyspace is deprecated and will be " + "removed in a future release.", DeprecationWarning, stacklevel=2) strategy = ks_metadata.replication_strategy if strategy: return strategy.make_token_replica_map(self.token_to_host_owner, self.ring) diff --git a/tests/integration/standard/test_metadata.py b/tests/integration/standard/test_metadata.py index 42a10f1dd9..26169dbfbd 100644 --- a/tests/integration/standard/test_metadata.py +++ b/tests/integration/standard/test_metadata.py @@ -200,7 +200,8 @@ def test_schema_metadata_disable_token_aware_routing(self): rs = no_schema_session.execute(statement) assert rs.one() is not None replicas = set(no_schema.metadata.get_replicas(self.ks_name, statement.routing_key)) - assert rs.response_future._current_host in replicas + # the first host attempted; retries may move the query off a replica + assert rs.response_future.attempted_hosts[0] in replicas finally: no_schema.shutdown() diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py index 4a18118830..fa8938c5a6 100644 --- a/tests/unit/test_control_connection.py +++ b/tests/unit/test_control_connection.py @@ -51,6 +51,7 @@ def __init__(self): self.partitioner = None self.token_map = {} self.removed_hosts = [] + self._keyspace_replication_strategies = {} def get_host(self, endpoint_or_address, port=None): if not isinstance(endpoint_or_address, EndPoint): @@ -519,9 +520,11 @@ def test_refresh_schema_disabled_fetches_replication_strategies(self): def test_refresh_schema_disabled_keyspace_event_fetches_replication_strategies(self): """ Keyspace schema change events are processed when schema metadata is - disabled to keep token-aware routing up to date. + disabled to keep token-aware routing up to date. The read is filtered + to the changed keyspace and merged into the known strategies. """ self.control_connection._schema_meta_enabled = False + self.cluster.metadata._keyspace_replication_strategies = {"other": SimpleStrategy({"replication_factor": "3"})} result = ResultMessage(kind=RESULT_KIND_ROWS) result.column_names = ["keyspace_name", "replication"] result.parsed_rows = [ @@ -533,9 +536,82 @@ def test_refresh_schema_disabled_keyspace_event_fetches_replication_strategies(s self.control_connection._refresh_schema( self.connection, target_type="KEYSPACE", change_type="CREATED", keyspace="ks") + query = self.connection.wait_for_response.call_args[0][0].query + assert "WHERE keyspace_name = 'ks'" in query self.cluster.metadata._update_replication_strategies.assert_called_once() strategies = self.cluster.metadata._update_replication_strategies.call_args[0][0] - assert set(strategies) == {"ks"} + assert set(strategies) == {"ks", "other"} + + def test_refresh_schema_disabled_keyspace_drop_removes_strategy(self): + """ + A dropped keyspace's strategy is removed: the filtered read returns no + row and the merge drops the keyspace from the known strategies. + """ + self.control_connection._schema_meta_enabled = False + self.cluster.metadata._keyspace_replication_strategies = { + "ks": SimpleStrategy({"replication_factor": "1"}), + "other": SimpleStrategy({"replication_factor": "3"}), + } + result = ResultMessage(kind=RESULT_KIND_ROWS) + result.column_names = ["keyspace_name", "replication"] + result.parsed_rows = [] + self.connection.wait_for_response = Mock(return_value=result) + self.cluster.metadata._update_replication_strategies = Mock() + + self.control_connection._refresh_schema( + self.connection, target_type="KEYSPACE", change_type="DROPPED", keyspace="ks") + + strategies = self.cluster.metadata._update_replication_strategies.call_args[0][0] + assert set(strategies) == {"other"} + + def test_refresh_schema_disabled_waits_for_schema_agreement(self): + """ + The replication-strategy read runs at ConsistencyLevel.ONE, so it must + only happen after schema agreement; a pre-agreement read can return the + stale pre-change replication settings. + """ + self.control_connection._schema_meta_enabled = False + + # one peer starts on a different schema version and converges later + self.connection.peer_results[1][1][2] = 'b' + + def converge_then_respond(*args, **kwargs): + if self.time.clock >= 1: + self.connection.peer_results[1][1][2] = 'a' + return _node_meta_results(self.connection.local_results, self.connection.peer_results) + + self.connection.wait_for_responses = Mock(side_effect=converge_then_respond) + + result = ResultMessage(kind=RESULT_KIND_ROWS) + result.column_names = ["keyspace_name", "replication"] + result.parsed_rows = [ + ["ks", {"class": "org.apache.cassandra.locator.SimpleStrategy", "replication_factor": "1"}], + ] + self.connection.wait_for_response = Mock(return_value=result) + self.cluster.metadata._update_replication_strategies = Mock() + + self.control_connection._refresh_schema( + self.connection, target_type="KEYSPACE", change_type="UPDATED", keyspace="ks") + + # slept until the disagreeing peer converged, then fetched strategies + assert self.time.clock > 0 + self.cluster.metadata._update_replication_strategies.assert_called_once() + + def test_refresh_schema_disabled_no_agreement_skips_strategies(self): + """ + Without schema agreement the strategies are not fetched at all, rather + than being fetched from a node with a stale schema. + """ + self.control_connection._schema_meta_enabled = False + # a peer that never agrees + self.connection.peer_results[1][1][2] = 'b' + self.connection.wait_for_response = Mock() + self.cluster.metadata._update_replication_strategies = Mock() + + assert self.control_connection._refresh_schema(self.connection) is False + + self.connection.wait_for_response.assert_not_called() + self.cluster.metadata._update_replication_strategies.assert_not_called() def test_refresh_schema_disabled_table_dropped_invalidates_tablets(self): """ diff --git a/tests/unit/test_metadata.py b/tests/unit/test_metadata.py index 571aa29537..170ae14b2b 100644 --- a/tests/unit/test_metadata.py +++ b/tests/unit/test_metadata.py @@ -780,10 +780,12 @@ class ScyllaKeyspaceConsistencyParsingTest(unittest.TestCase): """ def _parser_with_rows(self, rows): - # Build the parser without a connection and drive only the aggregation + # Build the parser without a connection and drive the real aggregation # step; _query_all's batching is exercised by the integration tests. - parser = SchemaParserV3.__new__(SchemaParserV3) + parser = SchemaParserV3(None, 1.0, 100, None) + parser.views_result = [] # populated by _query_all, not __init__ parser.scylla_keyspaces_result = rows + parser._aggregate_results() return parser def test_consistency_modes_are_mapped_from_rows(self): @@ -796,8 +798,7 @@ def test_consistency_modes_are_mapped_from_rows(self): {'keyspace_name': 'e', 'consistency': 'eventual'}, {'keyspace_name': 'n', 'consistency': None}, ]) - modes = {row["keyspace_name"]: _consistency_mode_from_string(row.get("consistency")) - for row in parser.scylla_keyspaces_result} + modes = parser.keyspace_consistency_modes assert modes['g'] == _ConsistencyMode.GLOBAL assert modes['l'] == _ConsistencyMode.LOCAL assert modes['e'] == _ConsistencyMode.EVENTUAL @@ -806,8 +807,9 @@ def test_consistency_modes_are_mapped_from_rows(self): def test_keyspace_absent_from_the_map_is_eventual(self): # Covers the whole-cluster fallbacks too: no rows is what a skipped read # (no TABLETS_ROUTING_V2) and a missing table/column both produce. - parser = SchemaParserV3.__new__(SchemaParserV3) - parser.keyspace_consistency_modes = {'g': _ConsistencyMode.GLOBAL} + parser = self._parser_with_rows([ + {'keyspace_name': 'g', 'consistency': 'global'}, + ]) assert parser.keyspace_consistency_modes.get( 'absent', _ConsistencyMode.EVENTUAL) == _ConsistencyMode.EVENTUAL