[BUG] Heartbeat background thread in "Transaction Consume-Process-Produce" paradigm
- 主要语言
- Python
- 星标
- 1.4k
- 派生
- 269
- 平均合并
- 1 天 1 小时
- 30 天内合并 PR
- 6
描述
**Describe the bug**
I'm noticing that heartbeats are not being sent from a background thread when using the "Transaction Consume-Process-Produce" paradigm described [here](https://aiokafka.readthedocs.io/en/stable/examples/transaction_example.html).
**Expected behaviour**
I expect the logger to print continuous heartbeats as consumer messages are processing but I notice that the heartbeats freeze when a worker is processing a job and continue after the processing is done.
**Environment (please complete the following information):**
- aiokafka version (`python -c "import aiokafka; print(aiokafka.__version__)"`): 0.7.0
- kafka-python version (`python -c "import kafka; print(kafka.__version__)"`): 2.0.2
- Kafka Broker version (`kafka-topics.sh --version`): 2.6.0 (Commit:62abe01bee039651)
- Other information (Confluent Cloud version, etc.):
**Reproducible example**
```python
import asyncio
import logging
from aiokafka import \
AIOKafkaConsumer, \
AIOKafkaProducer
import time
logging.basicConfig(level=logging.DEBUG)
def build_offset(msg):
tp = TopicPartition(msg.topic, msg.partition)
ret = {}
ret[tp] = msg.offset + 1
return ret
async def func():
loop = asyncio.get_event_loop()
consumer = AIOKafkaConsumer(
auto_offset_reset='earliest',
group_id="group",
loop=loop,
max_poll_records=1,
enable_auto_commit=False,
isolation_level="read_committed",
heartbeat_interval_ms=1000)
consumer.subscribe(["test_event"])
await consumer.start()
producer = AIOKafkaProducer(
loop=loop,
transactional_id="some-id"
)
await producer.start()
try:
while True:
msg = await consumer.getone()
print(msg)
async with producer.transaction():
commit_offsets = build_offset(msg);
time.sleep(5)
await producer.send("test_event", b"some val")
await producer.send_offsets_to_transaction(commit_offsets, "group")
finally:
await consumer.stop()
await producer.stop()
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(func())
```
**Additional Info**
The above example is a pared down version of the example provided in the docs (linked above). I'm simply grabbing an event off of the dummy `test_event` queue and sleeping for 5 seconds (to simulate an intensive processor load) and committing back to the same topic. When I run this code, I notice that the heartbeats occur once a second as expected before I fire off the first event to `test_event`. Then, something strange occurs: the heartbeats stop while the code is executing the `time.sleep(5)` line, and continue after that finishes. I don't think I'm missing print statements, because the logger prints out the heartbeat number, and they seem to be printing sequentially (despite the gap in heartbeats during the 5 second sleep). Here's a snippet of what I see when I run the above script with logging set to debug (filtering for just heartbeat events):
```
ConsumerRecord(topic='test_event', partition=0, offset=34, timestamp=1603958802338, timestamp_type=0, key=None, value=b'some val', checksum=None, serialized_key_size=-1, serialized_value_size=8, headers=())
DEBUG:aiokafka.consumer.group_coordinator:Heartbeat: group[17] aiokafka-0.7.0-e919e4d3-b8f2-402e-87f5-f857120d6a28
DEBUG:aiokafka.conn: Request 5: HeartbeatRequest_v1(group='group', generation_id=17, member_id='aiokafka-0.7.0-e919e4d3-b8f2-402e-87f5-f857120d6a28')
DEBUG:aiokafka.conn: Response 5: HeartbeatResponse_v1(throttle_time_ms=0, error_code=0)
ConsumerRecord(topic='test_event', partition=0, offset=36, timestamp=1603958807363, timestamp_type=0, key=None, value=b'some val', checksum=None, serialized_key_size=-1, serialized_value_size=8, headers=())
DEBUG:aiokafka.consumer.group_coordinator:Heartbeat: group[17] aiokafka-0.7.0-e919e4d3-b8f2-402e-87f5-f857120d6a28
DEBUG:aiokafka.conn: Request 6: HeartbeatRequest_v1(group='group', generation_id=17, member_id='aiokafka-0.7.0-e919e4d3-b8f2-402e-87f5-f857120d6a28')
DEBUG:aiokafka.conn: Response 6: HeartbeatResponse_v1(throttle_time_ms=0, error_code=0)
ConsumerRecord(topic='test_event', partition=0, offset=38, timestamp=1603959012488, timestamp_type=0, key=None, value=b'some val', checksum=None, serialized_key_size=-1, serialized_value_size=8, headers=())
DEBUG:aiokafka.consumer.group_coordinator:Heartbeat: group[17] aiokafka-0.7.0-e919e4d3-b8f2-402e-87f5-f857120d6a28
DEBUG:aiokafka.conn: Request 7: HeartbeatRequest_v1(group='group', generation_id=17, member_id='aiokafka-0.7.0-e919e4d3-b8f2-402e-87f5-f857120d6a28')
DEBUG:aiokafka.conn: Response 7: HeartbeatResponse_v1(throttle_time_ms=0, error_code=0)
ConsumerRecord(topic='test_event', partition=0, offset=40, timestamp=1603959017511, timestamp_type=0, key=None, value=b'some val', checksum=None, serialized_key_size=-1, serialized_value_size=8, headers=())
```
Clearly, even though the heartbeat interval was set to 1 second, only three heartbeats were printed out (there should have been 15 since the time interval between each message is roughly 5 seconds). My understanding was that aiokafka will send heartbeats via a separate background thread, as per [KIP 62](https://cwiki.apache.org/confluence/display/KAFKA/KIP-62%3A+Allow+consumer+to+send+heartbeats+from+a+background+thread), but it seems that this is not happening somehow. Any guidance here is appreciated!
贡献指南
评估
这个 Issue 还没有评估数据。