diff --git a/kafka/admin/_cluster.py b/kafka/admin/_cluster.py
index 978e876c4..7183228b6 100644
--- a/kafka/admin/_cluster.py
+++ b/kafka/admin/_cluster.py
@@ -349,7 +349,8 @@ def update_features(self, feature_updates, validate_only=False, timeout_ms=60000
             dict of {feature_name: 'OK' | error message}
         """
         return self._manager.run(self._async_update_features,
-                                 feature_updates, validate_only, timeout_ms)
+                                 feature_updates, validate_only, timeout_ms,
+                                 timeout_ms=timeout_ms)
 
 
 class UpdateFeatureType(EnumHelper, IntEnum):
diff --git a/kafka/admin/_partitions.py b/kafka/admin/_partitions.py
index 7ae7e2865..7930a246b 100644
--- a/kafka/admin/_partitions.py
+++ b/kafka/admin/_partitions.py
@@ -212,7 +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)
+        return self._manager.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 [
@@ -386,7 +388,8 @@ def list_partition_reassignments(self, topic_partitions=None, timeout_ms=None):
             ``'removing_replicas'`` (each a list of broker IDs).
         """
         return self._manager.run(
-            self._async_list_partition_reassignments, topic_partitions, timeout_ms)
+            self._async_list_partition_reassignments, topic_partitions, timeout_ms,
+            timeout_ms=timeout_ms)
 
     async def _async_describe_topic_partitions(self, topics, response_partition_limit, cursor):
         _Topic = DescribeTopicPartitionsRequest.TopicRequest
