control-connection: deprecate ControlConnection.wait_for_schema_agreement

Keep ControlConnection.wait_for_schema_agreement() as a compatibility wrapper, but move the existing implementation to _wait_for_schema_agreement() and deprecate the public method in favor of Session.wait_for_schema_agreement(). This lets the control-connection refresh path continue using the old logic internally without emitting warnings.

The control-connection wait path was designed for internal metadata refresh use, not as a user-facing schema agreement API. It observes schema agreement from the control connection's perspective and can report disagreement for hosts the session is not actually using, or fail because the control connection itself is transiently unhealthy. When called directly by user code that can produce false positives or failures that do not reflect whether the session can safely proceed.

Update the unit tests to call the internal helper everywhere a warning is not expected, add explicit deprecation coverage for the public wrapper, and set stacklevel=2 so the warning points at the caller instead of inside the driver.

Author: dkropachev

cassandra/cluster.py

@@ -3905,9 +3905,9 @@ def _refresh_schema(self, connection, preloaded_results=None, schema_agreement_w
         if self._cluster.is_shutdown:
             return False
 
-        agreed = self.wait_for_schema_agreement(connection,
-                                                preloaded_results=preloaded_results,
-                                                wait_time=schema_agreement_wait)
+        agreed = self._wait_for_schema_agreement(connection,
+                                                 preloaded_results=preloaded_results,
+                                                 wait_time=schema_agreement_wait)
 
         if not self._schema_meta_enabled and not force:
             log.debug("[control connection] Skipping schema refresh because schema metadata is disabled")
@@ -4198,6 +4198,14 @@ def _handle_schema_change(self, event):
         self._cluster.scheduler.schedule_unique(delay, self.refresh_schema, **event)
 
     def wait_for_schema_agreement(self, connection=None, preloaded_results=None, wait_time=None):
+        warn("ControlConnection.wait_for_schema_agreement is deprecated and will be removed in 4.0. "
+             "Use Session.wait_for_schema_agreement instead.", DeprecationWarning, stacklevel=2)
+        return self._wait_for_schema_agreement(connection=connection,
+                                               preloaded_results=preloaded_results,
+                                               wait_time=wait_time)
+
+    def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wait_time=None):
+
         total_timeout = wait_time if wait_time is not None else self._cluster.max_schema_agreement_wait
         if total_timeout <= 0:
             return True

tests/unit/test_control_connection.py

@@ -15,7 +15,7 @@
 import unittest
 
 from concurrent.futures import ThreadPoolExecutor
-from unittest.mock import Mock, ANY, call
+from unittest.mock import Mock, ANY, call, patch
 
 from cassandra import OperationTimedOut, SchemaTargetType, SchemaChangeType
 from cassandra.protocol import ResultMessage, RESULT_KIND_ROWS
@@ -210,16 +210,26 @@ def test_wait_for_schema_agreement(self):
         """
         Basic test with all schema versions agreeing
         """
-        assert self.control_connection.wait_for_schema_agreement()
+        assert self.control_connection._wait_for_schema_agreement()
         # the control connection should not have slept at all
         assert self.time.clock == 0
 
+    @patch('cassandra.cluster.warn')
+    def test_wait_for_schema_agreement_warns_about_deprecation(self, mocked_warn):
+        assert self.control_connection.wait_for_schema_agreement()
+
+        mocked_warn.assert_called_once()
+        warning_args, warning_kwargs = mocked_warn.call_args
+        assert 'ControlConnection.wait_for_schema_agreement is deprecated' in str(warning_args[0])
+        assert warning_args[1] is DeprecationWarning
+        assert warning_kwargs['stacklevel'] == 2
+
     def test_wait_for_schema_agreement_uses_preloaded_results_if_given(self):
         """
         wait_for_schema_agreement uses preloaded results if given for shared table queries
         """
         preloaded_results = self._matching_schema_preloaded_results
-        assert self.control_connection.wait_for_schema_agreement(preloaded_results=preloaded_results)
+        assert self.control_connection._wait_for_schema_agreement(preloaded_results=preloaded_results)
         # the control connection should not have slept at all
         assert self.time.clock == 0
         # the connection should not have made any queries if given preloaded results
@@ -230,7 +240,7 @@ def test_wait_for_schema_agreement_falls_back_to_querying_if_schemas_dont_match_
         wait_for_schema_agreement requery if schema does not match using preloaded results
         """
         preloaded_results = self._nonmatching_schema_preloaded_results
-        assert self.control_connection.wait_for_schema_agreement(preloaded_results=preloaded_results)
+        assert self.control_connection._wait_for_schema_agreement(preloaded_results=preloaded_results)
         # the control connection should not have slept at all
         assert self.time.clock == 0
         assert self.connection.wait_for_responses.call_count == 1
@@ -241,7 +251,7 @@ def test_wait_for_schema_agreement_fails(self):
         """
         # change the schema version on one node
         self.connection.peer_results[1][1][2] = 'b'
-        assert not self.control_connection.wait_for_schema_agreement()
+        assert not self.control_connection._wait_for_schema_agreement()
         # the control connection should have slept until it hit the limit
         assert self.time.clock >= self.cluster.max_schema_agreement_wait
 
@@ -262,7 +272,7 @@ def test_wait_for_schema_agreement_skipping(self):
         self.connection.peer_results[1][1][3] = 'c'
         self.cluster.metadata.get_host(DefaultEndPoint('192.168.1.1')).is_up = False
 
-        assert self.control_connection.wait_for_schema_agreement()
+        assert self.control_connection._wait_for_schema_agreement()
         assert self.time.clock == 0
 
     def test_wait_for_schema_agreement_rpc_lookup(self):
@@ -279,12 +289,12 @@ def test_wait_for_schema_agreement_rpc_lookup(self):
 
         # even though the new host has a different schema version, it's
         # marked as down, so the control connection shouldn't care
-        assert self.control_connection.wait_for_schema_agreement()
+        assert self.control_connection._wait_for_schema_agreement()
         assert self.time.clock == 0
 
         # but once we mark it up, the control connection will care
         host.is_up = True
-        assert not self.control_connection.wait_for_schema_agreement()
+        assert not self.control_connection._wait_for_schema_agreement()
         assert self.time.clock >= self.cluster.max_schema_agreement_wait
 
 
@@ -299,7 +309,7 @@ def test_wait_for_schema_agreement_none_timeout(self):
                                status_event_refresh_window=0)
         cc._connection = self.connection
         cc._time = self.time
-        assert cc.wait_for_schema_agreement()
+        assert cc._wait_for_schema_agreement()
 
     def test_refresh_nodes_and_tokens(self):
         self.control_connection.refresh_node_list_and_token_map()