From d55dca8d81e6506b0bb37cb6c1bb0dd6c6a939af Mon Sep 17 00:00:00 2001
From: Dana Powers <dana.powers@gmail.com>
Date: Tue, 28 Jul 2026 08:57:04 -0700
Subject: [PATCH 1/7] Migrate manager.run -> net.run

---
 kafka/admin/_acls.py                      |   6 +-
 kafka/admin/_cluster.py                   |  18 ++--
 kafka/admin/_configs.py                   |  12 ++-
 kafka/admin/_groups.py                    |  18 ++--
 kafka/admin/_partitions.py                |  19 ++--
 kafka/admin/_topics.py                    |   8 +-
 kafka/admin/_transactions.py              |  10 +-
 kafka/admin/_users.py                     |   4 +-
 kafka/admin/client.py                     |   2 +-
 kafka/consumer/fetcher.py                 |  10 +-
 test/admin/test_admin_concurrent.py       |   8 +-
 test/admin/test_admin_groups.py           |   4 +-
 test/consumer/test_coordinator.py         | 113 +++++++++++-----------
 test/consumer/test_fetcher.py             |  56 +++++------
 test/consumer/test_fetcher_mock_broker.py |  14 +--
 test/integration/test_sasl_integration.py |   2 +-
 test/producer/test_sender.py              |   4 +-
 17 files changed, 155 insertions(+), 153 deletions(-)

diff --git a/kafka/admin/_acls.py b/kafka/admin/_acls.py
index 3b3da0c16..849ea1b7e 100644
--- a/kafka/admin/_acls.py
+++ b/kafka/admin/_acls.py
@@ -55,7 +55,7 @@ def describe_acls(self, acl_filter):
             operation=acl_filter.operation,
             permission_type=acl_filter.permission_type
         )
-        response = self._manager.run(self._manager.send, request)
+        response = self._net.run(self._manager.send, request)
         return self._convert_describe_acls_response_to_acls(response)
 
     @staticmethod
@@ -132,7 +132,7 @@ def create_acls(self, acls):
         creations = [self._convert_create_acls_resource_request(acl) for acl in acls]
         min_version = 3 if any(creation.resource_type == ResourceType.USER for creation in creations) else 0
         request = CreateAclsRequest(creations=creations, min_version=min_version)
-        response = self._manager.run(self._manager.send, request)
+        response = self._net.run(self._manager.send, request)
         return self._convert_create_acls_response_to_acls(acls, response)
 
     @staticmethod
@@ -191,7 +191,7 @@ def delete_acls(self, acl_filters):
         filters = [self._convert_delete_acls_resource_request(acl) for acl in acl_filters]
         min_version = 3 if any(_filter.resource_type_filter == ResourceType.USER for _filter in filters) else 0
         request = DeleteAclsRequest(filters=filters, min_version=min_version)
-        response = self._manager.run(self._manager.send, request)
+        response = self._net.run(self._manager.send, request)
         return self._convert_delete_acls_response_to_matching_acls(acl_filters, response)
 
 
diff --git a/kafka/admin/_cluster.py b/kafka/admin/_cluster.py
index 7183228b6..c54bf93fa 100644
--- a/kafka/admin/_cluster.py
+++ b/kafka/admin/_cluster.py
@@ -91,7 +91,7 @@ def describe_cluster(self):
         Returns:
             A dict with cluster-wide metadata, excluding topic details.
         """
-        return self._manager.run(self._describe_cluster)
+        return self._net.run(self._describe_cluster)
 
     async def _async_describe_log_dirs(self, topic_partitions=(), brokers=None):
         request = DescribeLogDirsRequest(topics=topic_partitions)
@@ -119,7 +119,7 @@ def describe_log_dirs(self, topic_partitions=None, brokers=None):
             list of dicts, containing per-broker log-dir data
         """
         topic_partitions = self._get_topic_partitions(topic_partitions)
-        return self._manager.run(self._async_describe_log_dirs, topic_partitions, brokers)
+        return self._net.run(self._async_describe_log_dirs, topic_partitions, brokers)
 
     @staticmethod
     def _alter_replica_log_dirs_requests(replica_assignments):
@@ -176,7 +176,7 @@ def alter_replica_log_dirs(self, replica_assignments):
             dict mapping :class:`~kafka.TopicPartitionReplica` to the
             corresponding error class (``kafka.errors.NoError`` on success).
         """
-        return self._manager.run(self._async_alter_replica_log_dirs, replica_assignments)
+        return self._net.run(self._async_alter_replica_log_dirs, replica_assignments)
 
     async def _async_describe_quorum(self, topic, partition):
         _Topic = DescribeQuorumRequest.TopicData
@@ -213,7 +213,7 @@ def describe_metadata_quorum(self):
         Returns:
             dict matching the DescribeQuorumResponse shape.
         """
-        return self._manager.run(self._async_describe_quorum, '__cluster_metadata', 0)
+        return self._net.run(self._async_describe_quorum, '__cluster_metadata', 0)
 
     async def _async_get_broker_version_data(self, broker_id):
         conn = await self._manager.get_connection(broker_id)
@@ -221,7 +221,7 @@ async def _async_get_broker_version_data(self, broker_id):
 
     def get_broker_version_data(self, broker_id):
         """Return BrokerVersionData for a specific broker"""
-        return self._manager.run(self._async_get_broker_version_data, broker_id)
+        return self._net.run(self._async_get_broker_version_data, broker_id)
 
     def api_versions(self):
         api_versions = self._manager.broker_version_data.api_versions
@@ -272,7 +272,7 @@ def describe_features(self, send_request_to_controller=False):
                 - ``finalized_features_epoch``: int, or None if unknown
                   (broker did not report an epoch, or reported -1)
         """
-        return self._manager.run(self._async_describe_features, send_request_to_controller)
+        return self._net.run(self._async_describe_features, send_request_to_controller)
 
     @staticmethod
     def _build_feature_updates(feature_updates):
@@ -348,9 +348,9 @@ def update_features(self, feature_updates, validate_only=False, timeout_ms=60000
         Returns:
             dict of {feature_name: 'OK' | error message}
         """
-        return self._manager.run(self._async_update_features,
-                                 feature_updates, validate_only, timeout_ms,
-                                 timeout_ms=timeout_ms)
+        return self._net.run(
+            self._async_update_features, feature_updates, validate_only, timeout_ms,
+            timeout_ms=timeout_ms)
 
 
 class UpdateFeatureType(EnumHelper, IntEnum):
