fix: use reactor-aware executor factory for control-connection paging, fix mock contract and lint issue

- ControlConnection._try_connect and _refresh_node_list_and_token_map now
  build their paging ThreadPoolExecutor via Cluster._create_thread_pool_executor
  instead of instantiating concurrent.futures.ThreadPoolExecutor directly.
  The bare executor bypassed the Eventlet detection that factory exists for,
  reintroducing the known Python 3.7+/Eventlet hang for users configured
  with EventletConnection. GreenThreadPoolExecutor (futurist) subclasses
  concurrent.futures._base.Executor, so it supports the same `with ... as
  executor:` context-manager protocol, no restructuring needed.
- tests/unit/test_control_connection.py: MockConnection.wait_for_response
  now honors the fail_on_error=False contract, returning (True, result)
  instead of a bare result, matching cassandra/connection.py's
  Connection.fetch_all_pages/wait_for_response behavior. Added a regression
  test covering this. Also added MockCluster._create_thread_pool_executor
  so tests exercising the fixed call sites keep working.
- Renamed a loop variable that shadowed the `call` import from
  unittest.mock (Ruff F402) in test_topology_queries_use_paging.

Author: mykaul

cassandra/cluster.py

@@ -3935,7 +3935,7 @@ def _try_connect(self, endpoint):
                                        consistency_level=ConsistencyLevel.ONE,
                                        fetch_size=self._schema_meta_page_size)
 
-            with ThreadPoolExecutor(max_workers=2) as executor:
+            with self._cluster._create_thread_pool_executor(max_workers=2) as executor:
                 local_future = executor.submit(
                     connection.fetch_all_pages, local_query, self._timeout, False)
                 peers_future = executor.submit(
@@ -4103,7 +4103,7 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
                                        consistency_level=cl,
                                        fetch_size=self._schema_meta_page_size)
 
-            with ThreadPoolExecutor(max_workers=2) as executor:
+            with self._cluster._create_thread_pool_executor(max_workers=2) as executor:
                 peers_future = executor.submit(
                     connection.fetch_all_pages, peers_query, self._timeout)
                 local_future = executor.submit(

tests/unit/test_control_connection.py

@@ -121,6 +121,9 @@ def add_host(self, endpoint, datacenter, rack, signal=False, refresh_nodes=True,
     def remove_host(self, host):
         pass
 
+    def _create_thread_pool_executor(self, **kwargs):
+        return ThreadPoolExecutor(**kwargs)
+
     def on_up(self, host):
         pass
 
@@ -177,7 +180,7 @@ def wait_for_response_side_effect(query_msg, timeout=None, fail_on_error=True):
                 result.column_names = self.local_results[0]
                 result.parsed_rows = self.local_results[1]
             result.paging_state = None
-            return result
+            return result if fail_on_error else (True, result)
         self.wait_for_response = Mock(side_effect=wait_for_response_side_effect)
 
     def fetch_all_pages(self, query_msg, timeout, fail_on_error=True):
@@ -331,11 +334,25 @@ def test_topology_queries_use_paging(self):
         self.control_connection.refresh_node_list_and_token_map()
         assert self.connection.wait_for_response.called
         calls = self.connection.wait_for_response.call_args_list
-        for call in calls:
-            query_msg = call[0][0]
+        for response_call in calls:
+            query_msg = response_call[0][0]
             assert isinstance(query_msg, QueryMessage)
             assert query_msg.fetch_size == self.control_connection._schema_meta_page_size
 
+    def test_mock_connection_fetch_all_pages_honors_fail_on_error_false(self):
+        """
+        Regression test for MockConnection.fetch_all_pages/wait_for_response:
+        when fail_on_error=False, it must return (True, result), matching
+        the real Connection.fetch_all_pages/wait_for_response contract used
+        by ControlConnection._try_connect (which passes fail_on_error=False
+        and unpacks the result as a (success, result) tuple).
+        """
+        query_msg = QueryMessage(query="SELECT * FROM system.local",
+                                  consistency_level=ConsistencyLevel.ONE)
+        success, result = self.connection.fetch_all_pages(query_msg, timeout=5, fail_on_error=False)
+        assert success is True
+        assert isinstance(result, ResultMessage)
+
     def test_topology_queries_fetch_all_pages(self):
         from cassandra.connection import Connection as RealConnection
         mock_connection = MagicMock()