[utku0dogan]

Fixes #3183

When a broker stops answering while keeping TCP connections open (e.g. a paused container or silent network blackhole), KafkaConsumer.poll() previously entered a tight CPU-bound loop and leaked memory until request_timeout_ms expired:

1. Busy-loop (poll_timeout_ms == 0): Once heartbeat_interval_ms elapsed while an in-flight HeartbeatRequest (or async auto-commit) was still waiting on the unresponsive broker, coordinator.time_to_next_poll() continuously returned 0, clamping poll_timeout_ms = 0 in KafkaConsumer._poll_once and calling Fetcher.fetch_records(timeout_ms=0) thousands of times per second.
2. Callback / traceback retention on pending futures: Neither Fetcher.fetch_records nor NetBackend.await_for detached its temporary wakeup callbacks from still-pending futures upon timeout, retaining _wake -> wakeup -> wrapper -> KafkaTimeoutError.__traceback__ closures on long-lived in-flight Future instances across every timed-out wait.

Changes

• Add Future._remove_cb_eb(cb, eb) to detach (callback, errback) pairs under a single lock acquire if the future is still pending.
• Detach temporary callbacks in finally blocks in Fetcher.fetch_records and NetBackend.await_for when waited futures have not yet resolved.
• Track Heartbeat.heartbeat_in_flight so BaseCoordinator.time_to_next_heartbeat() returns heartbeat_interval_ms / 1000 instead of 0 while a heartbeat request is already in flight.
• Floor non-positive coordinator.time_to_next_poll() deadlines to retry_backoff_ms in KafkaConsumer._poll_once (after coordinator.poll() has already run), while preserving non-blocking poll(timeout_ms=0) semantics.