diff --git a/kafka/admin/_configs.py b/kafka/admin/_configs.py
index b4fa056d0..78ee8191d 100644
--- a/kafka/admin/_configs.py
+++ b/kafka/admin/_configs.py
@@ -186,8 +186,8 @@ def describe_configs(self, config_resources, include_synonyms=False, config_filt
         Returns:
             dict of {resource_type (str): {resource_name (str): {config_key: {config data}}}}
         """
-        return self._manager.run(self._async_describe_configs, config_resources,
-                                 include_synonyms, config_filter)
+        return self._net.run(
+            self._async_describe_configs, config_resources, include_synonyms, config_filter)
 
     @staticmethod
     def _list_config_resources_process_response(response):
@@ -226,7 +226,7 @@ def list_config_resources(self, resource_types=None):
         Returns:
             dict of {resource_type (str): [resource_name (str)]}
         """
-        return self._manager.run(self._async_list_config_resources, resource_types)
+        return self._net.run(self._async_list_config_resources, resource_types)
 
     async def _get_missing_modified_configs(self, config_resources):
         resource_lookups = [ConfigResource(resource.resource_type, resource.name) for resource in config_resources]
@@ -332,7 +332,8 @@ def alter_configs(self, config_resources, validate_only=False, raise_on_unknown=
         Returns:
             dict of {resource_type (str): {resource_name (str): Error/Result}}
         """
-        return self._manager.run(self._async_alter_configs, config_resources, validate_only, raise_on_unknown, incremental)
+        return self._net.run(
+            self._async_alter_configs, config_resources, validate_only, raise_on_unknown, incremental)
 
     async def _async_reset_configs(self, config_resources, validate_only=False, raise_on_unknown=True, incremental=None):
         if raise_on_unknown:
@@ -376,7 +377,8 @@ def reset_configs(self, config_resources, validate_only=False, raise_on_unknown=
         Returns:
             dict of {resource_type (str): {resource_name (str): Error/Result}}
         """
-        return self._manager.run(self._async_reset_configs, config_resources, validate_only, raise_on_unknown, incremental)
+        return self._net.run(
+            self._async_reset_configs, config_resources, validate_only, raise_on_unknown, incremental)
 
 
 class AlterConfigOp(EnumHelper, IntEnum):
diff --git a/kafka/admin/_groups.py b/kafka/admin/_groups.py
index 7a4070696..fbf12b0d8 100644
--- a/kafka/admin/_groups.py
+++ b/kafka/admin/_groups.py
@@ -110,7 +110,7 @@ def describe_groups(self, group_ids, group_coordinator_id=None, include_authoriz
                 of ConsumerSubscription and ConsumerAssignment metadata, and conversion
                 of acl set ints to semantic enums).
         """
-        return self._manager.run(self._async_describe_groups, group_ids, group_coordinator_id)
+        return self._net.run(self._async_describe_groups, group_ids, group_coordinator_id)
 
     # -- List groups --------------------------------------------------
 
@@ -179,8 +179,8 @@ def list_groups(self, broker_ids=None, states_filter=None, types_filter=None):
         Returns:
             List of group data dicts, with key/vals from ListGroupsRequest
         """
-        return self._manager.run(self._async_list_groups, broker_ids,
-                                 states_filter, types_filter)
+        return self._net.run(
+            self._async_list_groups, broker_ids, states_filter, types_filter)
 
     # -- List group offsets -------------------------------------------
 
@@ -303,7 +303,7 @@ def list_group_offsets(self, group_specs):
             group_specs = {group_id: None for group_id in group_specs}
         elif isinstance(group_specs, str):
             group_specs = {group_specs: None}
-        return self._manager.run(self._async_list_group_offsets, group_specs)
+        return self._net.run(self._async_list_group_offsets, group_specs)
 
     # -- Delete groups ------------------------------------------------
 
@@ -353,7 +353,7 @@ def delete_groups(self, group_ids, group_coordinator_id=None):
         Returns:
             A list of tuples (group_id, KafkaError)
         """
-        return self._manager.run(self._async_delete_groups, group_ids, group_coordinator_id)
+        return self._net.run(self._async_delete_groups, group_ids, group_coordinator_id)
 
     # -- Alter group offsets -----------------------------------------------
 
@@ -420,7 +420,7 @@ def alter_group_offsets(self, group_id, offsets, group_coordinator_id=None):
             partition-level :class:`~kafka.errors.KafkaError` class
             (``NoError`` on success).
         """
-        return self._manager.run(
+        return self._net.run(
             self._async_alter_group_offsets, group_id, offsets, group_coordinator_id)
 
     # -- Reset group offsets ----------------------------------------------
@@ -525,7 +525,7 @@ def reset_group_offsets(self, group_id, offset_specs, group_coordinator_id=None)
             {'error': :class:`~kafka.errors.KafkaError` class, 'offset': int}.
             The ``offset`` value is the post-clamp value that was committed.
         """
-        return self._manager.run(
+        return self._net.run(
             self._async_reset_group_offsets, group_id, offset_specs, group_coordinator_id)
 
     # -- Delete group offsets ----------------------------------------------
@@ -591,7 +591,7 @@ def delete_group_offsets(self, group_id, partitions, group_coordinator_id=None):
             KafkaError: If the response contains a top-level error (e.g.
                 ``GroupIdNotFoundError``, ``NonEmptyGroupError``).
         """
-        return self._manager.run(
+        return self._net.run(
             self._async_delete_group_offsets, group_id, partitions, group_coordinator_id)
 
     # -- Remove group members ---------------------------------------------
@@ -693,7 +693,7 @@ def remove_group_members(self, group_id, members, group_coordinator_id=None):
             UnsupportedVersionError: If the broker does not support batched
                 LeaveGroupRequest and any member uses ``group_instance_id``.
         """
-        return self._manager.run(
+        return self._net.run(
             self._async_remove_group_members, group_id, members, group_coordinator_id)
 
 
diff --git a/kafka/admin/_partitions.py b/kafka/admin/_partitions.py
index 7930a246b..f49c65b21 100644
--- a/kafka/admin/_partitions.py
+++ b/kafka/admin/_partitions.py
@@ -90,7 +90,7 @@ def create_partitions(self, topic_partitions, timeout_ms=None, validate_only=Fal
         def response_errors(r):
             for result in r.results:
                 yield Errors.for_code(result.error_code)
-        return self._manager.run(self._send_request_to_controller, request, response_errors, raise_errors)
+        return self._net.run(self._send_request_to_controller, request, response_errors, raise_errors)
 
     async def _async_get_leader_for_partitions(self, partitions):
         """Finds ID of the leader node for every given topic partition."""
@@ -212,9 +212,9 @@ def delete_records(self, records_to_delete, timeout_ms=None, partition_leader_id
         Returns:
             dict {topicPartition -> metadata}
         """
-        return self._manager.run(self._async_delete_records, records_to_delete,
-                                 timeout_ms, partition_leader_id,
-                                 timeout_ms=timeout_ms)
+        return self._net.run(
+            self._async_delete_records, records_to_delete, timeout_ms, partition_leader_id,
+            timeout_ms=timeout_ms)
 
     def _get_all_topic_partitions(self, topics=None):
         return [
@@ -262,7 +262,8 @@ def response_errors(r):
                 for partition in result.partition_result:
                     yield Errors.for_code(partition.error_code)
         ignore_errors = (Errors.ElectionNotNeededError,)
-        return self._manager.run(self._send_request_to_controller, request, response_errors, raise_errors, ignore_errors)
+        return self._net.run(
+            self._send_request_to_controller, request, response_errors, raise_errors, ignore_errors)
 
     @staticmethod
     def _process_alter_partition_reassignments_input(reassignments):
@@ -317,7 +318,7 @@ def alter_partition_reassignments(self, reassignments, timeout_ms=None):
 
         def top_level_error(r):
             yield Errors.for_code(r.error_code)
-        response = self._manager.run(
+        response = self._net.run(
             self._send_request_to_controller, request, top_level_error)
 
         results = {}
@@ -387,7 +388,7 @@ def list_partition_reassignments(self, topic_partitions=None, timeout_ms=None):
             with keys ``'replicas'``, ``'adding_replicas'``, and
             ``'removing_replicas'`` (each a list of broker IDs).
         """
-        return self._manager.run(
+        return self._net.run(
             self._async_list_partition_reassignments, topic_partitions, timeout_ms,
             timeout_ms=timeout_ms)
 
@@ -469,7 +470,7 @@ def describe_topic_partitions(self, topics, response_partition_limit=2000, curso
             ``next_cursor`` is None if pagination is complete, otherwise a
             dict with the next page's ``topic_name`` and ``partition_index``.
         """
-        return self._manager.run(
+        return self._net.run(
             self._async_describe_topic_partitions, topics, response_partition_limit, cursor)
 
     # -- List partition offsets --------------------------------------------
@@ -575,7 +576,7 @@ def list_partition_offsets(self, topic_partition_specs, isolation_level=Isolatio
             UnsupportedVersionError: If the broker does not support a version
                 of ListOffsetsRequest compatible with the requested specs.
         """
-        return self._manager.run(
+        return self._net.run(
             self._async_list_partition_offsets, topic_partition_specs, isolation_level, timeout_ms,
             timeout_ms=timeout_ms)
 
diff --git a/kafka/admin/_topics.py b/kafka/admin/_topics.py
index b983393e9..771c037dc 100644
--- a/kafka/admin/_topics.py
+++ b/kafka/admin/_topics.py
@@ -32,7 +32,7 @@ def list_topics(self):
         Returns:
             A list of topic name strings.
         """
-        metadata = self._manager.run(self._get_cluster_metadata, None)
+        metadata = self._net.run(self._get_cluster_metadata, None)
         return [t['name'] for t in metadata['topics']]
 
     def describe_topics(self, topics=None):
@@ -48,7 +48,7 @@ def describe_topics(self, topics=None):
         Returns:
             A list of dicts describing each topic (including partition info).
         """
-        metadata = self._manager.run(self._get_cluster_metadata, topics)
+        metadata = self._net.run(self._get_cluster_metadata, topics)
         return metadata['topics']
 
     @staticmethod
@@ -141,7 +141,7 @@ def create_topics(self, new_topics, timeout_ms=None, validate_only=False, raise_
         def response_errors(r):
             for topic in r.topics:
                 yield Errors.for_code(topic.error_code)
-        response = self._manager.run(self._send_request_to_controller, request, response_errors, raise_errors)
+        response = self._net.run(self._send_request_to_controller, request, response_errors, raise_errors)
         if wait_for_metadata:
             self.wait_for_topics([new_topic.name for new_topic in request.topics])
         result = response.to_dict()
@@ -251,7 +251,7 @@ def delete_topics(self, topics, timeout_ms=None, raise_errors=True):
         def response_errors(r):
             for response in r.responses:
                 yield Errors.for_code(response.error_code)
-        response = self._manager.run(self._send_request_to_controller, request, response_errors, raise_errors)
+        response = self._net.run(self._send_request_to_controller, request, response_errors, raise_errors)
         result = response.to_dict()
         result.pop('throttle_time_ms', None)
         result['topics'] = result.pop('responses')
diff --git a/kafka/admin/_transactions.py b/kafka/admin/_transactions.py
index eaa8907a8..7141fb53b 100644
--- a/kafka/admin/_transactions.py
+++ b/kafka/admin/_transactions.py
@@ -181,7 +181,7 @@ def list_transactions(self, broker_ids=None, producer_id_filters=None,
             dict: A dict mapping broker ``node_id`` to a list of
             :class:`TransactionListing`.
         """
-        return self._manager.run(
+        return self._net.run(
             self._async_list_transactions, broker_ids, producer_id_filters,
             state_filters, duration_filter_ms, transactional_id_pattern)
 
@@ -250,7 +250,7 @@ def describe_transactions(self, transactional_ids):
                 to its coordinator.
             BrokerResponseError: For any other per-id error.
         """
-        return self._manager.run(self._async_describe_transactions, transactional_ids)
+        return self._net.run(self._async_describe_transactions, transactional_ids)
 
     # -- DescribeProducers --------------------------------------------------
 
@@ -333,7 +333,7 @@ def describe_producers(self, partitions, broker_id=None):
                 ``NotLeaderOrFollowerError`` if the chosen broker is not
                 a replica).
         """
-        return self._manager.run(self._async_describe_producers, partitions, broker_id)
+        return self._net.run(self._async_describe_producers, partitions, broker_id)
 
     # -- AbortTransaction (WriteTxnMarkers) --------------------------------
 
@@ -382,7 +382,7 @@ def abort_transaction(self, spec):
             spec (:class:`AbortTransactionSpec`): Target partition,
                 producer id/epoch, and optional coordinator epoch.
         """
-        return self._manager.run(self._async_abort_transaction, spec)
+        return self._net.run(self._async_abort_transaction, spec)
 
     # -- find_hanging convenience ------------------------------------------
 
@@ -445,6 +445,6 @@ def find_hanging_transactions(self, broker_ids=None,
             ``state``, ``age_ms``, ``coordinator_id``,
             ``topic_partitions``.
         """
-        return self._manager.run(
+        return self._net.run(
             self._async_find_hanging_transactions, broker_ids, max_transaction_timeout_ms)
 
diff --git a/kafka/admin/_users.py b/kafka/admin/_users.py
index ddcb3cb1a..f768ee19c 100644
--- a/kafka/admin/_users.py
+++ b/kafka/admin/_users.py
@@ -70,7 +70,7 @@ def alter_user_scram_credentials(self, alterations):
         Returns:
             A dict mapping user name -> error message (or None on success).
         """
-        return self._manager.run(self._async_alter_user_scram_credentials, alterations)
+        return self._net.run(self._async_alter_user_scram_credentials, alterations)
 
     async def _async_describe_user_scram_credentials(self, users=None):
         if users is None:
@@ -118,7 +118,7 @@ def describe_user_scram_credentials(self, users=None):
             ``'error'`` (None or error message) and ``'credential_infos'``
             (list of {'mechanism': ScramMechanism, 'iterations': int}).
         """
-        return self._manager.run(self._async_describe_user_scram_credentials, users)
+        return self._net.run(self._async_describe_user_scram_credentials, users)
 
 
 class ScramMechanism(IntEnum):
diff --git a/kafka/admin/client.py b/kafka/admin/client.py
index 67878e8b0..98b36528c 100644
--- a/kafka/admin/client.py
+++ b/kafka/admin/client.py
@@ -275,7 +275,7 @@ def __init__(self, **configs):
         self._net = self._manager._net
 
         # Run all IO on a dedicated background thread; public admin methods
-        # block on cross-thread Events via self._manager.run(...).
+        # block on cross-thread Events via self._net.run(...).
         self._net.start()
 
         # Bootstrap on __init__
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py
index 9cfc8f7f9..96d30d72e 100644
--- a/kafka/consumer/fetcher.py
+++ b/kafka/consumer/fetcher.py
@@ -270,7 +270,7 @@ def send_fetches(self):
         Returns:
             List of Futures: each future resolves to a FetchResponse
         """
-        return self._manager.run(self._send_fetches_async)
+        return self._net.run(self._send_fetches_async)
 
     async def _send_fetches_async(self):
         futures = []
@@ -298,7 +298,7 @@ async def _clean_done_fetch_futures(self):
         # IO thread so it never races the foreground's list(self._fetch_futures)
         # read in fetch_records(). Defined async to enforce that -- the body
         # can only run by being driven on the IO loop (awaited from another
-        # coroutine, or scheduled via manager.run/call_soon), so the rebind
+        # coroutine, or scheduled via net.run/call_soon), so the rebind
         # always executes on the IO thread regardless of who initiates it.
         # The rebind is a single atomic attribute store, so a foreground reader
         # always sees either the old or the new deque, never a half-cleaned one.
@@ -470,7 +470,7 @@ async def _fetch_offsets_by_times_async(self, timestamps, timeout_ms=None):
                 delay = self.config['retry_backoff_ms'] / 1000
                 if timer.timeout_ms is not None:
                     delay = min(delay, timer.timeout_ms / 1000)
-                await self._manager._net.sleep(delay)
+                await self._net.sleep(delay)
 
             timer.maybe_raise()
 
@@ -734,7 +734,7 @@ async def _reset_offsets_async(self, timeout_ms=None):
                 if timer.timeout_ms is not None:
                     delay = min(delay, timer.timeout_ms / 1000)
                 if delay > 0:
-                    await self._manager._net.sleep(delay)
+                    await self._net.sleep(delay)
                 continue
 
             offset_resets = {}
@@ -1015,7 +1015,7 @@ async def _validate_offsets_async(self, timeout_ms=None):
                 if timer.timeout_ms is not None:
                     delay = min(delay, timer.timeout_ms / 1000)
                 if delay > 0:
-                    await self._manager._net.sleep(delay)
+                    await self._net.sleep(delay)
                 continue
 
             positions = {}
diff --git a/test/admin/test_admin_concurrent.py b/test/admin/test_admin_concurrent.py
index d2487a524..7b0cec42a 100644
--- a/test/admin/test_admin_concurrent.py
+++ b/test/admin/test_admin_concurrent.py
@@ -67,21 +67,21 @@ def test_close_unblocks_pending_callers(broker):
         bootstrap_servers='%s:%d' % (broker.host, broker.port),
         request_timeout_ms=5000,
     )
-    manager = admin._manager
+    net = admin._net
 
     blocked = threading.Event()
     released = threading.Event()
 
     async def blocker():
         blocked.set()
-        # Wait forever - only unblocks via manager.stop()
-        await manager._net.sleep(3600)
+        # Wait forever - only unblocks via net.stop()
+        await net.sleep(3600)
 
     result = {}
 
     def caller():
         try:
-            manager.run(blocker)
+            net.run(blocker)
         except BaseException as exc:
             result['exception'] = exc
         finally:
diff --git a/test/admin/test_admin_groups.py b/test/admin/test_admin_groups.py
index 66e603fdb..4db393c05 100644
--- a/test/admin/test_admin_groups.py
+++ b/test/admin/test_admin_groups.py
@@ -1025,8 +1025,8 @@ def find_handler(api_key, api_version, correlation_id, request_bytes):
         broker.respond_fn(FindCoordinatorRequest, find_handler)
         broker.respond_fn(FindCoordinatorRequest, find_handler)
 
-        group_node = admin._manager.run(admin._find_coordinator_id, 'x', 0)
-        txn_node = admin._manager.run(admin._find_coordinator_id, 'x', 1)
+        group_node = admin._net.run(admin._find_coordinator_id, 'x', 0)
+        txn_node = admin._net.run(admin._find_coordinator_id, 'x', 1)
 
         assert group_node == 1
         assert txn_node == 2
diff --git a/test/consumer/test_coordinator.py b/test/consumer/test_coordinator.py
index 1694bca94..25c3ee4c3 100644
--- a/test/consumer/test_coordinator.py
+++ b/test/consumer/test_coordinator.py
@@ -177,7 +177,7 @@ def test_join_complete(mocker, coordinator):
     assert assignor.on_assignment.call_count == 0
     assignment = ConsumerProtocolAssignment(0, [('foobar', [0, 1])], b'')
     generation = 12
-    coordinator._manager.run(
+    coordinator._net.run(
         coordinator._on_join_complete_async,
         generation, 'member-foo', 'roundrobin', assignment.encode())
     assert assignor.on_assignment.call_count == 1
@@ -192,7 +192,7 @@ def test_join_complete_with_sticky_assignor(mocker, coordinator):
     assert assignor.on_assignment.call_count == 0
     generation = 3
     assignment = ConsumerProtocolAssignment(0, [('foobar', [0, 1])], b'')
-    coordinator._manager.run(
+    coordinator._net.run(
         coordinator._on_join_complete_async,
         generation, 'member-foo', 'sticky', assignment.encode())
     assert assignor.on_assignment.call_count == 1
@@ -236,7 +236,7 @@ def test_on_join_prepare_async_invokes_sync_listener(mocker, coordinator):
     listener = mocker.MagicMock(spec=ConsumerRebalanceListener)
     coordinator._subscription.subscribe(topics=['foobar'], listener=listener)
 
-    coordinator._manager.run(coordinator._on_join_prepare_async, 0, 'member-foo')
+    coordinator._net.run(coordinator._on_join_prepare_async, 0, 'member-foo')
 
     assert listener.on_partitions_revoked.call_count == 1
     listener.on_partitions_revoked.assert_called_with(set())
@@ -254,7 +254,7 @@ async def on_partitions_assigned(self, assigned):
             calls.append(('assigned', assigned))
 
     coordinator._subscription.subscribe(topics=['foobar'], listener=MyListener())
-    coordinator._manager.run(coordinator._on_join_prepare_async, 0, 'member-foo')
+    coordinator._net.run(coordinator._on_join_prepare_async, 0, 'member-foo')
 
     assert calls == [('revoked', set())]
 
@@ -280,7 +280,7 @@ def on_partitions_assigned(self, assigned):
     log_warning = mocker.patch('kafka.coordinator.consumer.log.warning')
 
     # Below threshold: no warning.
-    coordinator._manager.run(coordinator._on_join_prepare_async, 0, 'member-foo')
+    coordinator._net.run(coordinator._on_join_prepare_async, 0, 'member-foo')
     assert not any(
         'Rebalance listener' in str(call.args[0])
         for call in log_warning.call_args_list)
@@ -288,7 +288,7 @@ def on_partitions_assigned(self, assigned):
     # Above threshold: drop the threshold to a tiny value and re-run.
     mocker.patch.object(coordinator, '_REBALANCE_LISTENER_WARN_SECS', 0.001)
     log_warning.reset_mock()
-    coordinator._manager.run(coordinator._on_join_prepare_async, 0, 'member-foo')
+    coordinator._net.run(coordinator._on_join_prepare_async, 0, 'member-foo')
     matching = [c for c in log_warning.call_args_list
                 if 'Rebalance listener' in str(c.args[0])]
     assert len(matching) == 1
@@ -311,7 +311,7 @@ def test_on_join_prepare_async_listener_exception_propagates(mocker, coordinator
     coordinator._subscription.subscribe(topics=['foobar'], listener=listener)
 
     with pytest.raises(Errors.KafkaError) as exc_info:
-        coordinator._manager.run(coordinator._on_join_prepare_async, 42, 'member-foo')
+        coordinator._net.run(coordinator._on_join_prepare_async, 42, 'member-foo')
     assert exc_info.value.__cause__ is crash
     # Cleanup still ran before the exception bubbled out.
     assert coordinator._is_leader is False
@@ -322,7 +322,7 @@ def test_on_join_prepare_async_skips_auto_commit_when_disabled(mocker, coordinat
     spy = mocker.spy(coordinator, '_commit_offsets_sync_async')
     coordinator._subscription.subscribe(topics=['foobar'])
 
-    coordinator._manager.run(coordinator._on_join_prepare_async, 0, 'member-foo')
+    coordinator._net.run(coordinator._on_join_prepare_async, 0, 'member-foo')
 
     assert spy.call_count == 0
 
@@ -335,7 +335,7 @@ async def _noop(*args, **kwargs):
                               side_effect=_noop)
     coordinator._subscription.subscribe(topics=['foobar'])
 
-    coordinator._manager.run(coordinator._on_join_prepare_async, 0, 'member-foo')
+    coordinator._net.run(coordinator._on_join_prepare_async, 0, 'member-foo')
 
     assert spy.call_count == 1
 
@@ -347,7 +347,7 @@ def test_on_join_complete_async_invokes_sync_listener(mocker, coordinator):
     coordinator._assignors = {assignor.name: assignor}
     assignment = ConsumerProtocolAssignment(0, [('foobar', [0, 1])], b'')
 
-    coordinator._manager.run(
+    coordinator._net.run(
         coordinator._on_join_complete_async,
         12, 'member-foo', 'roundrobin', assignment.encode())
 
@@ -371,7 +371,7 @@ async def on_partitions_assigned(self, assigned):
     coordinator._assignors = {assignor.name: assignor}
     assignment = ConsumerProtocolAssignment(0, [('foobar', [0, 1])], b'')
 
-    coordinator._manager.run(
+    coordinator._net.run(
         coordinator._on_join_complete_async,
         12, 'member-foo', 'roundrobin', assignment.encode())
 
@@ -392,7 +392,7 @@ def test_on_join_complete_async_listener_exception_propagates(mocker, coordinato
     assignment = ConsumerProtocolAssignment(0, [('foobar', [0, 1])], b'')
 
     with pytest.raises(Errors.KafkaError) as exc_info:
-        coordinator._manager.run(
+        coordinator._net.run(
             coordinator._on_join_complete_async,
             12, 'member-foo', 'roundrobin', assignment.encode())
     assert exc_info.value.__cause__ is crash
@@ -1184,7 +1184,7 @@ def test_do_join_and_sync_async_follower(request, broker, seeded_coord):
     broker.respond(SyncGroupRequest, _sync_response_object(
         assignment=expected_assignment))
 
-    result = seeded_coord._manager.run(seeded_coord._do_join_and_sync_async)
+    result = seeded_coord._net.run(seeded_coord._do_join_and_sync_async)
 
     assert result == expected_assignment
     assert seeded_coord._generation.generation_id == 42
@@ -1219,7 +1219,7 @@ def sync_handler(api_key, api_version, correlation_id, request_bytes):
     # Spy on _perform_assignment to confirm the leader path ran the assignor.
     spy = mocker.spy(seeded_coord, '_perform_assignment')
 
-    result = seeded_coord._manager.run(seeded_coord._do_join_and_sync_async)
+    result = seeded_coord._net.run(seeded_coord._do_join_and_sync_async)
 
     assert spy.call_count == 1
     leader_id, protocol_name, members_arg = spy.call_args[0]
@@ -1237,7 +1237,7 @@ def test_do_join_and_sync_async_coordinator_unknown(request, seeded_coord):
     request.addfinalizer(lambda: setattr(seeded_coord, 'state', MemberState.UNJOINED))
     seeded_coord.coordinator_id = None  # force coordinator_unknown
     with pytest.raises(Errors.CoordinatorNotAvailableError):
-        seeded_coord._manager.run(seeded_coord._do_join_and_sync_async)
+        seeded_coord._net.run(seeded_coord._do_join_and_sync_async)
 
 
 @pytest.mark.parametrize('error_code,error_type', [
@@ -1253,7 +1253,7 @@ def test_do_join_and_sync_async_join_error(request, broker, seeded_coord,
     request.addfinalizer(lambda: setattr(seeded_coord, 'state', MemberState.UNJOINED))
     broker.respond(JoinGroupRequest, _join_response_object(error_code=error_code))
     with pytest.raises(error_type):
-        seeded_coord._manager.run(seeded_coord._do_join_and_sync_async)
+        seeded_coord._net.run(seeded_coord._do_join_and_sync_async)
 
 
 @pytest.mark.parametrize('error_code,error_type', [
@@ -1271,7 +1271,7 @@ def test_do_join_and_sync_async_sync_error(request, broker, seeded_coord,
         leader='leader-x', member_id='member-1'))
     broker.respond(SyncGroupRequest, _sync_response_object(error_code=error_code))
     with pytest.raises(error_type):
-        seeded_coord._manager.run(seeded_coord._do_join_and_sync_async)
+        seeded_coord._net.run(seeded_coord._do_join_and_sync_async)
     # All sync errors flip rejoin_needed via request_rejoin().
     assert seeded_coord.rejoin_needed is True
 
@@ -1283,7 +1283,7 @@ def test_join_group_async_no_rejoin_returns_true(request, mocker, broker, seeded
     seeded_coord.state = MemberState.STABLE
 
     before = broker.requests_received
-    result = seeded_coord._manager.run(seeded_coord.join_group_async, 5000)
+    result = seeded_coord._net.run(seeded_coord.join_group_async, 5000)
 
     assert result is True
     assert broker.requests_received == before
@@ -1298,7 +1298,7 @@ def test_join_group_async_happy_path_follower(request, broker, seeded_coord):
     broker.respond(SyncGroupRequest, _sync_response_object(
         assignment=ConsumerProtocolAssignment(0, [('foobar', [0, 1])], b'').encode()))
 
-    result = seeded_coord._manager.run(seeded_coord.join_group_async, 5000)
+    result = seeded_coord._net.run(seeded_coord.join_group_async, 5000)
 
     assert result is True
     assert seeded_coord.state == MemberState.STABLE
@@ -1332,7 +1332,7 @@ def capturing_send(req, node_id=None, request_timeout_ms=None):
 
     mocker.patch.object(seeded_coord._manager, 'send', side_effect=capturing_send)
 
-    seeded_coord._manager.run(seeded_coord.join_group_async, 5000)
+    seeded_coord._net.run(seeded_coord.join_group_async, 5000)
 
     join_sends = [t for name, t in sends if 'JoinGroup' in name]
     assert join_sends, "expected at least one JoinGroupRequest send"
@@ -1358,7 +1358,7 @@ def test_join_group_async_retries_on_retriable_error(request, broker, seeded_coo
     broker.respond(SyncGroupRequest, _sync_response_object(
         assignment=ConsumerProtocolAssignment(0, [('foobar', [0, 1])], b'').encode()))
 
-    result = seeded_coord._manager.run(seeded_coord.join_group_async, 5000)
+    result = seeded_coord._net.run(seeded_coord.join_group_async, 5000)
 
     assert result is True
     assert seeded_coord.state == MemberState.STABLE
@@ -1372,7 +1372,7 @@ def test_join_group_async_raises_non_retriable(request, broker, seeded_coord):
         error_code=Errors.GroupAuthorizationFailedError.errno))
 
     with pytest.raises(Errors.GroupAuthorizationFailedError):
-        seeded_coord._manager.run(seeded_coord.join_group_async, 5000)
+        seeded_coord._net.run(seeded_coord.join_group_async, 5000)
 
 
 def test_join_group_async_returns_false_on_short_timeout_and_caches_task(
@@ -1411,7 +1411,7 @@ async def slow_join_handler(api_key, api_version, correlation_id, request_bytes)
     # await on JoinGroup is not timer-aware, this call hangs until the
     # connection's request_timeout_ms fires (~5s in the fixture).
     start = time.monotonic()
-    result = seeded_coord._manager.run(seeded_coord.join_group_async, 50)
+    result = seeded_coord._net.run(seeded_coord.join_group_async, 50)
     elapsed = time.monotonic() - start
     assert result is False
     assert elapsed < 1.0, (
@@ -1422,7 +1422,7 @@ async def slow_join_handler(api_key, api_version, correlation_id, request_bytes)
     # Second call: broker is still hanging. Should reuse the cached
     # in-flight task instead of sending a duplicate JoinGroup.
     start = time.monotonic()
-    result = seeded_coord._manager.run(seeded_coord.join_group_async, 50)
+    result = seeded_coord._net.run(seeded_coord.join_group_async, 50)
     elapsed = time.monotonic() - start
     assert result is False
     assert elapsed < 1.0
@@ -1434,7 +1434,7 @@ async def slow_join_handler(api_key, api_version, correlation_id, request_bytes)
         assignment=ConsumerProtocolAssignment(0, [('foobar', [0, 1])], b'').encode()))
     join_response_pending.success(None)
 
-    result = seeded_coord._manager.run(seeded_coord.join_group_async, 5000)
+    result = seeded_coord._net.run(seeded_coord.join_group_async, 5000)
     assert result is True
     assert join_request_count[0] == 1, (
         'duplicate JoinGroup sent on the success path')
@@ -1444,7 +1444,7 @@ async def slow_join_handler(api_key, api_version, correlation_id, request_bytes)
 @pytest.mark.parametrize("broker", [(0, 8, 0)], indirect=True)
 def test_join_group_async_unsupported_version(broker, coordinator):
     with pytest.raises(Errors.UnsupportedVersionError):
-        coordinator._manager.run(coordinator.join_group_async, None)
+        coordinator._net.run(coordinator.join_group_async, None)
 
 
 def test_ensure_active_group_async_happy_path(request, broker, seeded_coord):
@@ -1456,7 +1456,7 @@ def test_ensure_active_group_async_happy_path(request, broker, seeded_coord):
     broker.respond(SyncGroupRequest, _sync_response_object(
         assignment=ConsumerProtocolAssignment(0, [('foobar', [0, 1])], b'').encode()))
 
-    result = seeded_coord._manager.run(seeded_coord.ensure_active_group_async, 5000)
+    result = seeded_coord._net.run(seeded_coord.ensure_active_group_async, 5000)
 
     assert result is True
     assert seeded_coord.state == MemberState.STABLE
@@ -1465,7 +1465,7 @@ def test_ensure_active_group_async_happy_path(request, broker, seeded_coord):
 
 
 def test_ensure_active_group_sync_facade(request, broker, seeded_coord):
-    """The sync ensure_active_group facade dispatches via manager.run."""
+    """The sync ensure_active_group facade dispatches via net.run."""
     request.addfinalizer(lambda: setattr(seeded_coord, 'state', MemberState.UNJOINED))
     seeded_coord.rejoin_needed = True
     seeded_coord.state = MemberState.UNJOINED
@@ -1549,7 +1549,7 @@ async def fake_send():
     assert future.failed()
 
 
-def test_do_join_and_sync_async_join_protocol_type_mismatch(request, broker, manager, seeded_coord):
+def test_do_join_and_sync_async_join_protocol_type_mismatch(request, broker, net, seeded_coord):
     """KIP-559: JoinGroupResponse with mismatched protocol_type must raise
     InconsistentGroupProtocolError."""
     request.addfinalizer(lambda: setattr(seeded_coord, 'state', MemberState.UNJOINED))
@@ -1558,10 +1558,10 @@ def test_do_join_and_sync_async_join_protocol_type_mismatch(request, broker, man
         protocol_type='not-consumer'))
 
     with pytest.raises(Errors.InconsistentGroupProtocolError):
-        manager.run(seeded_coord._do_join_and_sync_async)
+        net.run(seeded_coord._do_join_and_sync_async)
 
 
-def test_do_join_and_sync_async_sync_protocol_type_mismatch(request, broker, manager, seeded_coord):
+def test_do_join_and_sync_async_sync_protocol_type_mismatch(request, broker, net, seeded_coord):
     """KIP-559: SyncGroupResponse with mismatched protocol_type must raise."""
     request.addfinalizer(lambda: setattr(seeded_coord, 'state', MemberState.UNJOINED))
     broker.respond(JoinGroupRequest, _join_response_object(
@@ -1570,10 +1570,10 @@ def test_do_join_and_sync_async_sync_protocol_type_mismatch(request, broker, man
         protocol_type='not-consumer'))
 
     with pytest.raises(Errors.InconsistentGroupProtocolError):
-        manager.run(seeded_coord._do_join_and_sync_async)
+        net.run(seeded_coord._do_join_and_sync_async)
 
 
-def test_do_join_and_sync_async_sync_protocol_name_mismatch(request, broker, manager, seeded_coord):
+def test_do_join_and_sync_async_sync_protocol_name_mismatch(request, broker, net, seeded_coord):
     """KIP-559: SyncGroupResponse with mismatched protocol_name must raise."""
     request.addfinalizer(lambda: setattr(seeded_coord, 'state', MemberState.UNJOINED))
     broker.respond(JoinGroupRequest, _join_response_object(
@@ -1583,7 +1583,7 @@ def test_do_join_and_sync_async_sync_protocol_name_mismatch(request, broker, man
         protocol_name='roundrobin'))
 
     with pytest.raises(Errors.InconsistentGroupProtocolError):
-        manager.run(seeded_coord._do_join_and_sync_async)
+        net.run(seeded_coord._do_join_and_sync_async)
 
 
 # ---------------------------------------------------------------------------
@@ -1648,7 +1648,7 @@ def test_eager_revokes_everything(self, mocker, coordinator):
         # Real generation - otherwise is_lost() trips and the lost
         # branch fires on_partitions_lost instead.
         coordinator._generation = Generation(42, 'member-foo', 'range')
-        coordinator._manager.run(coordinator._on_join_prepare_async, 42, 'member-foo')
+        coordinator._net.run(coordinator._on_join_prepare_async, 42, 'member-foo')
         listener.on_partitions_revoked.assert_called_once_with(
             {TopicPartition('foobar', 0), TopicPartition('foobar', 1)})
 
@@ -1663,7 +1663,7 @@ def test_cooperative_skips_global_revoke(self, mocker, client, metrics):
             # Real generation - otherwise is_lost() trips and the lost
             # branch fires on_partitions_lost instead.
             coord._generation = Generation(42, 'member-foo', 'cooperative-sticky')
-            coord._manager.run(coord._on_join_prepare_async, 42, 'member-foo')
+            coord._net.run(coord._on_join_prepare_async, 42, 'member-foo')
             # KIP-429: no global on_partitions_revoked in the prepare
             # phase - individual partitions are revoked later in
             # _on_join_complete based on the diff.
@@ -1691,7 +1691,7 @@ def test_cooperative_revokes_unsubscribed_topic_partitions_early(
             coord._subscription.change_subscription(['t'])
             coord._generation = Generation(42, 'mbr-1', 'cooperative-sticky')
 
-            coord._manager.run(coord._on_join_prepare_async, 42, 'mbr-1')
+            coord._net.run(coord._on_join_prepare_async, 42, 'mbr-1')
 
             # Only the unsubscribed-topic partitions should be revoked,
             # and only those should be dropped from the assignment.
@@ -1731,7 +1731,7 @@ def on_partitions_assigned(self, assigned):
             coord._subscription.change_subscription(['t'])
             coord._generation = Generation(42, 'mbr-1', 'cooperative-sticky')
 
-            coord._manager.run(coord._on_join_prepare_async, 42, 'mbr-1')
+            coord._net.run(coord._on_join_prepare_async, 42, 'mbr-1')
 
             assert listener.fetchable_at_revoke == {
                 TopicPartition('old', 0): False}
@@ -1752,7 +1752,7 @@ def test_cooperative_no_unsubscribed_topics_skips_listener(
                 TopicPartition('t', 0), TopicPartition('t', 1)])
             coord._generation = Generation(42, 'mbr-1', 'cooperative-sticky')
 
-            coord._manager.run(coord._on_join_prepare_async, 42, 'mbr-1')
+            coord._net.run(coord._on_join_prepare_async, 42, 'mbr-1')
 
             listener.on_partitions_revoked.assert_not_called()
             assert coord._subscription.assigned_partitions() == {
@@ -1786,7 +1786,7 @@ def test_cooperative_revoke_then_assign_diff(self, mocker, client, metrics):
                 TopicPartition('t', 2)])
 
             assignment_bytes = self._make_assignment_bytes('t', [1, 2, 3])
-            coord._manager.run(
+            coord._net.run(
                 coord._on_join_complete_async,
                 42, 'member-1', 'cooperative-sticky', assignment_bytes)
 
@@ -1807,7 +1807,7 @@ def test_cooperative_stable_assignment_no_listener_calls(
                 TopicPartition('t', 0), TopicPartition('t', 1)])
 
             assignment_bytes = self._make_assignment_bytes('t', [0, 1])
-            coord._manager.run(
+            coord._net.run(
                 coord._on_join_complete_async,
                 42, 'member-1', 'cooperative-sticky', assignment_bytes)
 
@@ -1830,7 +1830,7 @@ def test_cooperative_revoke_triggers_request_rejoin(
 
             # New assignment drops partition 1.
             assignment_bytes = self._make_assignment_bytes('t', [0])
-            coord._manager.run(
+            coord._net.run(
                 coord._on_join_complete_async,
                 42, 'member-1', 'cooperative-sticky', assignment_bytes)
 
@@ -1850,7 +1850,7 @@ def test_cooperative_no_revoke_no_extra_rejoin(
             spy = mocker.spy(coord, 'request_rejoin')
 
             assignment_bytes = self._make_assignment_bytes('t', [0, 1])
-            coord._manager.run(
+            coord._net.run(
                 coord._on_join_complete_async,
                 42, 'member-1', 'cooperative-sticky', assignment_bytes)
 
@@ -1883,7 +1883,7 @@ def test_cooperative_preserves_state_on_kept_partition(
 
             # Rebalance: keep partition 0, drop 1, add 2.
             assignment_bytes = self._make_assignment_bytes('t', [0, 2])
-            coord._manager.run(
+            coord._net.run(
                 coord._on_join_complete_async,
                 42, 'member-1', 'cooperative-sticky', assignment_bytes)
 
@@ -1920,7 +1920,7 @@ def test_cooperative_assignment_for_unsubscribed_topic_bails(
 
             # Assignment names a topic the consumer isn't subscribed to.
             bad_assignment = self._make_assignment_bytes('other-topic', [0])
-            coord._manager.run(
+            coord._net.run(
                 coord._on_join_complete_async,
                 42, 'member-1', 'cooperative-sticky', bad_assignment)
 
@@ -1973,7 +1973,7 @@ async def on_partitions_assigned(self, assigned):
                 calls.append(('assigned', assigned))
 
         lost = {TopicPartition('t', 0)}
-        coordinator._manager.run(OldAsync().on_partitions_lost, lost)
+        coordinator._net.run(OldAsync().on_partitions_lost, lost)
         assert calls == [('revoked', lost)]
 
     def test_generation_is_lost(self):
@@ -2015,7 +2015,7 @@ def test_on_join_prepare_fires_lost_and_clears_assignment(
         coordinator.reset_generation()
         assert coordinator._generation.is_lost() is True
 
-        coordinator._manager.run(
+        coordinator._net.run(
             coordinator._on_join_prepare_async, 0, 'member-foo')
 
         listener.on_partitions_lost.assert_called_once_with(
@@ -2037,7 +2037,7 @@ def test_on_join_prepare_skips_auto_commit_when_lost(
             coordinator, '_commit_offsets_sync_async')
         coordinator.reset_generation()
 
-        coordinator._manager.run(
+        coordinator._net.run(
             coordinator._on_join_prepare_async, 0, 'member-foo')
 
         commit_spy.assert_not_called()
@@ -2055,14 +2055,14 @@ def test_on_join_prepare_after_lost_then_normal(
         coordinator.reset_generation()
 
         # First call: lost path.
-        coordinator._manager.run(
+        coordinator._net.run(
             coordinator._on_join_prepare_async, 0, 'member-foo')
         # Re-assign and install a real generation; the next prepare
         # should fire revoked, not lost.
         coordinator._generation = Generation(42, 'mbr-1', 'range')
         coordinator._subscription.assign_from_subscribed([TopicPartition('t', 0)])
         listener.reset_mock()
-        coordinator._manager.run(
+        coordinator._net.run(
             coordinator._on_join_prepare_async, 42, 'mbr-1')
         listener.on_partitions_lost.assert_not_called()
         listener.on_partitions_revoked.assert_called_once_with(
@@ -2142,7 +2142,7 @@ async def on_partitions_lost(self, lost):
         coordinator._subscription.subscribe(topics=['t'], listener=AsyncListener())
         coordinator._subscription.assign_from_subscribed([TopicPartition('t', 0)])
         coordinator.reset_generation()
-        coordinator._manager.run(
+        coordinator._net.run(
             coordinator._on_join_prepare_async, 0, 'member-foo')
 
         assert calls == [('lost', {TopicPartition('t', 0)})]
@@ -2160,7 +2160,7 @@ def test_lost_listener_exception_propagates(self, mocker, coordinator):
         coordinator.reset_generation()
 
         with pytest.raises(Errors.KafkaError) as exc_info:
-            coordinator._manager.run(
+            coordinator._net.run(
                 coordinator._on_join_prepare_async, 0, 'member-foo')
         assert exc_info.value.__cause__ is crash
 
@@ -2267,7 +2267,7 @@ def test_cooperative_both_listeners_called_even_if_revoked_throws(
 
             assignment_bytes = self._make_assignment_bytes('t', [1, 2])
             with pytest.raises(Errors.KafkaError):
-                coord._manager.run(
+                coord._net.run(
                     coord._on_join_complete_async,
                     42, 'member-1', 'cooperative-sticky', assignment_bytes)
 
@@ -2294,7 +2294,7 @@ def test_cooperative_assigned_throw_propagates(
 
             assignment_bytes = self._make_assignment_bytes('t', [0, 1])
             with pytest.raises(Errors.KafkaError) as exc_info:
-                coord._manager.run(
+                coord._net.run(
                     coord._on_join_complete_async,
                     42, 'member-1', 'cooperative-sticky', assignment_bytes)
             assert exc_info.value.__cause__ is assigned_crash
@@ -2331,7 +2331,7 @@ def test_eager_assignment_for_unsubscribed_topic_bails(
         old_deadline = coordinator.next_auto_commit_deadline
 
         bad_assignment = self._make_assignment_bytes('other-topic', [0])
-        coordinator._manager.run(
+        coordinator._net.run(
             coordinator._on_join_complete_async,
             42, 'member-1', assignor_name, bad_assignment)
 
@@ -2497,9 +2497,8 @@ def test_close_no_autocommit_still_revokes(self, mocker, coordinator):
 def _dispatch_heartbeat(coord):
     """Dispatch a single _send_heartbeat_request and pump the network until
     the resulting future resolves."""
-    manager = coord._manager
-    future = manager.call_soon(coord._send_heartbeat_request)
     net = coord._manager._net
+    future = net.call_soon_with_future(coord._send_heartbeat_request)
     net.wait_for(future, timeout_ms=5000, raise_error=False)
     return future
 
diff --git a/test/consumer/test_fetcher.py b/test/consumer/test_fetcher.py
index 0f119e7e2..e70cb7e82 100644
--- a/test/consumer/test_fetcher.py
+++ b/test/consumer/test_fetcher.py
@@ -173,7 +173,7 @@ def test_reset_offsets_if_needed(fetcher, topic, mocker):
 
 
 def test__reset_offsets_async_waits_for_metadata_when_leader_unknown(
-        fetcher, manager, mocker):
+        fetcher, net, mocker):
     """If the leader is unknown, _reset_offsets_async should wait for a
     metadata refresh and retry within the timer budget. Regression for the
     test_kafka_consumer_position_after_seek_to_end integration test where
@@ -205,7 +205,7 @@ async def fake_send(node_id, timestamps_and_epochs):
         return ({tp: OffsetAndTimestamp(42, None, -1)}, set())
     mocker.patch.object(fetcher, '_send_list_offsets_request', side_effect=fake_send)
 
-    manager.run(fetcher._reset_offsets_async, 1000)
+    net.run(fetcher._reset_offsets_async, 1000)
 
     assert not fetcher._subscriptions.assignment[tp].awaiting_reset
     assert fetcher._subscriptions.assignment[tp].position.offset == 42
@@ -214,7 +214,7 @@ async def fake_send(node_id, timestamps_and_epochs):
 
 
 def test__reset_offsets_async_bails_when_leader_permanently_unknown(
-        fetcher, manager, mocker):
+        fetcher, net, mocker):
     """If metadata refresh doesn't resolve the leader within the timer
     budget, _reset_offsets_async should bail rather than spin forever.
     """
@@ -231,7 +231,7 @@ def test__reset_offsets_async_bails_when_leader_permanently_unknown(
 
     # 50ms timer caps the spin even though leader stays unknown forever.
     start = time.monotonic()
-    manager.run(fetcher._reset_offsets_async, 50)
+    net.run(fetcher._reset_offsets_async, 50)
     elapsed = time.monotonic() - start
     assert elapsed < 1.0, (
         '_reset_offsets_async did not respect timer budget; took %.2fs' % elapsed)
@@ -240,7 +240,7 @@ def test__reset_offsets_async_bails_when_leader_permanently_unknown(
     assert fetcher._subscriptions.assignment[tp].awaiting_reset
 
 
-def test__reset_offsets_async(fetcher, manager, mocker):
+def test__reset_offsets_async(fetcher, net, mocker):
     tp0 = TopicPartition("topic", 0)
     tp1 = TopicPartition("topic", 1)
     fetcher._subscriptions.subscribe(topics=["topic"])
@@ -260,8 +260,8 @@ async def fake_send(node_id, timestamps_and_epochs):
     mocker.patch.object(fetcher, '_send_list_offsets_request', side_effect=fake_send)
 
     # _reset_offsets_async loops until partitions_needing_reset is empty;
-    # manager.run waits for the whole driver to complete.
-    manager.run(fetcher._reset_offsets_async, 1000)
+    # net.run waits for the whole driver to complete.
+    net.run(fetcher._reset_offsets_async, 1000)
 
     assert not fetcher._subscriptions.assignment[tp0].awaiting_reset
     assert not fetcher._subscriptions.assignment[tp1].awaiting_reset
@@ -270,7 +270,7 @@ async def fake_send(node_id, timestamps_and_epochs):
 
 
 def test__reset_offsets_async_retries_after_retriable_failure(
-        fetcher, manager, mocker):
+        fetcher, net, mocker):
     """A retriable per-partition error (NotLeader, etc.) sets the partition
     into retry_backoff_ms backoff. _reset_offsets_async must sleep for that
     backoff and retry, rather than exit and rely on an outer caller to
@@ -298,7 +298,7 @@ async def fake_send(node_id, timestamps_and_epochs):
                         side_effect=fake_send)
 
     start = time.monotonic()
-    manager.run(fetcher._reset_offsets_async, 1000)
+    net.run(fetcher._reset_offsets_async, 1000)
     elapsed = time.monotonic() - start
 
     assert call_count[0] == 2, (
@@ -316,7 +316,7 @@ def test__send_list_offsets_requests(fetcher, manager, net, mocker):
 
     pending = []
     async def fake_send(node_id, timestamps):
-        f = fetcher._manager.create_future()  # awaited below
+        f = net.create_future()  # awaited below
         pending.append(f)
         return await f
     mocked_send = mocker.patch.object(fetcher, "_send_list_offsets_request", side_effect=fake_send)
@@ -332,12 +332,12 @@ async def fake_send(node_id, timestamps):
 
     # Leader == None
     with pytest.raises(StaleMetadata):
-        manager.run(fetcher._send_list_offsets_requests, {tp: 0})
+        net.run(fetcher._send_list_offsets_requests, {tp: 0})
     assert not mocked_send.called
 
     # Leader == -1
     with pytest.raises(StaleMetadata):
-        manager.run(fetcher._send_list_offsets_requests, {tp: 0})
+        net.run(fetcher._send_list_offsets_requests, {tp: 0})
     assert not mocked_send.called
 
     # Leader == 0, send failed
@@ -574,7 +574,7 @@ def test_fetch_records_not_idle_when_reset_task_pending(fetcher, mocker):
     assert idle is False
 
 
-def test_clean_done_fetch_futures_removes_out_of_order_done(fetcher):
+def test_clean_done_fetch_futures_removes_out_of_order_done(fetcher, net):
     """Completed fetch futures must be dropped regardless of position, not
     just a contiguous run from the head. With multiple brokers, fetches
     complete out of order, so a head-only cleanup would strand an
@@ -588,7 +588,7 @@ def test_clean_done_fetch_futures_removes_out_of_order_done(fetcher):
     fetcher._fetch_futures.extend([done_head, pending, done_tail])
 
     # async def: must be driven on the IO loop (it rebinds _fetch_futures).
-    fetcher._manager.run(fetcher._clean_done_fetch_futures)
+    net.run(fetcher._clean_done_fetch_futures)
 
     # Only the still-in-flight future survives.
     assert list(fetcher._fetch_futures) == [pending]
@@ -616,7 +616,7 @@ def test_in_flight_fetches_is_read_only(fetcher):
     assert list(fetcher._fetch_futures) == [done]
 
 
-def test_clean_done_fetch_futures_only_mutates_when_driven_on_loop(fetcher):
+def test_clean_done_fetch_futures_only_mutates_when_driven_on_loop(fetcher, net):
     """_clean_done_fetch_futures is async so its rebind of _fetch_futures can
     only run when driven on the IO loop. A stray *synchronous* call from the
     foreground yields an inert coroutine (no mutation) rather than a silent
@@ -633,7 +633,7 @@ def test_clean_done_fetch_futures_only_mutates_when_driven_on_loop(fetcher):
         coro.close()  # avoid "coroutine was never awaited" warning
 
     # Driven on the IO loop, it actually evicts the completed future.
-    fetcher._manager.run(fetcher._clean_done_fetch_futures)
+    net.run(fetcher._clean_done_fetch_futures)
     assert list(fetcher._fetch_futures) == [pending]
 
 
@@ -697,7 +697,7 @@ def realistic_wait_for(wakeup, timeout_ms=None, raise_error=False):
     assert len(records[tp]) == 1
 
 
-def test_fetch_records_blocks_once_stale_fetch_is_cleaned(fetcher, mocker):
+def test_fetch_records_blocks_once_stale_fetch_is_cleaned(fetcher, net, mocker):
     """Multi-broker busy-loop regression. A stale (already-drained) completion
     stranded behind an in-flight fetch must be evicted by the cleanup so the
     wait actually blocks on the in-flight future instead of returning
@@ -709,7 +709,7 @@ def test_fetch_records_blocks_once_stale_fetch_is_cleaned(fetcher, mocker):
     # could never reach the stale completion behind it.
     fetcher._fetch_futures.extend([inflight, stale_done])
 
-    fetcher._manager.run(fetcher._clean_done_fetch_futures)
+    net.run(fetcher._clean_done_fetch_futures)
     assert list(fetcher._fetch_futures) == [inflight]
 
     mocker.patch.object(fetcher, 'send_fetches', return_value=None)
@@ -1083,7 +1083,7 @@ def test_partition_records_compacted_offset(mocker):
     assert msgs[0].offset == fetch_offset + 1
 
 
-def test_reset_offsets_paused(subscription_state, client, manager, net, mocker):
+def test_reset_offsets_paused(subscription_state, client, net, mocker):
     fetcher = Fetcher(client, subscription_state)
     tp = TopicPartition('foo', 0)
     subscription_state.assign_from_user([tp])
@@ -1096,7 +1096,7 @@ async def fake_send(node_id, timestamps_and_epochs):
     mocker.patch.object(fetcher._client, 'ready', return_value=True)
     mocker.patch.object(fetcher, '_send_list_offsets_request', side_effect=fake_send)
     mocker.patch.object(fetcher._client.cluster, "leader_for_partition", return_value=0)
-    manager.run(fetcher._reset_offsets_async, 1000)
+    net.run(fetcher._reset_offsets_async, 1000)
 
     assert not subscription_state.is_offset_reset_needed(tp)
     assert not subscription_state.is_fetchable(tp) # because tp is paused
@@ -1104,7 +1104,7 @@ async def fake_send(node_id, timestamps_and_epochs):
     assert subscription_state.position(tp) == OffsetAndMetadata(10, '', -1)
 
 
-def test_reset_offsets_paused_without_valid(subscription_state, client, manager, net, mocker):
+def test_reset_offsets_paused_without_valid(subscription_state, client, net, mocker):
     fetcher = Fetcher(client, subscription_state)
     tp = TopicPartition('foo', 0)
     subscription_state.assign_from_user([tp])
@@ -1117,7 +1117,7 @@ async def fake_send(node_id, timestamps_and_epochs):
     mocker.patch.object(fetcher._client, 'ready', return_value=True)
     mocker.patch.object(fetcher, '_send_list_offsets_request', side_effect=fake_send)
     mocker.patch.object(fetcher._client.cluster, "leader_for_partition", return_value=0)
-    manager.run(fetcher._reset_offsets_async, 1000)
+    net.run(fetcher._reset_offsets_async, 1000)
 
     assert not subscription_state.is_offset_reset_needed(tp)
     assert not subscription_state.is_fetchable(tp) # because tp is paused
@@ -1262,13 +1262,13 @@ async def fake_send(ts):
         result = fetcher.offsets_by_times(timestamps, timeout_ms=10000)
         assert result == {tp0: offset0, tp1: offset1}
 
-    def test_timeout_raises(self, fetcher, mocker):
+    def test_timeout_raises(self, fetcher, net, mocker):
         tp = TopicPartition('test', 0)
         timestamps = {tp: 1000}
 
         # Awaits a future that never completes
         async def fake_send(ts):
-            await fetcher._manager.create_future()
+            await net.create_future()
         mocker.patch.object(fetcher, '_send_list_offsets_requests', side_effect=fake_send)
 
         with pytest.raises(Errors.KafkaTimeoutError):
@@ -1428,7 +1428,7 @@ def test_maybe_validate_positions_no_op_when_epoch_current(fetcher, mocker):
 
 
 def test_validate_offsets_async_clears_validation_on_matching_epoch(
-        fetcher, manager, mocker):
+        fetcher, net, mocker):
     """OffsetForLeaderEpoch response with matching epoch and end_offset >=
     position.offset clears awaiting_validation."""
     tp = TopicPartition('foobar', 0)
@@ -1450,7 +1450,7 @@ async def fake_send(node_id, partitions_to_positions):
     mocker.patch.object(fetcher, '_send_offset_for_leader_epoch_request',
                         side_effect=fake_send)
 
-    manager.run(fetcher._validate_offsets_async, 1000)
+    net.run(fetcher._validate_offsets_async, 1000)
 
     assert not fetcher._subscriptions.assignment[tp].awaiting_validation
     assert fetcher._subscriptions.assignment[tp].position.leader_epoch == 5
@@ -1618,7 +1618,7 @@ def test_validate_offsets_async_undefined_raises_when_no_reset_policy(
 
 
 def test_validate_offsets_async_retries_on_fenced_epoch(
-        fetcher, manager, mocker):
+        fetcher, net, mocker):
     """FENCED_LEADER_EPOCH on the validation RPC sleeps retry_backoff_ms
     and retries within a single _validate_offsets_async invocation."""
     tp = TopicPartition('foobar', 0)
@@ -1648,7 +1648,7 @@ async def fake_send(node_id, partitions_to_positions):
                         side_effect=fake_send)
 
     start = time.monotonic()
-    manager.run(fetcher._validate_offsets_async, 1000)
+    net.run(fetcher._validate_offsets_async, 1000)
     elapsed = time.monotonic() - start
 
     assert call_count[0] == 2, (
diff --git a/test/consumer/test_fetcher_mock_broker.py b/test/consumer/test_fetcher_mock_broker.py
index f1d733820..ff0210bb7 100644
--- a/test/consumer/test_fetcher_mock_broker.py
+++ b/test/consumer/test_fetcher_mock_broker.py
@@ -135,7 +135,7 @@ def handler(api_key, api_version, correlation_id, request_bytes):
         fetcher.maybe_validate_positions()
         assert fetcher._subscriptions.assignment[tp].awaiting_validation
 
-        manager.run(fetcher._validate_offsets_async, 1000)
+        net.run(fetcher._validate_offsets_async, 1000)
 
         assert 'version' in captured, 'OffsetForLeaderEpochRequest never reached broker'
         assert not fetcher._subscriptions.assignment[tp].awaiting_validation
@@ -177,7 +177,7 @@ def handler(api_key, api_version, correlation_id, request_bytes):
         # Round 1: poll marks + drives validation to completion.
         fetcher.maybe_validate_positions()
         assert fetcher._subscriptions.assignment[tp].awaiting_validation
-        manager.run(fetcher._validate_offsets_async, 1000)
+        net.run(fetcher._validate_offsets_async, 1000)
 
         assert fetcher._cached_log_truncation is None
         assert not fetcher._subscriptions.assignment[tp].awaiting_validation
@@ -218,7 +218,7 @@ def test_seek_forces_revalidation_of_new_position(
 
         # Validate the first position against the current leader (epoch 5).
         fetcher.maybe_validate_positions()
-        manager.run(fetcher._validate_offsets_async, 1000)
+        net.run(fetcher._validate_offsets_async, 1000)
         assert not fetcher._subscriptions.assignment[tp].awaiting_validation
 
         # User seeks to a different offset whose record epoch (4) is below the
@@ -242,7 +242,7 @@ def test_diverged_seeks_to_endpoint_with_policy(
         broker.respond(OffsetForLeaderEpochRequest,
                        _ofle_response(error_code=0, leader_epoch=3, end_offset=80))
 
-        manager.run(fetcher._validate_offsets_async, 1000)
+        net.run(fetcher._validate_offsets_async, 1000)
 
         assert fetcher._cached_log_truncation is None
         assert not fetcher._subscriptions.assignment[tp].awaiting_reset
@@ -269,7 +269,7 @@ def test_diverged_raises_when_no_reset_policy(
         broker.respond(OffsetForLeaderEpochRequest,
                        _ofle_response(error_code=0, leader_epoch=3, end_offset=80))
 
-        manager.run(fetcher._validate_offsets_async, 1000)
+        net.run(fetcher._validate_offsets_async, 1000)
 
         with pytest.raises(Errors.LogTruncationError) as exc_info:
             fetcher.validate_offsets_if_needed()
@@ -316,7 +316,7 @@ def test_undefined_response_raises_when_no_reset_policy(
         broker.respond(OffsetForLeaderEpochRequest,
                        _ofle_response(error_code=0, leader_epoch=-1, end_offset=-1))
 
-        manager.run(fetcher._validate_offsets_async, 1000)
+        net.run(fetcher._validate_offsets_async, 1000)
 
         with pytest.raises(Errors.LogTruncationError) as exc_info:
             fetcher.validate_offsets_if_needed()
@@ -370,7 +370,7 @@ def test_validation_retries_on_fenced_epoch_response(
             error_code=0, leader_epoch=5, end_offset=100))
 
         start = time.monotonic()
-        manager.run(fetcher._validate_offsets_async, 2000)
+        net.run(fetcher._validate_offsets_async, 2000)
         elapsed = time.monotonic() - start
 
         assert not fetcher._subscriptions.assignment[tp].awaiting_validation
diff --git a/test/integration/test_sasl_integration.py b/test/integration/test_sasl_integration.py
index 886383ebb..a79cfc139 100644
--- a/test/integration/test_sasl_integration.py
+++ b/test/integration/test_sasl_integration.py
@@ -77,7 +77,7 @@ def test_client(request, sasl_kafka):
 
     # Low-level SASL round-trip via KafkaConnectionManager directly (no compat
     # shim, no poll()): the manager owns + auto-starts/closes its net, so this
-    # is just bootstrap + manager.run(coro) and runs on any backend.
+    # is just bootstrap + net.run(coro) and runs on any backend.
     manager = KafkaConnectionManager(**client_params(sasl_kafka, 'client'))
     try:
         manager.bootstrap(timeout_ms=5000)  # auto-starts the owned net
diff --git a/test/producer/test_sender.py b/test/producer/test_sender.py
index ca4edbc7b..5f361daa7 100644
--- a/test/producer/test_sender.py
+++ b/test/producer/test_sender.py
@@ -49,9 +49,9 @@ def notify(self):
 
 def _drive(sender, coro_method):
     """Run one of the sender's coroutine methods to completion on the test
-    selector (no IO thread is started in unit tests, so manager.run drives
+    selector (no IO thread is started in unit tests, so net.run drives
     the loop on the calling thread)."""
-    return sender._manager.run(coro_method)
+    return sender._net.run(coro_method)
 
 
 def _partition_response(error_cls=None, **kwargs):

From df9f70e9121557a238a5ebe4dabcef1e79f675bb Mon Sep 17 00:00:00 2001
From: Dana Powers <dana.powers@gmail.com>
Date: Tue, 28 Jul 2026 10:30:23 -0700
Subject: [PATCH 2/7] move manager.run tests -> backend abstract

---
 test/net/backend/test_abstract.py | 82 ++++++++++++++++++++++++++++++
 test/net/test_manager.py          | 83 -------------------------------
 2 files changed, 82 insertions(+), 83 deletions(-)

diff --git a/test/net/backend/test_abstract.py b/test/net/backend/test_abstract.py
index 5b177da3e..e8fb599e3 100644
--- a/test/net/backend/test_abstract.py
+++ b/test/net/backend/test_abstract.py
@@ -11,6 +11,7 @@
 
 import pytest
 
+import kafka.errors as Errors
 from kafka.net.backend.abstract import (
     NetBackend, NetTransport, resolve_backend, register_backend, _BACKENDS,
 )
@@ -221,3 +222,84 @@ def test_kafkanetclient_honors_explicit_instance(self):
         c = KafkaNetClient(net=sel, bootstrap_servers='localhost:9092')
         assert c._net is sel
         sel.close()
+
+
+class TestNetBackendRun:
+    def test_run_function(self, net):
+        def test_coro():
+            return 42
+        assert net.run(test_coro) == 42
+
+    def test_run_async_coro_function(self, net):
+        async def test_coro():
+            return 100
+        assert net.run(test_coro) == 100
+
+    def test_run_async_coro_with_args(self, net):
+        async def test_coro(foo):
+            return foo
+        assert net.run(test_coro, 123) == 123
+
+    def test_run_async_coro(self, net):
+        async def test_coro():
+            return 49
+        assert net.run(test_coro()) == 49
+
+    def test_run_async_chain(self, net):
+        async def test_coro_foo():
+            return 'foo!'
+        async def test_coro_bar():
+            return await test_coro_foo()
+        assert net.run(test_coro_bar()) == 'foo!'
+
+    def test_run_raises(self, net):
+        async def bad_coro():
+            raise ValueError('bad_coro')
+        with pytest.raises(ValueError, match='bad_coro'):
+            net.run(bad_coro)
+
+    def test_call_soon_does_not_raise(self, net):
+        async def bad_coro():
+            raise ValueError('bad_coro')
+        future = net.call_soon_with_future(bad_coro)
+        assert not future.is_done
+        net.poll(future=future)
+        assert future.failed()
+        assert isinstance(future.exception, ValueError)
+        assert future.exception.args[0] == 'bad_coro'
+
+    def test_run_survives_gc_during_poll(self, net, monkeypatch):
+        """Regression: an aggressive gc.collect() between _poll_once
+        iterations must not close orphan-cycle suspended coroutines and mask
+        the real result with GeneratorExit.
+
+        The wrapper Future returned by net.call_soon_with_future pins its Task via
+        a no-op callback so the cycle (Future_yielded <-> _poll_once cb <->
+        Task <-> coroutine <-> Future_yielded) has an external reference
+        for as long as the wrapper Future is pending.
+        """
+        import gc
+        from kafka.net.backend.selector import NetworkSelector
+
+        # Force a GC cycle on every _poll_once entry to deterministically
+        # trigger the orphan-collection race that was masking timeouts in CI.
+        orig_poll_once = NetworkSelector._poll_once
+
+        def aggressive_poll_once(self, timeout=None):
+            gc.collect()
+            return orig_poll_once(self, timeout)
+        monkeypatch.setattr(NetworkSelector, '_poll_once', aggressive_poll_once)
+
+        async def hangs_then_times_out():
+            # Awaits a bare (loop-awaitable) future that nothing references
+            # externally -- exactly the orphan-cycle shape that CPython's gc
+            # collects.
+            await net.create_future()
+
+        # wait_for should fail with KafkaTimeoutError, not GeneratorExit.
+        async def waiter():
+            inner = net.call_soon_with_future(hangs_then_times_out)
+            return await net.await_for(inner, timeout_ms=50, raise_error=True)
+
+        with pytest.raises(Errors.KafkaTimeoutError):
+            net.run(waiter)
diff --git a/test/net/test_manager.py b/test/net/test_manager.py
index b70ebe58d..616b4d634 100644
--- a/test/net/test_manager.py
+++ b/test/net/test_manager.py
@@ -1,11 +1,9 @@
 import socket
 import threading
-import time
 from unittest.mock import MagicMock, patch
 
 import pytest
 
-from kafka.cluster import ClusterMetadata
 from kafka.future import Future
 from kafka.net.backend.selector import NetworkSelector
 from kafka.net.connection import KafkaConnection
@@ -460,84 +458,3 @@ def test_connection_made_refuses_closed_connection(self, net):
         assert conn.closed
         with pytest.raises(Errors.KafkaConnectionError):
             conn.connection_made(MagicMock())
-
-
-class TestKafkaConnectionManagerRun:
-    def test_run_function(self, manager):
-        def test_coro():
-            return 42
-        assert manager.run(test_coro) == 42
-
-    def test_run_async_coro_function(self, manager):
-        async def test_coro():
-            return 100
-        assert manager.run(test_coro) == 100
-
-    def test_run_async_coro_with_args(self, manager):
-        async def test_coro(foo):
-            return foo
-        assert manager.run(test_coro, 123) == 123
-
-    def test_run_async_coro(self, manager):
-        async def test_coro():
-            return 49
-        assert manager.run(test_coro()) == 49
-
-    def test_run_async_chain(self, manager):
-        async def test_coro_foo():
-            return 'foo!'
-        async def test_coro_bar():
-            return await test_coro_foo()
-        assert manager.run(test_coro_bar()) == 'foo!'
-
-    def test_run_raises(self, manager):
-        async def bad_coro():
-            raise ValueError('bad_coro')
-        with pytest.raises(ValueError, match='bad_coro'):
-            manager.run(bad_coro)
-
-    def test_call_soon_does_not_raise(self, manager, net):
-        async def bad_coro():
-            raise ValueError('bad_coro')
-        future = manager.call_soon(bad_coro)
-        assert not future.is_done
-        net.poll(future=future)
-        assert future.failed()
-        assert isinstance(future.exception, ValueError)
-        assert future.exception.args[0] == 'bad_coro'
-
-    def test_run_survives_gc_during_poll(self, manager, monkeypatch):
-        """Regression: an aggressive gc.collect() between _poll_once
-        iterations must not close orphan-cycle suspended coroutines and mask
-        the real result with GeneratorExit.
-
-        The wrapper Future returned by manager.call_soon pins its Task via
-        a no-op callback so the cycle (Future_yielded <-> _poll_once cb <->
-        Task <-> coroutine <-> Future_yielded) has an external reference
-        for as long as the wrapper Future is pending.
-        """
-        import gc
-        from kafka.net.backend.selector import NetworkSelector
-
-        # Force a GC cycle on every _poll_once entry to deterministically
-        # trigger the orphan-collection race that was masking timeouts in CI.
-        orig_poll_once = NetworkSelector._poll_once
-
-        def aggressive_poll_once(self, timeout=None):
-            gc.collect()
-            return orig_poll_once(self, timeout)
-        monkeypatch.setattr(NetworkSelector, '_poll_once', aggressive_poll_once)
-
-        async def hangs_then_times_out():
-            # Awaits a bare (loop-awaitable) future that nothing references
-            # externally -- exactly the orphan-cycle shape that CPython's gc
-            # collects.
-            await manager.create_future()
-
-        # wait_for should fail with KafkaTimeoutError, not GeneratorExit.
-        async def waiter():
-            inner = manager.call_soon(hangs_then_times_out)
-            return await manager._net.await_for(inner, timeout_ms=50, raise_error=True)
-
-        with pytest.raises(Errors.KafkaTimeoutError):
-            manager.run(waiter)

From 59e7cabdebcf205f5ce249008a1e39d0609d27ce Mon Sep 17 00:00:00 2001
From: Dana Powers <dana.powers@gmail.com>
Date: Tue, 28 Jul 2026 10:57:34 -0700
Subject: [PATCH 3/7] drop manager.run

---
 kafka/net/manager.py | 17 -----------------
 1 file changed, 17 deletions(-)

diff --git a/kafka/net/manager.py b/kafka/net/manager.py
index c607f75af..e18aed7b4 100644
--- a/kafka/net/manager.py
+++ b/kafka/net/manager.py
@@ -447,20 +447,3 @@ def call_soon(self, coro, *args):
         Returns: Future
         """
         return self._net.call_soon_with_future(coro, *args)
-
-    def run(self, coro, *args, timeout_ms=None):
-        """Schedules coro on the event loop, blocks until complete, returns value or raises.
-
-        If an IO thread is running (via start()), the caller thread blocks on
-        a cross-thread Event while the coroutine runs on the IO thread. Safe
-        to call concurrently from multiple caller threads.
-
-        If no IO thread is running, falls back to driving the loop on the
-        caller thread (legacy behavior).
-
-        The blocking wait is bounded by ``timeout_ms`` (or the backend's
-        ``default_api_timeout_ms`` when None) plus a grace margin; see
-        :meth:`NetworkSelector.run`.
-        """
-        self._maybe_start()
-        return self._net.run(coro, *args, timeout_ms=timeout_ms)

From 64573bb915dc506ff5793222adc933aedf945d54 Mon Sep 17 00:00:00 2001
From: Dana Powers <dana.powers@gmail.com>
Date: Tue, 28 Jul 2026 11:01:26 -0700
Subject: [PATCH 4/7] Move selector specific tests to test_selector

---
 test/net/backend/test_abstract.py | 46 -----------------------------
 test/net/backend/test_selector.py | 48 +++++++++++++++++++++++++++++++
 2 files changed, 48 insertions(+), 46 deletions(-)

diff --git a/test/net/backend/test_abstract.py b/test/net/backend/test_abstract.py
index e8fb599e3..990eaee71 100644
--- a/test/net/backend/test_abstract.py
+++ b/test/net/backend/test_abstract.py
@@ -257,49 +257,3 @@ async def bad_coro():
             raise ValueError('bad_coro')
         with pytest.raises(ValueError, match='bad_coro'):
             net.run(bad_coro)
-
-    def test_call_soon_does_not_raise(self, net):
-        async def bad_coro():
-            raise ValueError('bad_coro')
-        future = net.call_soon_with_future(bad_coro)
-        assert not future.is_done
-        net.poll(future=future)
-        assert future.failed()
-        assert isinstance(future.exception, ValueError)
-        assert future.exception.args[0] == 'bad_coro'
-
-    def test_run_survives_gc_during_poll(self, net, monkeypatch):
-        """Regression: an aggressive gc.collect() between _poll_once
-        iterations must not close orphan-cycle suspended coroutines and mask
-        the real result with GeneratorExit.
-
-        The wrapper Future returned by net.call_soon_with_future pins its Task via
-        a no-op callback so the cycle (Future_yielded <-> _poll_once cb <->
-        Task <-> coroutine <-> Future_yielded) has an external reference
-        for as long as the wrapper Future is pending.
-        """
-        import gc
-        from kafka.net.backend.selector import NetworkSelector
-
-        # Force a GC cycle on every _poll_once entry to deterministically
-        # trigger the orphan-collection race that was masking timeouts in CI.
-        orig_poll_once = NetworkSelector._poll_once
-
-        def aggressive_poll_once(self, timeout=None):
-            gc.collect()
-            return orig_poll_once(self, timeout)
-        monkeypatch.setattr(NetworkSelector, '_poll_once', aggressive_poll_once)
-
-        async def hangs_then_times_out():
-            # Awaits a bare (loop-awaitable) future that nothing references
-            # externally -- exactly the orphan-cycle shape that CPython's gc
-            # collects.
-            await net.create_future()
-
-        # wait_for should fail with KafkaTimeoutError, not GeneratorExit.
-        async def waiter():
-            inner = net.call_soon_with_future(hangs_then_times_out)
-            return await net.await_for(inner, timeout_ms=50, raise_error=True)
-
-        with pytest.raises(Errors.KafkaTimeoutError):
-            net.run(waiter)
diff --git a/test/net/backend/test_selector.py b/test/net/backend/test_selector.py
index 730e72c65..3bcf91020 100644
--- a/test/net/backend/test_selector.py
+++ b/test/net/backend/test_selector.py
@@ -1259,3 +1259,51 @@ def test_socket_options_applied(self, net):
             call(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
         ])
         assert sock.setsockopt.call_count == 2
+
+
+class TestMisc:
+    def test_call_soon_does_not_raise(self, net):
+        async def bad_coro():
+            raise ValueError('bad_coro')
+        future = net.call_soon_with_future(bad_coro)
+        assert not future.is_done
+        net.poll(future=future)
+        assert future.failed()
+        assert isinstance(future.exception, ValueError)
+        assert future.exception.args[0] == 'bad_coro'
+
+    def test_run_survives_gc_during_poll(self, net, monkeypatch):
+        """Regression: an aggressive gc.collect() between _poll_once
+        iterations must not close orphan-cycle suspended coroutines and mask
+        the real result with GeneratorExit.
+
+        The wrapper Future returned by net.call_soon_with_future pins its Task via
+        a no-op callback so the cycle (Future_yielded <-> _poll_once cb <->
+        Task <-> coroutine <-> Future_yielded) has an external reference
+        for as long as the wrapper Future is pending.
+        """
+        import gc
+        from kafka.net.backend.selector import NetworkSelector
+
+        # Force a GC cycle on every _poll_once entry to deterministically
+        # trigger the orphan-collection race that was masking timeouts in CI.
+        orig_poll_once = NetworkSelector._poll_once
+
+        def aggressive_poll_once(self, timeout=None):
+            gc.collect()
+            return orig_poll_once(self, timeout)
+        monkeypatch.setattr(NetworkSelector, '_poll_once', aggressive_poll_once)
+
+        async def hangs_then_times_out():
+            # Awaits a bare (loop-awaitable) future that nothing references
+            # externally -- exactly the orphan-cycle shape that CPython's gc
+            # collects.
+            await net.create_future()
+
+        # wait_for should fail with KafkaTimeoutError, not GeneratorExit.
+        async def waiter():
+            inner = net.call_soon_with_future(hangs_then_times_out)
+            return await net.await_for(inner, timeout_ms=50, raise_error=True)
+
+        with pytest.raises(Errors.KafkaTimeoutError):
+            net.run(waiter)

From d535140af29c16a4021919464c46b3f241cd8153 Mon Sep 17 00:00:00 2001
From: Dana Powers <dana.powers@gmail.com>
Date: Tue, 28 Jul 2026 11:01:56 -0700
Subject: [PATCH 5/7] Move manager.create_future -> net.create_future

---
 kafka/cluster.py              |  2 +-
 kafka/coordinator/base.py     |  2 +-
 kafka/net/manager.py          | 10 +---------
 test/consumer/test_fetcher.py |  2 +-
 4 files changed, 4 insertions(+), 12 deletions(-)

diff --git a/kafka/cluster.py b/kafka/cluster.py
index 1e95f9201..4cf3cc6b7 100644
--- a/kafka/cluster.py
+++ b/kafka/cluster.py
@@ -150,7 +150,7 @@ async def refresh_metadata(self, node_id=None):
             log.debug('Metadata refresh already in flight; awaiting existing')
             await self._refresh_future
             return
-        self._refresh_future = self._manager.create_future()
+        self._refresh_future = self._manager._net.create_future()
         try:
             await self._do_refresh_metadata(node_id)
         except Exception as exc:
diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py
index d19058370..622323a42 100644
--- a/kafka/coordinator/base.py
+++ b/kafka/coordinator/base.py
@@ -357,7 +357,7 @@ async def ensure_coordinator_ready_async(self, timeout_ms=None):
                 if maybe_coordinator_id is None:
                     # Pre-failed sibling of lookup_coordinator(); consumed via
                     # wait_for on the loop, so mint it from the backend too.
-                    future = self._manager.create_future().failure(Errors.NodeNotReadyError('coordinator'))
+                    future = self._net.create_future().failure(Errors.NodeNotReadyError('coordinator'))
                 else:
                     self.coordinator_id = maybe_coordinator_id
                     return not timer.expired
diff --git a/kafka/net/manager.py b/kafka/net/manager.py
index e18aed7b4..1b24d6c52 100644
--- a/kafka/net/manager.py
+++ b/kafka/net/manager.py
@@ -321,7 +321,7 @@ def send(self, request, node_id=None, request_timeout_ms=None):
         except Errors.NodeNotReadyError as e:
             # Pre-failed sibling of send_request()'s create_future(); awaited
             # on the loop by the same caller, so mint it from the backend too.
-            return self.create_future().failure(e)
+            return self._net.create_future().failure(e)
         else:
             return conn.send_request(request, request_timeout_ms=request_timeout_ms)
 
@@ -431,14 +431,6 @@ def close(self, node_id=None, timeout_ms=None):
             if self._owns_net and not self._net.on_io_thread():
                 self._net.close()
 
-    def create_future(self):
-        """Create a Future suitable for awaiting on the underlying loop.
-
-        Forwards to the backend so loop coroutines get the backend's native
-        awaitable type. See ``NetworkSelector.create_future``.
-        """
-        return self._net.create_future()
-
     def call_soon(self, coro, *args):
         """Accepts a coroutine / awaitable / function and schedules it on the event loop.
 
diff --git a/test/consumer/test_fetcher.py b/test/consumer/test_fetcher.py
index e70cb7e82..86141f268 100644
--- a/test/consumer/test_fetcher.py
+++ b/test/consumer/test_fetcher.py
@@ -370,7 +370,7 @@ def test__send_list_offsets_requests_multiple_nodes(fetcher, manager, net, mocke
 
     send_futures = []
     async def fake_send(node_id, timestamps):
-        f = fetcher._manager.create_future()  # awaited below
+        f = net.create_future()  # awaited below
         send_futures.append((node_id, timestamps, f))
         return await f
     mocked_send = mocker.patch.object(fetcher, "_send_list_offsets_request", side_effect=fake_send)

From 10c20870d98a10174cb2bc78b78343a7d86e0287 Mon Sep 17 00:00:00 2001
From: Dana Powers <dana.powers@gmail.com>
Date: Tue, 28 Jul 2026 11:05:46 -0700
Subject: [PATCH 6/7] drop create_future manager test

---
 test/net/test_manager.py | 6 ------
 1 file changed, 6 deletions(-)

diff --git a/test/net/test_manager.py b/test/net/test_manager.py
index 616b4d634..ccd468d32 100644
--- a/test/net/test_manager.py
+++ b/test/net/test_manager.py
@@ -69,12 +69,6 @@ def test_api_versions(self, net):
         m = KafkaConnectionManager(net, api_version=(1, 0))
         assert m.broker_version_data == BrokerVersionData((1, 0))
 
-    def test_create_future_forwards_to_backend(self, net):
-        m = KafkaConnectionManager(net)
-        fut = m.create_future()
-        assert isinstance(fut, Future)
-        assert not fut.is_done
-
     def test_client_dns_lookup_default(self, net):
         m = KafkaConnectionManager(net)
         assert m.config['client_dns_lookup'] == 'use_all_dns_ips'

From 3e35327c38ada591f3811d593ab363740db41c20 Mon Sep 17 00:00:00 2001
From: Dana Powers <dana.powers@gmail.com>
Date: Tue, 28 Jul 2026 13:30:13 -0700
Subject: [PATCH 7/7] fixup fetcher test

---
 test/consumer/test_fetcher_mock_broker.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/test/consumer/test_fetcher_mock_broker.py b/test/consumer/test_fetcher_mock_broker.py
index ff0210bb7..6c559d4c7 100644
--- a/test/consumer/test_fetcher_mock_broker.py
+++ b/test/consumer/test_fetcher_mock_broker.py
@@ -354,7 +354,7 @@ def test_fenced_epoch_on_fetch_marks_validation_then_succeeds(
         assert not fetcher._subscriptions.assignment[tp].awaiting_validation
 
     def test_validation_retries_on_fenced_epoch_response(
-            self, broker, manager, fetcher):
+            self, broker, net, fetcher):
         """FENCED_LEADER_EPOCH in the OffsetForLeaderEpoch response itself
         is retried after ``retry_backoff_ms`` within a single
         ``_validate_offsets_async`` invocation."""