From e2320d750c5f42ef7681244508a2ea486ad566ea Mon Sep 17 00:00:00 2001 From: Yaniv Kaul Date: Wed, 19 Aug 2026 10:49:49 +0300 Subject: [PATCH 1/3] Change prepare_on_all_hosts default to False In multi-DC deployments, eager preparation previously ran on every host with an open connection pool, including remote hosts that are rarely or never queried on the happy path. Since an UNPREPARED response always triggers on-demand reprepare and retry, eager preparation is purely a latency optimization and does not affect correctness. Disabling it by default avoids paying for that optimization on rarely-used remote hosts. --- CHANGELOG.rst | 4 ++++ cassandra/cluster.py | 9 ++++++--- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 26725cfedc..993c510dcb 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -58,6 +58,10 @@ Others come through unchanged. Previously ``DRIVER_NAME`` and ``DRIVER_VERSION`` could be overridden, which misreported the driver to the server for the life of the connection and, in the clients table, to the operator reading the row. +* ``Cluster.prepare_on_all_hosts`` now defaults to ``False``. In multi-DC deployments eager + preparation previously ran on every pooled host, including remote hosts that are rarely or + never queried. Disabling it by default avoids that waste; an ``UNPREPARED`` response still + triggers on-demand reprepare and retry, so correctness is unaffected. * ``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 d858f5835e..43007f5f91 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -985,11 +985,14 @@ def default_retry_policy(self, policy): establish connection pools. This can cause a rush of connections and queries if not mitigated with this factor. """ - prepare_on_all_hosts = True + prepare_on_all_hosts = False """ Specifies whether statements should be prepared on all hosts, or just one. - This can reasonably be disabled on long-running applications with numerous clients preparing statements on startup, + When enabled, statements are eagerly prepared on every host with an open connection pool. In multi-DC + deployments this includes remote hosts that are rarely or never queried on the happy path; preparing on them + is purely a latency optimization, since an ``UNPREPARED`` response always triggers on-demand reprepare and + retry. It can be enabled on long-running applications with numerous clients preparing statements on startup, where a randomized initial condition of the load balancing policy can be expected to distribute prepares from different clients across the cluster. """ @@ -1205,7 +1208,7 @@ def __init__(self, schema_metadata_page_size=1000, address_translator=None, status_event_refresh_window=2, - prepare_on_all_hosts=True, + prepare_on_all_hosts=False, reprepare_on_up=True, execution_profiles=None, allow_beta_protocol_version=False, From d46a26cc2ec22389eb3a3ccb941025411a76eeb0 Mon Sep 17 00:00:00 2001 From: Yaniv Kaul Date: Sun, 23 Aug 2026 10:16:40 +0300 Subject: [PATCH 2/3] tests: use the last query trace id, not the first prepare_on_all_hosts now defaults to False, so a query against a host that hasn't prepared the statement yet can get UNPREPARED and silently retry. That appends an earlier, incomplete trace to the response future before the one that actually produced the result, so indexing get_query_trace_ids() with [0] picked the wrong (empty) trace. get_query_trace() (singular) already uses the last trace; match that. --- tests/integration/standard/test_shard_aware.py | 5 ++++- tests/integration/standard/test_tablets.py | 8 ++++++-- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/tests/integration/standard/test_shard_aware.py b/tests/integration/standard/test_shard_aware.py index 6daba6e26f..654ddb20df 100644 --- a/tests/integration/standard/test_shard_aware.py +++ b/tests/integration/standard/test_shard_aware.py @@ -71,7 +71,10 @@ def verify_same_shard_in_tracing(self, results, shard_name): assert shard_name in event.thread_name assert 'querying locally' in "\n".join([event.description for event in events]) - trace_id = results.response_future.get_query_trace_ids()[0] + # Use the last trace id: prepare_on_all_hosts defaults to False now, so a query + # against a host that hasn't prepared the statement yet can get UNPREPARED and + # retry, which appends an earlier, incomplete trace before the one that matters. + trace_id = results.response_future.get_query_trace_ids()[-1] traces = self.session.execute("SELECT * FROM system_traces.events WHERE session_id = %s", (trace_id,)) events = [event for event in traces] for event in events: diff --git a/tests/integration/standard/test_tablets.py b/tests/integration/standard/test_tablets.py index 0491b15f3f..4a37b7a84f 100644 --- a/tests/integration/standard/test_tablets.py +++ b/tests/integration/standard/test_tablets.py @@ -37,7 +37,10 @@ def verify_hosts_in_tracing(self, results, expected): assert len(host_set) == expected assert 'locally' in "\n".join([event.description for event in events]) - trace_id = results.response_future.get_query_trace_ids()[0] + # Use the last trace id: prepare_on_all_hosts defaults to False now, so a query + # against a host that hasn't prepared the statement yet can get UNPREPARED and + # retry, which appends an earlier, incomplete trace before the one that matters. + trace_id = results.response_future.get_query_trace_ids()[-1] traces = self.session.execute("SELECT * FROM system_traces.events WHERE session_id = %s", (trace_id,)) events = [event for event in traces] host_set = set() @@ -63,7 +66,8 @@ def verify_same_shard_in_tracing(self, results): assert len(shard_set) == 1 assert 'locally' in "\n".join([event.description for event in events]) - trace_id = results.response_future.get_query_trace_ids()[0] + # See verify_hosts_in_tracing: use the last trace id, not the first. + trace_id = results.response_future.get_query_trace_ids()[-1] traces = self.session.execute("SELECT * FROM system_traces.events WHERE session_id = %s", (trace_id,)) events = [event for event in traces] shard_set = set() From 3f6d5ba6acf89cd6da5e308c0d13478a6c5a37cb Mon Sep 17 00:00:00 2001 From: Yaniv Kaul Date: Sun, 23 Aug 2026 19:05:04 +0300 Subject: [PATCH 3/3] Add a post-connect warm-up window for prepare_on_all_hosts Broadcasting a PREPARE to every pooled host on every prepare() call is mainly useful right after a Session connects, when hosts are freshly discovered and different callers are likely to hit different hosts in quick succession. In steady state, traffic for a given prepared statement usually concentrates on a stable subset of replicas via token-aware routing, so the broadcast is normally wasted work, and an UNPREPARED response already triggers reprepare-and-retry on demand. Leave prepare_on_all_hosts unset (Cluster's new sentinel default) to get eager broadcast only during prepare_on_all_hosts_warmup_seconds (15s) after Session connects, then fall back to the lazy behavior. Explicitly passing True or False still pins that behavior for the life of the cluster, unaffected by the warm-up window. This is independent of Cluster._prepare_all_queries/reprepare_on_up, which already handles hosts that join after a statement was prepared. Signed-off-by: Yaniv Michael Kaul --- CHANGELOG.rst | 15 +- cassandra/cluster.py | 93 ++++++++++-- tests/integration/standard/test_query.py | 31 ++++ tests/unit/test_cluster.py | 172 +++++++++++++++++++++++ 4 files changed, 295 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 993c510dcb..a3b57388b9 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -58,10 +58,17 @@ Others come through unchanged. Previously ``DRIVER_NAME`` and ``DRIVER_VERSION`` could be overridden, which misreported the driver to the server for the life of the connection and, in the clients table, to the operator reading the row. -* ``Cluster.prepare_on_all_hosts`` now defaults to ``False``. In multi-DC deployments eager - preparation previously ran on every pooled host, including remote hosts that are rarely or - never queried. Disabling it by default avoids that waste; an ``UNPREPARED`` response still - triggers on-demand reprepare and retry, so correctness is unaffected. +* ``Cluster.prepare_on_all_hosts`` now defaults to unset instead of ``True``. In multi-DC + deployments eager preparation previously ran on every pooled host, including remote hosts + that are rarely or never queried. Left unset, a ``Session`` now eagerly prepares on all + hosts only during a short warm-up window after it connects (``prepare_on_all_hosts_warmup_seconds``, + default 15s), when hosts have just been discovered and many different statements are likely + to hit many different hosts in quick succession; afterwards it falls back to the lazy + behavior (``prepare_on_all_hosts=False``), since steady-state traffic for a given prepared + statement usually concentrates on a stable subset of replicas via token-aware routing. + Passing ``prepare_on_all_hosts=True`` or ``False`` explicitly disables the warm-up and pins + the old, unconditional behavior for the life of the cluster. An ``UNPREPARED`` response + still triggers on-demand reprepare and retry, so correctness is unaffected either way. * ``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 43007f5f91..8c5aa8673d 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -985,16 +985,57 @@ def default_retry_policy(self, policy): establish connection pools. This can cause a rush of connections and queries if not mitigated with this factor. """ - prepare_on_all_hosts = False + _prepare_on_all_hosts = False + _prepare_on_all_hosts_explicit = False + + @property + def prepare_on_all_hosts(self): + """ + Specifies whether statements should be prepared on all hosts, or just one. + + When enabled, statements are eagerly prepared on every host with an open connection pool. In multi-DC + deployments this includes remote hosts that are rarely or never queried on the happy path; preparing on them + is purely a latency optimization, since an ``UNPREPARED`` response always triggers on-demand reprepare and + retry. It can be enabled on long-running applications with numerous clients preparing statements on startup, + where a randomized initial condition of the load balancing policy can be expected to distribute prepares from + different clients across the cluster. + + If left unset (the default), a :class:`.Session` instead applies :attr:`.prepare_on_all_hosts_warmup_seconds`: + it behaves as if this were ``True`` for a short warm-up window right after the session connects, then as if + ``False`` afterwards. Explicitly assigning ``True`` or ``False``, whether to the :class:`.Cluster` + constructor or to this attribute at any later point, disables the warm-up behavior and pins this to the + given value for the lifetime of the cluster. + """ + return self._prepare_on_all_hosts + + @prepare_on_all_hosts.setter + def prepare_on_all_hosts(self, value): + self._prepare_on_all_hosts = value + self._prepare_on_all_hosts_explicit = True + + prepare_on_all_hosts_warmup_seconds = 15 """ - Specifies whether statements should be prepared on all hosts, or just one. + Length, in seconds, of the warm-up window used to decide whether :meth:`.Session.prepare` eagerly prepares + on all pooled hosts, when :attr:`.prepare_on_all_hosts` was not explicitly set by the caller. + + Right after a :class:`.Session` connects, hosts have just been discovered and different callers/tests + typically prepare many different statements against many different hosts in quick succession; eagerly + broadcasting each prepare avoids a burst of ``UNPREPARED``/reprepare/retry round trips during that period. + In steady state, query traffic for a given prepared statement usually concentrates on a stable subset of + replicas (via token-aware routing), so broadcasting to every host is normally wasted work, and the driver + falls back to lazy on-demand reprepare (the same behavior as ``prepare_on_all_hosts=False``). - When enabled, statements are eagerly prepared on every host with an open connection pool. In multi-DC - deployments this includes remote hosts that are rarely or never queried on the happy path; preparing on them - is purely a latency optimization, since an ``UNPREPARED`` response always triggers on-demand reprepare and - retry. It can be enabled on long-running applications with numerous clients preparing statements on startup, - where a randomized initial condition of the load balancing policy can be expected to distribute prepares from - different clients across the cluster. + The window is measured from when the :class:`.Session` finished establishing its initial connection pools, + not from the first call to :meth:`.Session.prepare`. An application that waits well past connect before + ever calling ``prepare()`` (lazy-first-use) will not benefit from the warm-up window, since by then hosts + are no longer "freshly discovered" and the startup thundering-herd risk this is meant to mitigate has + already passed. + + Setting this to zero (or a falsy value) disables the warm-up behavior entirely, equivalent to leaving + :attr:`.prepare_on_all_hosts` at its unset default with no warm-up: statements are never eagerly broadcast + unless the flag is set explicitly. + + Has no effect when :attr:`.prepare_on_all_hosts` was explicitly set by the caller. """ reprepare_on_up = True @@ -1208,7 +1249,7 @@ def __init__(self, schema_metadata_page_size=1000, address_translator=None, status_event_refresh_window=2, - prepare_on_all_hosts=False, + prepare_on_all_hosts=_NOT_SET, reprepare_on_up=True, execution_profiles=None, allow_beta_protocol_version=False, @@ -1225,7 +1266,8 @@ def __init__(self, application_info:Optional[ApplicationInfoBase]=None, client_routes_config:Optional[ClientRoutesConfig]=None, allow_control_connection_query_fallback:Optional[ControlConnectionQueryFallback]=ControlConnectionQueryFallback.Disabled, - driver_config_reporting_enabled=True + driver_config_reporting_enabled=True, + prepare_on_all_hosts_warmup_seconds=15 ): """ ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as @@ -1494,7 +1536,12 @@ def __init__(self, self.topology_event_refresh_window = topology_event_refresh_window self.status_event_refresh_window = status_event_refresh_window self.connect_timeout = connect_timeout - self.prepare_on_all_hosts = prepare_on_all_hosts + if prepare_on_all_hosts is _NOT_SET: + self._prepare_on_all_hosts = False + self._prepare_on_all_hosts_explicit = False + else: + self.prepare_on_all_hosts = prepare_on_all_hosts + self.prepare_on_all_hosts_warmup_seconds = prepare_on_all_hosts_warmup_seconds self.reprepare_on_up = reprepare_on_up self.shard_aware_options = ShardAwareOptions(opts=shard_aware_options) @@ -1789,6 +1836,8 @@ def connect(self, keyspace=None, wait_for_all_pools=False): session = self._new_session(keyspace) if wait_for_all_pools: wait_futures(session._initial_connect_futures) + # reset so the warm-up window starts after all pools are up, not just the first + session._connect_time = time.time() self._set_default_dbaas_consistency(session) @@ -2655,6 +2704,9 @@ def __init__(self, cluster, hosts, keyspace=None): raise NoHostAvailable(msg, [h.address for h in hosts]) self.session_id = uuid.uuid4() + # marks when this session finished its initial pool setup; used to gauge whether we're + # still in the post-connect warm-up window for prepare_on_all_hosts (see _should_prepare_on_all_hosts) + self._connect_time = time.time() if self.cluster.column_encryption_policy is not None: try: @@ -3251,7 +3303,7 @@ def prepare(self, query, custom_payload=None, keyspace=None): self.cluster.add_prepared(response.query_id, prepared_statement) - if self.cluster.prepare_on_all_hosts: + if self._should_prepare_on_all_hosts(): host = future._current_host try: self.prepare_on_all_hosts(prepared_statement.query_string, host, prepared_keyspace) @@ -3260,6 +3312,23 @@ def prepare(self, query, custom_payload=None, keyspace=None): return prepared_statement + def _should_prepare_on_all_hosts(self): + """ + Decide whether this prepare() call should eagerly broadcast to all pooled hosts. + + If the user explicitly set Cluster.prepare_on_all_hosts, that choice always wins. Otherwise, act as + if it were True during the post-connect warm-up window (see prepare_on_all_hosts_warmup_seconds) and + False afterwards. + """ + cluster = self.cluster + if cluster._prepare_on_all_hosts_explicit: + return cluster.prepare_on_all_hosts + + warmup_seconds = cluster.prepare_on_all_hosts_warmup_seconds + if not warmup_seconds: + return False + return (time.time() - self._connect_time) <= warmup_seconds + def prepare_on_all_hosts(self, query, excluded_host, keyspace=None): """ Prepare the given query on all hosts, excluding ``excluded_host``. diff --git a/tests/integration/standard/test_query.py b/tests/integration/standard/test_query.py index 5f1d5bfc19..b342a4ceb4 100644 --- a/tests/integration/standard/test_query.py +++ b/tests/integration/standard/test_query.py @@ -527,6 +527,37 @@ def test_prepare_on_all_hosts(self): session.execute(select_statement, (1, ), host=host) assert 2 == mock_handler.get_message_count('debug', "Re-preparing") + def test_prepare_on_all_hosts_default_and_explicit_true(self): + """ + Regression test for the prepare_on_all_hosts default flip to False. + + test_prepare_on_all_hosts above pins prepare_on_all_hosts=False explicitly, so it + can't catch a regression in the class attribute or constructor default. Disable the + warm-up shim (warmup_seconds=0) so the unset default is exercised deterministically, + and also cover the explicit True opt-in to eager preparation. + """ + with MockLoggingHandler().set_module_name(cluster.__name__) as mock_handler: + clus = TestCluster(reprepare_on_up=False, prepare_on_all_hosts_warmup_seconds=0) + self.addCleanup(clus.shutdown) + assert clus.prepare_on_all_hosts is False + + session = clus.connect(wait_for_all_pools=True) + select_statement = session.prepare("SELECT k FROM test3rf.test WHERE k = ?") + for host in clus.metadata.all_hosts(): + session.execute(select_statement, (1, ), host=host) + assert 2 == mock_handler.get_message_count('debug', "Re-preparing") + + with MockLoggingHandler().set_module_name(cluster.__name__) as mock_handler: + clus = TestCluster(prepare_on_all_hosts=True, reprepare_on_up=False) + self.addCleanup(clus.shutdown) + assert clus.prepare_on_all_hosts is True + + session = clus.connect(wait_for_all_pools=True) + select_statement = session.prepare("SELECT k FROM test3rf.test WHERE k = ?") + for host in clus.metadata.all_hosts(): + session.execute(select_statement, (1, ), host=host) + assert 0 == mock_handler.get_message_count('debug', "Re-preparing") + def test_prepare_batch_statement(self): """ Test to validate a prepared statement used inside a batch statement is correctly handled diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 74ed346c68..c5a00bdb0b 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -14,8 +14,11 @@ import unittest from concurrent.futures import Future +import inspect +import itertools import logging import socket +import time from types import SimpleNamespace from unittest.mock import patch, Mock @@ -190,6 +193,41 @@ def test_port_range(self): with pytest.raises(ValueError): cluster = Cluster(contact_points=['127.0.0.1'], port=invalid_port) + def test_prepare_on_all_hosts_defaults_to_unset(self): + cluster = Cluster() + self.addCleanup(cluster.shutdown) + assert cluster.prepare_on_all_hosts is False + assert cluster._prepare_on_all_hosts_explicit is False + + explicit_cluster = Cluster(prepare_on_all_hosts=True) + self.addCleanup(explicit_cluster.shutdown) + assert explicit_cluster.prepare_on_all_hosts is True + assert explicit_cluster._prepare_on_all_hosts_explicit is True + + def test_reprepare_on_up_keeps_its_positional_slot(self): + # guards against prepare_on_all_hosts_warmup_seconds shifting reprepare_on_up's + # position and silently breaking callers that pass it positionally; the names + # below are the pre-existing params, in their pre-existing order, up to and + # including reprepare_on_up (see e.g. commit 3583fdcf's Cluster.__init__) + pre_existing_params_up_to_reprepare_on_up = [ + 'contact_points', 'port', 'compression', 'auth_provider', 'load_balancing_policy', + 'reconnection_policy', 'default_retry_policy', 'conviction_policy_factory', + 'metrics_enabled', 'connection_class', 'ssl_options', 'sockopts', 'cql_version', + 'protocol_version', 'executor_threads', 'max_schema_agreement_wait', + 'control_connection_timeout', 'idle_heartbeat_interval', 'schema_event_refresh_window', + 'topology_event_refresh_window', 'connect_timeout', 'schema_metadata_enabled', + 'token_metadata_enabled', 'schema_metadata_page_size', 'address_translator', + 'status_event_refresh_window', 'prepare_on_all_hosts', + ] + sig = inspect.signature(Cluster.__init__) + args = [sig.parameters[name].default for name in pre_existing_params_up_to_reprepare_on_up] + args.append(False) # positionally where reprepare_on_up used to be (and must still be) + cluster = Cluster(*args) + self.addCleanup(cluster.shutdown) + assert cluster.reprepare_on_up is False + assert cluster.prepare_on_all_hosts is False + assert cluster._prepare_on_all_hosts_explicit is False + def test_control_connection_query_fallback_modes(self): default_cluster = Cluster() self.addCleanup(default_cluster.shutdown) @@ -421,6 +459,140 @@ def test_connection_factory_ignores_a_caller_supplied_session_id_and_reporter(se assert factory.call_args.kwargs['driver_config_reporter'] is None +class PrepareOnAllHostsWarmupTest(unittest.TestCase): + """ + Covers the post-connect warm-up window that decides whether Session.prepare() + eagerly broadcasts to all pooled hosts when Cluster.prepare_on_all_hosts was + left unset. See Session._should_prepare_on_all_hosts. + """ + + def _make_session(self, **cluster_kwargs): + cluster = Cluster(**cluster_kwargs) + self.addCleanup(cluster.shutdown) + host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4()) + host.set_up() + cluster.metadata.add_or_return_host(host) + return Session(cluster, [host]) + + @mock_session_pools + def test_within_warmup_window_prepares_eagerly_by_default(self, *_): + session = self._make_session() + session._connect_time = time.time() + + assert session._should_prepare_on_all_hosts() is True + + @mock_session_pools + def test_after_warmup_window_falls_back_to_lazy_by_default(self, *_): + session = self._make_session() + session._connect_time = time.time() - session.cluster.prepare_on_all_hosts_warmup_seconds - 1 + + assert session._should_prepare_on_all_hosts() is False + + @mock_session_pools + def test_explicit_true_is_respected_even_after_warmup_elapses(self, *_): + session = self._make_session(prepare_on_all_hosts=True) + session._connect_time = time.time() - session.cluster.prepare_on_all_hosts_warmup_seconds - 1 + + assert session._should_prepare_on_all_hosts() is True + + @mock_session_pools + def test_explicit_false_is_respected_even_within_warmup_window(self, *_): + session = self._make_session(prepare_on_all_hosts=False) + session._connect_time = time.time() + + assert session._should_prepare_on_all_hosts() is False + + @mock_session_pools + def test_runtime_assignment_after_construction_is_respected(self, *_): + session = self._make_session() + session._connect_time = time.time() - session.cluster.prepare_on_all_hosts_warmup_seconds - 1 + + session.cluster.prepare_on_all_hosts = True + assert session._should_prepare_on_all_hosts() is True + + session._connect_time = time.time() + session.cluster.prepare_on_all_hosts = False + assert session._should_prepare_on_all_hosts() is False + + @mock_session_pools + def test_zero_warmup_seconds_disables_eager_behavior(self, *_): + session = self._make_session(prepare_on_all_hosts_warmup_seconds=0) + session._connect_time = time.time() + + assert session._should_prepare_on_all_hosts() is False + + @mock_session_pools + def test_prepare_uses_should_prepare_on_all_hosts_decision(self, *_): + session = self._make_session() + session._connect_time = time.time() + + message = Mock(query_id=b'qid', bind_metadata=[], pk_indexes=[], column_metadata=[], + result_metadata_id=None, is_lwt=False) + future = Mock() + future.result.return_value.one.return_value = message + future._current_host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4()) + + with patch('cassandra.cluster.ResponseFuture', return_value=future), \ + patch.object(session.cluster, 'add_prepared'), \ + patch.object(Session, 'prepare_on_all_hosts') as prepare_on_all_hosts: + session.prepare("SELECT * FROM t") + + assert prepare_on_all_hosts.call_count == 1 + + session._connect_time = time.time() - session.cluster.prepare_on_all_hosts_warmup_seconds - 1 + with patch('cassandra.cluster.ResponseFuture', return_value=future), \ + patch.object(session.cluster, 'add_prepared'), \ + patch.object(Session, 'prepare_on_all_hosts') as prepare_on_all_hosts: + session.prepare("SELECT * FROM t") + + assert prepare_on_all_hosts.call_count == 0 + + @mock_session_pools + def test_wait_for_all_pools_starts_warmup_window_after_slow_pool_connects(self, *_): + # Cluster.connect(wait_for_all_pools=True) must reset the warm-up window's start + # to after Session.__init__'s own wait (first pool up) finishes waiting for every + # initial pool, not leave it at Session.__init__'s earlier timestamp. + cluster = Cluster() + self.addCleanup(cluster.shutdown) + host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4()) + host.set_up() + cluster.metadata.add_or_return_host(host) + + # A monotonically increasing counter, not a fixed-size list: DEBUG-level log + # records also call time.time() on some Python versions, so the exact number + # of calls before connect()'s reset can't be pinned down to a fixed count. + timestamps = itertools.count(100.0, 100.0) + with patch.object(cluster.control_connection, 'connect'), \ + patch.object(cluster, '_populate_hosts'), \ + patch.object(cluster.profile_manager, 'check_supported'), \ + patch('cassandra.cluster.time.time', side_effect=lambda: next(timestamps)): + session = cluster.connect(wait_for_all_pools=True) + + # connect()'s reset must be later than the first timestamp ever handed out + # (Session.__init__'s own stamp), proving it didn't just keep that early value. + assert session._connect_time > 100.0 + + @mock_session_pools + def test_prepare_all_queries_on_host_up_is_unaffected_by_flag_or_warmup(self, *_): + # Cluster._prepare_all_queries (the reprepare_on_up path for late-joining hosts) + # is a separate mechanism from prepare_on_all_hosts/warmup and must keep firing + # regardless of either. + session = self._make_session(prepare_on_all_hosts=False, prepare_on_all_hosts_warmup_seconds=0) + session._connect_time = time.time() - 1000 + cluster = session.cluster + + prepared_statement = Mock(query_string="SELECT * FROM t", keyspace=None) + cluster._prepared_statements = {b'qid': prepared_statement} + + new_host = Host("127.0.0.2", SimpleConvictionPolicy, host_id=uuid.uuid4()) + new_host.set_up() + + with patch.object(cluster, 'connection_factory') as connection_factory: + cluster._prepare_all_queries(new_host) + + assert connection_factory.call_count == 1 + + class SchedulerTest(unittest.TestCase): # TODO: this suite could be expanded; for now just adding a test covering a ticket