schema-agreement: isolate session queries from shared executor

Session.wait_for_schema_agreement() originally dispatched its per-host probes through Session.submit(), which routes onto the cluster-wide executor shared with pool creation and metadata work. That executor is small by default, so unrelated queued tasks can delay schema-agreement probes long enough to produce false timeouts or make the wait look hung before the queries even start.

Create a short-lived executor sized to the schema-agreement parallelism, build it through Cluster._create_thread_pool_executor() so the existing reactor-specific executor selection still applies, and shut it down after the check completes. Update the unit tests to inject the dedicated executor and add coverage proving schema agreement still succeeds when the shared executor is blocked.

Author: dkropachev

cassandra/cluster.py

@@ -3441,51 +3441,56 @@ def _get_schema_mismatches_for_scope(self, deadline, scope: Literal['rack', 'dc'
             consistency_level=cl)
 
         parallelism = max(1, min(parallelism, len(hosts)))
-        for offset in range(0, len(hosts), parallelism):
-            remaining = deadline - time.time()
-            if remaining <= 0:
-                for host in hosts[offset:]:
-                    errors[host.endpoint] = "Timed out before querying host"
-                break
-
-            futures = {}
-            for host in hosts[offset:offset + parallelism]:
-                future = self.submit(self._query_local_schema_version, host, query, deadline)
-                if future is None:
-                    errors[host.endpoint] = "Schema agreement executor unavailable"
+        # Use a dedicated executor so schema agreement does not get blocked by
+        # unrelated work already queued on the shared session executor.
+        executor = self.cluster._create_thread_pool_executor(max_workers=parallelism)
+        try:
+            for offset in range(0, len(hosts), parallelism):
+                remaining = deadline - time.time()
+                if remaining <= 0:
+                    for host in hosts[offset:]:
+                        errors[host.endpoint] = "Timed out before querying host"
+                    break
+
+                futures = {}
+                for host in hosts[offset:offset + parallelism]:
+                    futures[executor.submit(self._query_local_schema_version, host, query, deadline)] = host
+
+                if not futures:
                     continue
-                futures[future] = host
 
-            if not futures:
-                continue
-
-            remaining = deadline - time.time()
-            if remaining <= 0:
-                for future, host in futures.items():
+                remaining = deadline - time.time()
+                if remaining <= 0:
+                    for future, host in futures.items():
+                        future.cancel()
+                        errors[host.endpoint] = "Timed out before querying host"
+                    for host in hosts[offset + parallelism:]:
+                        errors[host.endpoint] = "Timed out before querying host"
+                    break
+
+                done, not_done = wait_futures(tuple(futures), timeout=remaining)
+                for future in not_done:
                     future.cancel()
+                    host = futures[future]
                     errors[host.endpoint] = "Timed out before querying host"
-                for host in hosts[offset + parallelism:]:
-                    errors[host.endpoint] = "Timed out before querying host"
-                break
-
-            done, not_done = wait_futures(tuple(futures), timeout=remaining)
-            for future in not_done:
-                future.cancel()
-                host = futures[future]
-                errors[host.endpoint] = "Timed out before querying host"
-
-            for future in done:
-                host = futures[future]
-                schema_version, error = future.result()
-                if error is not None:
-                    errors[host.endpoint] = error
-                    continue
-                versions[schema_version].add(host.endpoint)
 
-            if not_done:
-                for host in hosts[offset + parallelism:]:
-                    errors[host.endpoint] = "Timed out before querying host"
-                break
+                for future in done:
+                    host = futures[future]
+                    schema_version, error = future.result()
+                    if error is not None:
+                        errors[host.endpoint] = error
+                        continue
+                    versions[schema_version].add(host.endpoint)
+
+                if not_done:
+                    for host in hosts[offset + parallelism:]:
+                        errors[host.endpoint] = "Timed out before querying host"
+                    break
+        finally:
+            try:
+                executor.shutdown(wait=False, cancel_futures=True)
+            except TypeError:
+                executor.shutdown(wait=False)
 
         if len(versions) == 1 and None not in versions and not errors:
             log.debug("[session] Local schemas match")

tests/unit/test_cluster.py

@@ -450,16 +450,20 @@ def test_wait_for_schema_agreement_limits_parallel_queries_to_default(self, mock
         session, _ = self._new_schema_agreement_session(["a"] * 11)
         batch_sizes = []
 
-        def submit(fn, host, query, deadline):
-            future = Future()
-            future.set_result(fn(host, query, deadline))
-            return future
-
         def fake_wait(futures, timeout=None, return_when=None):
             batch_sizes.append(len(futures))
             return set(futures), set()
 
-        session.submit = Mock(side_effect=submit)
+        class ImmediateExecutor(object):
+            def submit(self, fn, *args, **kwargs):
+                future = Future()
+                future.set_result(fn(*args, **kwargs))
+                return future
+
+            def shutdown(self, wait=False, cancel_futures=False):
+                return None
+
+        session.cluster._create_thread_pool_executor = Mock(return_value=ImmediateExecutor())
         mocked_wait_futures.side_effect = fake_wait
 
         assert session.wait_for_schema_agreement(wait_time=1)

tests/unit/test_session_schema_agreement.py

@@ -1,8 +1,9 @@
-from concurrent.futures import Future
+from concurrent.futures import Future, ThreadPoolExecutor
 from datetime import timedelta
 from types import SimpleNamespace
 from unittest.mock import Mock
 import uuid
+from threading import Event
 
 import pytest
 
@@ -33,6 +34,19 @@ def _get_connection_for_routing_key(self):
         return self.connection
 
 
+class ImmediateExecutor:
+    def submit(self, fn, *args, **kwargs):
+        future = Future()
+        try:
+            future.set_result(fn(*args, **kwargs))
+        except Exception as exc:
+            future.set_exception(exc)
+        return future
+
+    def shutdown(self, wait=False, cancel_futures=False):
+        return None
+
+
 def _mock_schema_response(schema_version):
     return SimpleNamespace(column_names=["schema_version"], parsed_rows=[[schema_version]])
 
@@ -53,28 +67,20 @@ def _new_session(schema_versions, distances=None, metadata_request_timeout=timed
 
     cluster = SimpleNamespace(
         max_schema_agreement_wait=10,
+        executor=ImmediateExecutor(),
         control_connection=SimpleNamespace(
             _timeout=timeout,
             _metadata_request_timeout=metadata_request_timeout,
         ),
     )
+    cluster._create_thread_pool_executor = Mock(return_value=ImmediateExecutor())
 
     session = Session.__new__(Session)
     session.cluster = cluster
     session._profile_manager = SimpleNamespace(distance=lambda host: distance_map.get(host, HostDistance.LOCAL))
     session._pools = {}
     session.is_shutdown = False
 
-    def submit(fn, host, query, deadline):
-        future = Future()
-        try:
-            future.set_result(fn(host, query, deadline))
-        except Exception as exc:
-            future.set_exception(exc)
-        return future
-
-    session.submit = submit
-
     for host, schema_version in zip(hosts, schema_versions):
         connection = Mock(endpoint=host.endpoint)
         connection.wait_for_response.return_value = _mock_schema_response(schema_version)
@@ -115,3 +121,18 @@ def test_wait_for_schema_agreement_bypasses_non_positive_wait_time(wait_time, cl
 
     for connection in connections.values():
         assert connection.wait_for_response.call_count == 0
+
+
+def test_wait_for_schema_agreement_is_not_blocked_by_shared_executor():
+    session, _, _ = _new_session(["a", "a"])
+    busy_executor = ThreadPoolExecutor(max_workers=1)
+    block = Event()
+
+    session.cluster.executor = busy_executor
+    busy_executor.submit(block.wait)
+
+    try:
+        assert session.wait_for_schema_agreement(wait_time=0.1)
+    finally:
+        block.set()
+        busy_executor.shutdown(wait=True)