Long running consumer stop receiving messages and it's state is not closed

**Describe the bug**
I have a long running fastapi app which will start a consumer on app start and consume messages from kafka. The consumer works fine for maybe a week and then it suddenly stop receiving any messages while the consumer is not closed. It happened twice and I checked the log, the common error in the log before the consumer stopped consuming messages was `Failed fetch messages from 1001: [Error 7] RequestTimedOutError`. After I restarted the service, the consumer can pick up from the last stopped offset and reconsume all the missed messages.

**Expected behaviour**
A long running consumer can keep running for long time and consume messages.

**Environment (please complete the following information):**
- aiokafka version: 0.10.0 and 0.11.0
- Kafka Broker version: 3.1.0
- Other information: aiokafka 0.7.2 works fine

**Reproducible example**

I use the following scripts to reproduce the problem. So the sleep(40) will make the request timeout, conn.send() will raise Timeout exception in version 0.10.0 and 0.11.0, which make client.send() raise RequestTimedOutError. But this not happen in aiokafka 0.7.2. I've went through these issues (#983 and #802 ), and found that in conn.send, if we use asyncio.wait_for instead of utils.wait_for, then the consumer can work after timeout. My guess is these two methods have some differences which cause the problem, I tried to find the root cause but failed, so I just post my findings here.

```python
import asyncio
import logging
import time

import aiokafka

async def consume_task(consumer):
async for msg in consumer:
print("New Message Received",msg)
# await consumer.commit()

async def lag_task(producer):
while True:
print("sleeping")
await asyncio.sleep(10)
print("inducing lag")
time.sleep(40)
print("sending message")
await producer.send_and_wait(
topic='test_topic',
value=b'a message',
)

async def main():
async with (
aiokafka.AIOKafkaProducer(bootstrap_servers='kafka:29092') as producer,
aiokafka.AIOKafkaConsumer(
'test_topic',
bootstrap_servers='kafka:29092',
group_id='some_group',
enable_auto_commit=False
) as consumer,
):
consumer._fetcher._fetch_task.add_done_callback(lambda a: print("fetch task done"))
consumer._client._sync_task.add_done_callback(lambda a: print("client sync task done"))
consumer._coordinator._heartbeat_task.add_done_callback(lambda a: print("coordinator heartbeat task done"))
consumer._coordinator._commit_refresh_task.add_done_callback(lambda a: print("coordinator commit offset refresh task done"))
await consumer.seek_to_end()

task1 = asyncio.create_task(consume_task(consumer))
task2 = asyncio.create_task(lag_task(producer))

await asyncio.wait([task1, task2], return_when=asyncio.FIRST_COMPLETED)
print("something finished")

if __name__ == '__main__':
logging.basicConfig(level=logging.ERROR,format = '[%(asctime)s] p%(process)s {%(pathname)s:%(lineno)d} %(levelname)s - %(message)s' )
asyncio.run(main())
```

With the above script, I used aiokafka 0.7.2, aiokafka 0.10.0 and an updated conn.py (change line 453 from `return wait_for(fut, self._request_timeout)` to `return asyncio.wait_for(fut, self._request_timeout)`) in aiokafka 0.10.0. The output is like below

**aiokafka 0.7.2**
```sleeping
inducing lag
sending message
Heartbeat failed: local member_id was not recognized; resetting and re-joining group
Heartbeat session expired - marking coordinator dead
Marking the coordinator dead (node 1001)for group some_group.
coordinator heartbeat task done
sleeping
New Message Received ConsumerRecord(topic='test_topic', partition=0, offset=11, timestamp=1722407528852, timestamp_type=0, key=None, value=b'a message', checksum=None, serialized_key_size=-1, serialized_value_size=9, headers=())
coordinator commit offset refresh task done
inducing lag
sending message
Heartbeat failed: local member_id was not recognized; resetting and re-joining group
Heartbeat session expired - marking coordinator dead
Marking the coordinator dead (node 1001)for group some_group.
sleeping
New Message Received ConsumerRecord(topic='test_topic', partition=0, offset=12, timestamp=1722407578861, timestamp_type=0, key=None, value=b'a message', checksum=None, serialized_key_size=-1, serialized_value_size=9, headers=())
inducing lag
```

**aiokafka 0.10.0**
```sleeping
inducing lag
sending message
Failed fetch messages from 1001: [Error 7] RequestTimedOutError
Heartbeat failed: local member_id was not recognized; resetting and re-joining group
Heartbeat session expired - marking coordinator dead
Marking the coordinator dead (node 1001)for group some_group.
coordinator heartbeat task done
sleeping
coordinator commit offset refresh task done
fetch task done
inducing lag
sending message
Heartbeat failed: local member_id was not recognized; resetting and re-joining group
Heartbeat session expired - marking coordinator dead
Marking the coordinator dead (node 1001)for group some_group.
sleeping
inducing lag
sending message
Heartbeat failed: local member_id was not recognized; resetting and re-joining group
Heartbeat session expired - marking coordinator dead
Marking the coordinator dead (node 1001)for group some_group.
sleeping
inducing lag
```

**aiokafka 0.11.0**
```
sleeping
inducing lag
sending message
[2024-08-05 07:41:49,015] p2865 {/usr/local/lib/python3.11/site-packages/aiokafka/consumer/fetcher.py:693} ERROR - Failed fetch messages from 1001: [Error 7] RequestTimedOutError
[2024-08-05 07:41:49,018] p2865 {/usr/local/lib/python3.11/site-packages/aiokafka/consumer/group_coordinator.py:798} ERROR - Heartbeat session expired - marking coordinator dead
coordinator heartbeat task done
sleeping
coordinator commit offset refresh task done
inducing lag
sending message
[2024-08-05 07:42:39,027] p2865 {/usr/local/lib/python3.11/site-packages/aiokafka/consumer/fetcher.py:693} ERROR - Failed fetch messages from 1001: [Error 7] RequestTimedOutError
[2024-08-05 07:42:39,028] p2865 {/usr/local/lib/python3.11/site-packages/aiokafka/consumer/group_coordinator.py:798} ERROR - Heartbeat session expired - marking coordinator dead
sleeping
inducing lag
sending message
[2024-08-05 07:43:29,031] p2865 {/usr/local/lib/python3.11/site-packages/aiokafka/consumer/fetcher.py:693} ERROR - Failed fetch messages from 1001: [Error 7] RequestTimedOutError
[2024-08-05 07:43:29,032] p2865 {/usr/local/lib/python3.11/site-packages/aiokafka/consumer/group_coordinator.py:798} ERROR - Heartbeat session expired - marking coordinator dead
sleeping
inducing lag
```

**aiokafka 0.10.0 after change wait_for to asyncio.wait_for() in conn.py::send()**
```sleeping
inducing lag
sending message
Heartbeat failed: local member_id was not recognized; resetting and re-joining group
Heartbeat session expired - marking coordinator dead
Marking the coordinator dead (node 1001)for group some_group.
coordinator heartbeat task done
New Message Received ConsumerRecord(topic='test_topic', partition=0, offset=20, timestamp=1722408250464, timestamp_type=0, key=None, value=b'a message', checksum=None, serialized_key_size=-1, serialized_value_size=9, headers=())
sleeping
coordinator commit offset refresh task done
inducing lag
sending message
Heartbeat failed: local member_id was not recognized; resetting and re-joining group
Heartbeat session expired - marking coordinator dead
Marking the coordinator dead (node 1001)for group some_group.
sleeping
New Message Received ConsumerRecord(topic='test_topic', partition=0, offset=21, timestamp=1722408300480, timestamp_type=0, key=None, value=b'a message', checksum=None, serialized_key_size=-1, serialized_value_size=9, headers=())
inducing lag
sending message
Heartbeat failed: local member_id was not recognized; resetting and re-joining group
Heartbeat session expired - marking coordinator dead
Marking the coordinator dead (node 1001)for group some_group.
sleeping
New Message Received ConsumerRecord(topic='test_topic', partition=0, offset=22, timestamp=1722408350485, timestamp_type=0, key=None, value=b'a message', checksum=None, serialized_key_size=-1, serialized_value_size=9, headers=())
```

**aiokafka 0.11.0 after change wait_for to asyncio.wait_for() in conn.py::send()**
```
sleeping
inducing lag
sending message
[2024-08-05 07:48:29,823] p4507 {/usr/local/lib/python3.11/site-packages/aiokafka/consumer/group_coordinator.py:798} ERROR - Heartbeat session expired - marking coordinator dead
coordinator heartbeat task done
sleeping
New Message Received ConsumerRecord(topic='test_topic', partition=0, offset=522, timestamp=1722844109821, timestamp_type=0, key=None, value=b'a message', checksum=None, serialized_key_size=-1, serialized_value_size=9, headers=())
coordinator commit offset refresh task done
inducing lag
sending message
[2024-08-05 07:49:19,830] p4507 {/usr/local/lib/python3.11/site-packages/aiokafka/consumer/group_coordinator.py:798} ERROR - Heartbeat session expired - marking coordinator dead
sleeping
inducing lag
sending message
[2024-08-05 07:50:09,835] p4507 {/usr/local/lib/python3.11/site-packages/aiokafka/consumer/group_coordinator.py:798} ERROR - Heartbeat session expired - marking coordinator dead
sleeping
New Message Received ConsumerRecord(topic='test_topic', partition=0, offset=524, timestamp=1722844209833, timestamp_type=0, key=None, value=b'a message', checksum=None, serialized_key_size=-1, serialized_value_size=9, headers=())
inducing lag
```