@@ -573,7 +576,8 @@ def list_partition_offsets(self, topic_partition_specs, isolation_level=Isolatio
                 of ListOffsetsRequest compatible with the requested specs.
         """
         return self._manager.run(
-            self._async_list_partition_offsets, topic_partition_specs, isolation_level, timeout_ms)
+            self._async_list_partition_offsets, topic_partition_specs, isolation_level, timeout_ms,
+            timeout_ms=timeout_ms)
 
 
 class NewPartitions:
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py
index 14a0f9ba4..4b1ec1452 100644
--- a/kafka/consumer/fetcher.py
+++ b/kafka/consumer/fetcher.py
@@ -93,6 +93,7 @@ class Fetcher:
         'metrics': None,
         'metric_group_prefix': 'consumer',
         'request_timeout_ms': 30000,
+        'default_api_timeout_ms': 60000,
         'retry_backoff_ms': 100,
         'enable_incremental_fetch_sessions': True,
         'isolation_level': 'read_uncommitted',
@@ -258,7 +259,7 @@ def _wake(_):
             fut.add_both(_wake)
 
         try:
-            self._net.run(self._manager.wait_for, wakeup, timeout_ms)
+            self._net.run(self._manager.wait_for, wakeup, timeout_ms, timeout_ms=timeout_ms)
         except Errors.KafkaTimeoutError:
             pass
 
@@ -395,7 +396,8 @@ def offsets_by_times(self, timestamps, timeout_ms=None):
             timestamps: {TopicPartition: int} dict with timestamps to fetch
                 offsets by. -1 for the latest available, -2 for the earliest
                 available. Otherwise timestamp is treated as epoch milliseconds.
-            timeout_ms (int, optional): The maximum time in milliseconds to block.
+            timeout_ms (int, optional): Maximum time in milliseconds to block.
+                Defaults to default_api_timeout_ms.
 
         Returns:
             {TopicPartition: OffsetAndTimestamp or None}: Mapping of partition to
@@ -404,9 +406,12 @@ def offsets_by_times(self, timestamps, timeout_ms=None):
                 will be None.
 
         Raises:
-            KafkaTimeoutError if timeout_ms provided
+            KafkaTimeoutError: if not completed within timeout_ms (default
+                default_api_timeout_ms).
         """
-        offsets = self._net.run(self._fetch_offsets_by_times_async, timestamps, timeout_ms)
+        if timeout_ms is None:
+            timeout_ms = self.config['default_api_timeout_ms']
+        offsets = self._net.run(self._fetch_offsets_by_times_async, timestamps, timeout_ms, timeout_ms=timeout_ms)
         for tp in timestamps:
             if tp not in offsets:
                 offsets[tp] = None
@@ -481,13 +486,15 @@ def beginning_offsets(self, partitions, timeout_ms=None):
 
         Arguments:
             partitions ([TopicPartition]): List of partitions for list offsets.
-            timeout_ms (int, optional): The maximum time in milliseconds to block.
+            timeout_ms (int, optional): Maximum time in milliseconds to block.
+                Defaults to default_api_timeout_ms.
 
         Returns:
             {TopicPartition: int}: Mapping of partition to retrieved offset.
 
         Raises:
-            KafkaTimeoutError if timeout_ms provided.
+            KafkaTimeoutError: if not completed within timeout_ms (default
+                default_api_timeout_ms).
         """
         return self.beginning_or_end_offset(
             partitions, OffsetSpec.EARLIEST, timeout_ms)
@@ -500,13 +507,15 @@ def end_offsets(self, partitions, timeout_ms=None):
 
         Arguments:
             partitions ([TopicPartition]): List of partitions for list offsets.
-            timeout_ms (int, optional): The maximum time in milliseconds to block.
+            timeout_ms (int, optional): Maximum time in milliseconds to block.
+                Defaults to default_api_timeout_ms.
 
         Returns:
             {TopicPartition: int}: Mapping of partition to retrieved offset.
 
         Raises:
-            KafkaTimeoutError if timeout_ms provided.
+            KafkaTimeoutError: if not completed within timeout_ms (default
+                default_api_timeout_ms).
         """
         return self.beginning_or_end_offset(
             partitions, OffsetSpec.LATEST, timeout_ms)
@@ -522,7 +531,8 @@ def beginning_or_end_offset(self, partitions, timestamp, timeout_ms=None):
             timestamp (int or OffsetSpec): OffsetSpec.LATEST (-1) for the latest
                 available, OffsetSpec.EARLIEST (-2) for the earliest available.
                 Otherwise timestamp is treated as epoch milliseconds.
-            timeout_ms (int, optional): The maximum time in milliseconds to block.
+            timeout_ms (int, optional): Maximum time in milliseconds to block.
+                Defaults to default_api_timeout_ms.
 
         Returns:
             {TopicPartition: int}: Mapping of partition to retrieved offset.
@@ -530,10 +540,13 @@ def beginning_or_end_offset(self, partitions, timestamp, timeout_ms=None):
         Raises:
             UnsupportedVersionError if broker does not support any compatible
                 ListOffsetsRequest api version.
-            KafkaTimeoutError if timeout_ms provided.
+            KafkaTimeoutError: if not completed within timeout_ms (default
+                default_api_timeout_ms).
         """
         timestamps = dict([(tp, timestamp) for tp in partitions])
-        offsets = self._net.run(self._fetch_offsets_by_times_async, timestamps, timeout_ms)
+        if timeout_ms is None:
+            timeout_ms = self.config['default_api_timeout_ms']
+        offsets = self._net.run(self._fetch_offsets_by_times_async, timestamps, timeout_ms, timeout_ms=timeout_ms)
         for tp in timestamps:
             offsets[tp] = offsets[tp].offset
         return offsets
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index becd80ac2..ac52f7f06 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -679,8 +679,8 @@ def commit(self, offsets=None, timeout_ms=None):
         message read if a consumer is restarted, the committed offset should be
         the next message your application should consume, i.e.: last_offset + 1.
 
-        Blocks until either the commit succeeds or an unrecoverable error is
-        encountered (in which case it is thrown to the caller).
+        Blocks until the commit succeeds, an unrecoverable error is encountered
+        (in which case it is thrown to the caller), or the timeout expires.
 
         Currently only supports kafka-topic offset storage (not zookeeper).
 
@@ -688,10 +688,14 @@ def commit(self, offsets=None, timeout_ms=None):
             offsets (dict, optional): {TopicPartition: OffsetAndMetadata} dict
                 to commit with the configured group_id. Defaults to currently
                 consumed offsets for all subscribed partitions.
+            timeout_ms (numeric, optional): Maximum time in milliseconds to
+                block. Defaults to default_api_timeout_ms.
 
         Raises:
             IncompatibleBrokerVersion: if broker version < 0.8.1
             IllegalStateError: if group_id is None
+            KafkaTimeoutError: if the commit does not complete within timeout_ms
+                (default default_api_timeout_ms).
         """
         if self.config['api_version'] < (0, 8, 1):
             raise Errors.IncompatibleBrokerVersion('Requires >= Kafka 0.8.1')
@@ -734,6 +738,8 @@ def committed(self, partition, metadata=False, timeout_ms=None):
             partition (TopicPartition): The partition to check.
             metadata (bool, optional): If True, return OffsetAndMetadata struct
                 instead of offset int. Default: False.
+            timeout_ms (numeric, optional): Maximum time in milliseconds to
+                block. Defaults to default_api_timeout_ms.
 
         Returns:
             The last committed offset (int or OffsetAndMetadata), or None if there was no prior commit.
@@ -742,7 +748,8 @@ def committed(self, partition, metadata=False, timeout_ms=None):
             IncompatibleBrokerVersion: if broker version < 0.8.1
             IllegalStateError: if group_id is None
             TypeError: if partition is not TopicPartition
-            KafkaTimeoutError: if timeout_ms provided
+            KafkaTimeoutError: if the offsets are not fetched within timeout_ms
+                (default default_api_timeout_ms).
             BrokerResponseError: if OffsetFetchRequest raises an error.
         """
         if self.config['api_version'] < (0, 8, 1):
@@ -756,31 +763,44 @@ def committed(self, partition, metadata=False, timeout_ms=None):
             return None
         return committed[partition] if metadata else committed[partition].offset
 
-    def _fetch_all_topic_metadata(self):
+    def _fetch_all_topic_metadata(self, timeout_ms=None):
         """A blocking call that fetches topic metadata for all topics in the
         cluster that the user is authorized to view.
         """
+        if timeout_ms is None:
+            timeout_ms = self.config['default_api_timeout_ms']
+        timer = Timer(timeout_ms)
         if self._cluster.metadata_refresh_in_progress:
             future = self._cluster.request_update()
-            self._net.run(self._manager.wait_for, future, None)
+            self._net.run(self._manager.wait_for, future, timer.timeout_ms, timeout_ms=timer.timeout_ms)
         stash = self._cluster.need_all_topic_metadata
-        self._cluster.need_all_topic_metadata = True
-        future = self._cluster.request_update()
-        self._net.run(self._manager.wait_for, future, None)
-        self._cluster.need_all_topic_metadata = stash
+        try:
+            self._cluster.need_all_topic_metadata = True
+            future = self._cluster.request_update()
+            self._net.run(self._manager.wait_for, future, timer.timeout_ms, timeout_ms=timer.timeout_ms)
+        finally:
+            self._cluster.need_all_topic_metadata = stash
 
-    def topics(self):
+    def topics(self, timeout_ms=None):
         """Get all topics the user is authorized to view.
         This will always issue a remote call to the cluster to fetch the latest
         information.
 
+        Arguments:
+            timeout_ms (numeric, optional): Maximum time in milliseconds to
+                block. Defaults to default_api_timeout_ms.
+
         Returns:
             set: topics
+
+        Raises:
+            KafkaTimeoutError: if topic metadata is not fetched within
+                timeout_ms (default default_api_timeout_ms).
         """
-        self._fetch_all_topic_metadata()
+        self._fetch_all_topic_metadata(timeout_ms=timeout_ms)
         return self._cluster.topics()
 
-    def partitions_for_topic(self, topic):
+    def partitions_for_topic(self, topic, timeout_ms=None):
         """This method first checks the local metadata cache for information
         about the topic. If the topic is not found (either because the topic
         does not exist, the user is not authorized to view the topic, or the
@@ -789,13 +809,19 @@ def partitions_for_topic(self, topic):
 
         Arguments:
             topic (str): Topic to check.
+            timeout_ms (numeric, optional): Maximum time in milliseconds to
+                block. Defaults to default_api_timeout_ms.
 
         Returns:
             set: Partition ids
+
+        Raises:
+            KafkaTimeoutError: if topic metadata is not fetched within
+                timeout_ms (default default_api_timeout_ms).
         """
         partitions = self._cluster.partitions_for_topic(topic)
         if partitions is None:
-            self._fetch_all_topic_metadata()
+            self._fetch_all_topic_metadata(timeout_ms=timeout_ms)
             partitions = self._cluster.partitions_for_topic(topic)
         return partitions or set()
 
@@ -1218,14 +1244,15 @@ def offsets_for_times(self, timestamps, timeout_ms=None):
         partition. ``None`` will also be returned for the partition if there
         are no messages in it.
 
-        Note: This method may block indefinitely if the partition does not exist
-            and no timeout_ms provided.
+        Note: This method blocks up to timeout_ms (default
+            default_api_timeout_ms) if the partition does not exist.
 
         Arguments:
             timestamps (dict): ``{TopicPartition: int}`` mapping from partition
                 to the timestamp to look up. Unit should be milliseconds since
                 beginning of the epoch (midnight Jan 1, 1970 (UTC))
             timeout_ms (int, optional): Milliseconds to block fetching offsets.
+                Defaults to default_api_timeout_ms.
 
         Returns:
             ``{TopicPartition: OffsetAndTimestamp}``: mapping from partition
@@ -1236,9 +1263,10 @@ def offsets_for_times(self, timestamps, timeout_ms=None):
             ValueError: If the target timestamp is negative
             UnsupportedVersionError: If the broker does not support looking
                 up the offsets by timestamp.
-            KafkaTimeoutError: If fetch failed in request_timeout_ms
+            KafkaTimeoutError: if the offsets are not fetched within timeout_ms
+                (default default_api_timeout_ms)
         """
-        timeout_ms = self.config['request_timeout_ms'] if timeout_ms is None else timeout_ms
+        timeout_ms = self.config['default_api_timeout_ms'] if timeout_ms is None else timeout_ms
         for tp, ts in timestamps.items():
             timestamps[tp] = int(ts)
             if ts < 0:
@@ -1253,13 +1281,14 @@ def beginning_offsets(self, partitions, timeout_ms=None):
         This method does not change the current consumer position of the
         partitions.
 
-        Note: This method may block indefinitely if the partition does not exist
-            and no timeout_ms provided.
+        Note: This method blocks up to timeout_ms (default
+            default_api_timeout_ms) if the partition does not exist.
 
         Arguments:
             partitions (list): List of TopicPartition instances to fetch
                 offsets for.
             timeout_ms (int, optional): Milliseconds to block fetching offsets.
+                Defaults to default_api_timeout_ms.
 
         Returns:
             ``{TopicPartition: int}``: The earliest available offsets for the
@@ -1268,9 +1297,10 @@ def beginning_offsets(self, partitions, timeout_ms=None):
         Raises:
             UnsupportedVersionError: If the broker does not support looking
                 up the offsets by timestamp.
-            KafkaTimeoutError: If fetch failed in timeout_ms.
+            KafkaTimeoutError: if not completed within timeout_ms (default
+                default_api_timeout_ms).
         """
-        timeout_ms = self.config['request_timeout_ms'] if timeout_ms is None else timeout_ms
+        timeout_ms = self.config['default_api_timeout_ms'] if timeout_ms is None else timeout_ms
         offsets = self._fetcher.beginning_offsets(partitions, timeout_ms)
         return offsets
 
@@ -1282,13 +1312,14 @@ def end_offsets(self, partitions, timeout_ms=None):
         This method does not change the current consumer position of the
         partitions.
 
-        Note: This method may block indefinitely if the partition does not exist
-            and no timeout_ms provided.
+        Note: This method blocks up to timeout_ms (default
+            default_api_timeout_ms) if the partition does not exist.
 
         Arguments:
             partitions (list): List of TopicPartition instances to fetch
                 offsets for.
             timeout_ms (int, optional): Milliseconds to block fetching offsets.
+                Defaults to default_api_timeout_ms.
 
         Returns:
             ``{TopicPartition: int}``: The end offsets for the given partitions.
@@ -1296,9 +1327,10 @@ def end_offsets(self, partitions, timeout_ms=None):
         Raises:
             UnsupportedVersionError: If the broker does not support looking
                 up the offsets by timestamp.
-            KafkaTimeoutError: If fetch failed in timeout_ms
+            KafkaTimeoutError: if not completed within timeout_ms (default
+                default_api_timeout_ms).
         """
-        timeout_ms = self.config['request_timeout_ms'] if timeout_ms is None else timeout_ms
+        timeout_ms = self.config['default_api_timeout_ms'] if timeout_ms is None else timeout_ms
         offsets = self._fetcher.end_offsets(partitions, timeout_ms)
         return offsets
 
diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py
index 8016e3e67..746f2ecad 100644
--- a/kafka/coordinator/base.py
+++ b/kafka/coordinator/base.py
@@ -106,6 +106,7 @@ class BaseCoordinator(ABC):
         'heartbeat_interval_ms': 3000,
         'max_poll_interval_ms': 300000,
         'request_timeout_ms': 30000,
+        'default_api_timeout_ms': 60000,
         'retry_backoff_ms': 100,
         'api_version': (0, 10, 1),
         'metrics': None,
@@ -331,11 +332,14 @@ def ensure_coordinator_ready(self, timeout_ms=None):
 
         Keyword Arguments:
             timeout_ms (numeric, optional): Maximum number of milliseconds to
-                block waiting to find coordinator. Default: None.
+                block waiting to find coordinator. Defaults to
+                default_api_timeout_ms.
 
         Returns: True is coordinator found before timeout_ms, else False
         """
-        return self._net.run(self.ensure_coordinator_ready_async, timeout_ms)
+        if timeout_ms is None:
+            timeout_ms = self.config['default_api_timeout_ms']
+        return self._net.run(self.ensure_coordinator_ready_async, timeout_ms, timeout_ms=timeout_ms)
 
     async def ensure_coordinator_ready_async(self, timeout_ms=None):
         """Async variant of :meth:`ensure_coordinator_ready`.
@@ -450,11 +454,13 @@ def ensure_active_group(self, timeout_ms=None):
 
         Keyword Arguments:
             timeout_ms (numeric, optional): Maximum number of milliseconds to
-                block waiting to join group. Default: None.
+                block waiting to join group. Defaults to default_api_timeout_ms.
 
         Returns: True if group initialized before timeout_ms, else False
         """
-        return self._net.run(self.ensure_active_group_async, timeout_ms)
+        if timeout_ms is None:
+            timeout_ms = self.config['default_api_timeout_ms']
+        return self._net.run(self.ensure_active_group_async, timeout_ms, timeout_ms=timeout_ms)
 
     async def ensure_active_group_async(self, timeout_ms=None):
         """Async variant of :meth:`ensure_active_group`."""
@@ -1048,7 +1054,9 @@ def is_dynamic_member(self):
 
     def maybe_leave_group(self, reason=None, timeout_ms=None):
         """Leave the current group and reset local generation/member_id."""
-        return self._net.run(self.maybe_leave_group_async, reason, timeout_ms)
+        if timeout_ms is None:
+            timeout_ms = self.config['default_api_timeout_ms']
+        return self._net.run(self.maybe_leave_group_async, reason, timeout_ms, timeout_ms=timeout_ms)
 
     async def maybe_leave_group_async(self, reason=None, timeout_ms=None):
         if not self._use_group_apis:
diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index b8d1378be..dfa26057f 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -622,7 +622,9 @@ def need_rejoin(self):
 
     def refresh_committed_offsets_if_needed(self, timeout_ms=None):
         """Fetch committed offsets for assigned partitions."""
-        return self._net.run(self.refresh_committed_offsets_if_needed_async, timeout_ms)
+        if timeout_ms is None:
+            timeout_ms = self.config['default_api_timeout_ms']
+        return self._net.run(self.refresh_committed_offsets_if_needed_async, timeout_ms, timeout_ms=timeout_ms)
 
     async def refresh_committed_offsets_if_needed_async(self, timeout_ms=None):
         missing_fetch_positions = set(self._subscription.missing_fetch_positions())
@@ -640,16 +642,21 @@ def fetch_committed_offsets(self, partitions, timeout_ms=None):
 
         Arguments:
             partitions (list of TopicPartition): partitions to fetch
+            timeout_ms (numeric, optional): Maximum time in milliseconds to
+                block. Defaults to default_api_timeout_ms.
 
         Returns:
             dict: {TopicPartition: OffsetAndMetadata}
 
         Raises:
-            KafkaTimeoutError if timeout_ms provided
+            KafkaTimeoutError: if the offsets are not fetched within timeout_ms
+                (default default_api_timeout_ms).
         """
         if not partitions:
             return {}
-        return self._net.run(self.fetch_committed_offsets_async, partitions, timeout_ms)
+        if timeout_ms is None:
+            timeout_ms = self.config['default_api_timeout_ms']
+        return self._net.run(self.fetch_committed_offsets_async, partitions, timeout_ms, timeout_ms=timeout_ms)
 
     async def fetch_committed_offsets_async(self, partitions, timeout_ms=None):
         """Async variant of :meth:`fetch_committed_offsets`."""
@@ -815,13 +822,17 @@ def _maybe_set_async_commit_fenced(exc):
     def commit_offsets_sync(self, offsets, timeout_ms=None):
         """Commit specific offsets synchronously.
 
-        This method will retry until the commit completes successfully or an
-        unrecoverable error is encountered.
+        This method retries until the commit completes successfully, an
+        unrecoverable error is encountered, or the timeout expires.
 
         Arguments:
             offsets (dict {TopicPartition: OffsetAndMetadata}): what to commit
+            timeout_ms (numeric, optional): Maximum time in milliseconds to
+                block. Defaults to default_api_timeout_ms.
 
-        Raises error on failure
+        Raises:
+            KafkaTimeoutError: if the commit does not complete within timeout_ms
+                (default default_api_timeout_ms).
         """
         if not self._use_offset_apis:
             raise Errors.UnsupportedVersionError('OffsetCommitRequest requires 0.8.1+ broker')
@@ -829,14 +840,19 @@ def commit_offsets_sync(self, offsets, timeout_ms=None):
            not all(map(lambda v: isinstance(v, OffsetAndMetadata), offsets.values())):
             raise TypeError('offsets must be dict[TopicPartition, OffsetAndMetadata]')
         self._invoke_completed_offset_commit_callbacks()
-        return self._net.run(self._commit_offsets_sync_async, offsets, timeout_ms)
+        # Resolve the all-in API deadline once (default default_api_timeout_ms)
+        # and bound both the coroutine's own Timer and the cross-thread run()
+        # backstop from the same value (#3121).
+        if timeout_ms is None:
+            timeout_ms = self.config['default_api_timeout_ms']
+        return self._net.run(self._commit_offsets_sync_async, offsets, timeout_ms,
+                             timeout_ms=timeout_ms)
 
     async def _commit_offsets_sync_async(self, offsets, timeout_ms=None):
         if not offsets:
             return
-        # Default to request_timeout_ms, matching offsets_by_times / _reset_offsets_async
         if timeout_ms is None:
-            timeout_ms = self.config['request_timeout_ms']
+            timeout_ms = self.config['default_api_timeout_ms']
         timer = Timer(timeout_ms)
         while True:
             await self.ensure_coordinator_ready_async(timeout_ms=timer.timeout_ms)
diff --git a/kafka/net/compat.py b/kafka/net/compat.py
index 2df88018b..19c03e223 100644
--- a/kafka/net/compat.py
+++ b/kafka/net/compat.py
@@ -6,6 +6,7 @@
 import kafka.errors as Errors
 from kafka.net.backend import resolve_backend
 from kafka.net.manager import KafkaConnectionManager
+from kafka.util import Timer
 
 
 log = logging.getLogger(__name__)
@@ -105,14 +106,15 @@ def get_broker_version(self, timeout_ms=None):
         return self._manager.broker_version
 
     def check_version(self, node_id=None, timeout_ms=10000):
+        timer = Timer(timeout_ms)
         if not self._manager.bootstrapped:
-            self._manager.bootstrap(timeout_ms)
+            self._manager.bootstrap(timer.timeout_ms)
         if node_id is None:
             return self._manager.broker_version
         async def _check_version(broker_id, timeout_ms):
             conn = await self._manager.get_connection(broker_id, timeout_ms=timeout_ms)
             return conn.broker_version
-        return self._net.run(_check_version, node_id, timeout_ms)
+        return self._net.run(_check_version, node_id, timer.timeout_ms, timeout_ms=timer.timeout_ms)
 
     # Request sending
 
diff --git a/kafka/net/manager.py b/kafka/net/manager.py
index 11ff89412..2dd824465 100644
--- a/kafka/net/manager.py
+++ b/kafka/net/manager.py
@@ -215,7 +215,7 @@ def _maybe_start(self):
 
     def bootstrap(self, timeout_ms=None, refresh=True):
         self._maybe_start()
-        self._net.run(self.bootstrap_async, timeout_ms, refresh)
+        self._net.run(self.bootstrap_async, timeout_ms, refresh, timeout_ms=timeout_ms)
 
     @property
     def bootstrapped(self):
diff --git a/kafka/net/selector.py b/kafka/net/selector.py
index 6f96f7916..8e9a621ca 100644
--- a/kafka/net/selector.py
+++ b/kafka/net/selector.py
@@ -333,6 +333,8 @@ def _bridge_deadline_secs(self, timeout_ms):
         coroutine's own (equal) deadline wins the race on a healthy loop.
         """
         op_ms = timeout_ms if timeout_ms is not None else self.config['default_api_timeout_ms']
+        if op_ms >= threading.TIMEOUT_MAX:
+            return None
         return (op_ms + self.config['bridge_grace_ms']) / 1000
 
     def _bridge_timeout(self, coro, timeout_ms):
@@ -364,7 +366,7 @@ def run(self, coro, *args, timeout_ms=None):
         deadline_secs = self._bridge_deadline_secs(timeout_ms)
         if self._io_thread is None:
             future = self.call_soon_with_future(coro, *args)
-            self.poll(timeout_ms=deadline_secs * 1000, future=future)
+            self.poll(future=future, timeout_ms=deadline_secs * 1000 if deadline_secs is not None else None)
             if not future.is_done:
                 raise self._bridge_timeout(coro, timeout_ms)
             if future.exception is not None:
diff --git a/test/consumer/test_consumer.py b/test/consumer/test_consumer.py
index 2026978c4..5125a8c7e 100644
--- a/test/consumer/test_consumer.py
+++ b/test/consumer/test_consumer.py
@@ -1,7 +1,10 @@
+from unittest.mock import PropertyMock
+
 import pytest
 
 from kafka import ConsumerGroupMetadata, KafkaConsumer, TopicPartition
-from kafka.errors import KafkaConfigurationError, IllegalStateError
+from kafka.errors import KafkaConfigurationError, IllegalStateError, KafkaTimeoutError
+from kafka.future import Future
 from kafka.util import Timer
 
 
@@ -26,6 +29,32 @@ def test_default_api_timeout_smaller_than_request_timeout_raises():
                       request_timeout_ms=70000, default_api_timeout_ms=60000)
 
 
+def test_fetch_all_topic_metadata_restores_flag_on_timeout(mocker):
+    """_fetch_all_topic_metadata must restore need_all_topic_metadata even when
+    the bounded metadata wait times out (issue #3121). Otherwise a timed-out
+    topics()/partitions_for_topic() would leave the consumer permanently
+    fetching all-topic metadata."""
+    consumer = KafkaConsumer(api_version=(0, 10, 0))
+    run_mock = mocker.patch.object(consumer._net, 'run')
+    try:
+        original = consumer._cluster.need_all_topic_metadata
+        # Skip the in-progress branch so the try/finally block is exercised.
+        mocker.patch.object(type(consumer._cluster), 'metadata_refresh_in_progress',
+                            new_callable=PropertyMock, return_value=False)
+        mocker.patch.object(consumer._cluster, 'request_update', return_value=Future())
+        run_mock.side_effect = KafkaTimeoutError('boom')
+
+        with pytest.raises(KafkaTimeoutError):
+            consumer.topics(timeout_ms=100)
+
+        assert consumer._cluster.need_all_topic_metadata == original
+    finally:
+        # Let close() no-op through the mocked run() instead of raising.
+        run_mock.side_effect = None
+        run_mock.return_value = None
+        consumer.close()
+
+
 def test_subscription_copy():
     consumer = KafkaConsumer('foo', api_version=(0, 10, 0))
     sub = consumer.subscription()
diff --git a/test/consumer/test_coordinator.py b/test/consumer/test_coordinator.py
index 78489896d..c67fb2d8c 100644
--- a/test/consumer/test_coordinator.py
+++ b/test/consumer/test_coordinator.py
@@ -557,6 +557,23 @@ async def fake_send_retry_then_success(_offsets):
     assert ret == 'fizzbuzz'
 
 
+def test_commit_offsets_sync_forwards_operation_timeout_to_run(coordinator, offsets, mocker):
+    """Facade wiring (#3121): commit_offsets_sync resolves None ->
+    default_api_timeout_ms and forwards the operation deadline to net.run()'s
+    backstop kwarg, so an explicit long timeout is not cut short by the default
+    bridge deadline."""
+    run = mocker.patch.object(coordinator._net, 'run', return_value=None)
+
+    # Explicit timeout is forwarded as run()'s backstop.
+    coordinator.commit_offsets_sync(offsets, timeout_ms=123000)
+    assert run.call_args.kwargs['timeout_ms'] == 123000
+
+    # No timeout -> default_api_timeout_ms (not request_timeout_ms).
+    run.reset_mock()
+    coordinator.commit_offsets_sync(offsets)
+    assert run.call_args.kwargs['timeout_ms'] == coordinator.config['default_api_timeout_ms']
+
+
 @pytest.mark.parametrize(
     'api_version,group_id,enable,error,has_auto_commit,commit_offsets,warn,exc', [
         ((0, 8, 0), 'foobar', True, None, False, False, True, False),
diff --git a/test/consumer/test_fetcher.py b/test/consumer/test_fetcher.py
index 03b09aaff..5ebab981c 100644
--- a/test/consumer/test_fetcher.py
+++ b/test/consumer/test_fetcher.py
@@ -642,7 +642,7 @@ def _capture_wakeup(fetcher, mocker):
     without blocking, mirroring net.run(manager.wait_for, wakeup, timeout)."""
     captured = {}
 
-    def fake_run(coro, *args):
+    def fake_run(coro, *args, timeout_ms=None):
         captured['wakeup'] = args[0]
         return None
 
@@ -674,8 +674,10 @@ def test_fetch_records_no_stall_when_response_arrives_before_wait(fetcher, topic
 
     outcome = {'stalled': None}
 
-    def realistic_run(coro, wakeup, timeout_ms):
-        # Stand-in for net.run(manager.wait_for, wakeup, timeout_ms).
+    def realistic_run(coro, wakeup, wait_timeout_ms, timeout_ms=None):
+        # Stand-in for net.run(manager.wait_for, wakeup, wait_timeout_ms,
+        # timeout_ms=...). The positional wait_for timeout and the run() backstop
+        # kwarg carry the same value.
         if wakeup.is_done:
             outcome['stalled'] = False
             # Emulate the IO thread having buffered the response that