Use net not manager for .run / .create_future

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)
 
 

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,15 +213,15 @@ 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)
         return conn.broker_version_data
 
     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):

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):

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)
 
 

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)
 

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')

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)
 

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):

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__

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: