Add proper paging support for ControlConnection topology queries

- Implement _fetch_all_pages helper method to handle multi-page results
- Apply paging to all topology queries that use fetch_size
- Ensures all pages are fetched for large clusters (>1000 nodes)
- Add test to verify multi-page fetching works correctly

This addresses the review feedback that wait_for_responses doesn't
handle paging_state, so we need to manually fetch all pages.

Co-authored-by: mykaul <4655593+mykaul@users.noreply.github.com>

Author: Copilot

cassandra/cluster.py

@@ -3659,6 +3659,10 @@ def _try_connect(self, host):
                 peers_result = connection.wait_for_response(
                     peers_query, timeout=self._timeout)
 
+            # Fetch all pages if there are more results
+            peers_result = self._fetch_all_pages(connection, peers_result, peers_query, self._timeout)
+            local_result = self._fetch_all_pages(connection, local_result, local_query, self._timeout)
+
             shared_results = (peers_result, local_result)
             self._refresh_node_list_and_token_map(connection, preloaded_results=shared_results)
             self._refresh_schema(connection, preloaded_results=shared_results, schema_agreement_wait=-1)
@@ -3719,6 +3723,26 @@ def _submit(self, *args, **kwargs):
             pass
         return None
 
+    def _fetch_all_pages(self, connection, result, query_msg, timeout):
+        """
+        Fetch all pages for a paged query result.
+        Returns the result with all parsed_rows combined from all pages.
+        """
+        if not result.paging_state:
+            return result
+        
+        all_rows = list(result.parsed_rows)
+        
+        while result.paging_state:
+            query_msg.paging_state = result.paging_state
+            result = connection.wait_for_response(query_msg, timeout=timeout)
+            if result.parsed_rows:
+                all_rows.extend(result.parsed_rows)
+        
+        # Update the result with all rows
+        result.parsed_rows = all_rows
+        return result
+
     def shutdown(self):
         # stop trying to reconnect (if we are)
         with self._reconnection_lock:
@@ -3807,6 +3831,10 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
                                        fetch_size=self._schema_meta_page_size)
             peers_result, local_result = connection.wait_for_responses(
                 peers_query, local_query, timeout=self._timeout)
+            
+            # Fetch all pages if there are more results
+            peers_result = self._fetch_all_pages(connection, peers_result, peers_query, self._timeout)
+            local_result = self._fetch_all_pages(connection, local_result, local_query, self._timeout)
 
         peers_result = dict_factory(peers_result.column_names, peers_result.parsed_rows)
 
@@ -3866,6 +3894,9 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
                         success, local_rpc_address_result = connection.wait_for_response(
                             local_rpc_address_query, timeout=self._timeout, fail_on_error=False)
                         if success:
+                            # Fetch all pages if there are more results (though system.local has only 1 row)
+                            local_rpc_address_result = self._fetch_all_pages(connection, local_rpc_address_result, 
+                                                                             local_rpc_address_query, self._timeout)
                             row = dict_factory(
                                 local_rpc_address_result.column_names,
                                 local_rpc_address_result.parsed_rows)
@@ -4107,6 +4138,10 @@ def wait_for_schema_agreement(self, connection=None, preloaded_results=None, wai
                     timeout = min(self._timeout, total_timeout - elapsed)
                     peers_result, local_result = connection.wait_for_responses(
                         peers_query, local_query, timeout=timeout)
+                    
+                    # Fetch all pages if there are more results
+                    peers_result = self._fetch_all_pages(connection, peers_result, peers_query, timeout)
+                    local_result = self._fetch_all_pages(connection, local_result, local_query, timeout)
                 except OperationTimedOut as timeout:
                     log.debug("[control connection] Timed out waiting for "
                               "response during schema agreement check: %s", timeout)

tests/unit/test_control_connection.py

@@ -17,7 +17,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from unittest.mock import Mock, ANY, call
 
-from cassandra import OperationTimedOut, SchemaTargetType, SchemaChangeType
+from cassandra import OperationTimedOut, SchemaTargetType, SchemaChangeType, ConsistencyLevel
 from cassandra.protocol import ResultMessage, RESULT_KIND_ROWS, QueryMessage
 from cassandra.cluster import ControlConnection, _Scheduler, ProfileManager, EXEC_PROFILE_DEFAULT, ExecutionProfile
 from cassandra.pool import Host
@@ -323,6 +323,54 @@ def test_topology_queries_use_paging(self):
             assert isinstance(query_msg, QueryMessage)
             assert query_msg.fetch_size == self.control_connection._schema_meta_page_size
 
+    def test_topology_queries_fetch_all_pages(self):
+        """
+        Test that topology queries fetch all pages when results are paged
+        """
+        from unittest.mock import MagicMock
+        
+        # Create mock connection
+        mock_connection = MagicMock()
+        mock_connection.endpoint = DefaultEndPoint("192.168.1.0")
+        mock_connection.original_endpoint = mock_connection.endpoint
+        
+        # Create first page of peers results with paging_state
+        first_page = ResultMessage(kind=RESULT_KIND_ROWS)
+        first_page.column_names = ["rpc_address", "peer", "schema_version", "data_center", "rack", "tokens", "host_id"]
+        first_page.parsed_rows = [["192.168.1.1", "10.0.0.1", "a", "dc1", "rack1", ["1", "101", "201"], "uuid2"]]
+        first_page.paging_state = b"has_more_pages"
+        
+        # Create second page of peers results without paging_state
+        second_page = ResultMessage(kind=RESULT_KIND_ROWS)
+        second_page.column_names = ["rpc_address", "peer", "schema_version", "data_center", "rack", "tokens", "host_id"]
+        second_page.parsed_rows = [["192.168.1.2", "10.0.0.2", "a", "dc1", "rack1", ["2", "102", "202"], "uuid3"]]
+        second_page.paging_state = None
+        
+        # Create local result without paging
+        local_result = ResultMessage(kind=RESULT_KIND_ROWS)
+        local_result.column_names = ["rpc_address", "schema_version", "cluster_name", "data_center", "rack", "partitioner", "release_version", "tokens", "host_id"]
+        local_result.parsed_rows = [["192.168.1.0", "a", "foocluster", "dc1", "rack1", "Murmur3Partitioner", "2.2.0", ["0", "100", "200"], "uuid1"]]
+        local_result.paging_state = None
+        
+        # Setup mock: first call returns first page, second call returns second page
+        mock_connection.wait_for_responses.return_value = (first_page, local_result)
+        mock_connection.wait_for_response.return_value = second_page
+        
+        # Test _fetch_all_pages
+        self.control_connection._connection = mock_connection
+        query_msg = QueryMessage(query="SELECT * FROM system.peers", consistency_level=ConsistencyLevel.ONE, fetch_size=1000)
+        
+        result = self.control_connection._fetch_all_pages(mock_connection, first_page, query_msg, timeout=5)
+        
+        # Verify that both pages were fetched
+        assert len(result.parsed_rows) == 2
+        assert result.parsed_rows[0][0] == "192.168.1.1"
+        assert result.parsed_rows[1][0] == "192.168.1.2"
+        assert result.paging_state is None
+        
+        # Verify wait_for_response was called once to fetch the second page
+        assert mock_connection.wait_for_response.called
+
     def test_refresh_nodes_and_tokens_with_invalid_peers(self):
         def refresh_and_validate_added_hosts():
             self.connection.wait_for_responses = Mock(return_value=_node_meta_results(