julien-duponchelle / julien-duponchelle/python-mysql-replication

New Feature: asyncio support for BinLogStreamReader

Open
#593 13 comments 7 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
2.4k
Forks
690
PR merge metrics
No merged PRs in 30d

Description

Hi python-mysql-replication community, I want to request asyncio support for BinLogStreamReader.

My use case here is to use BinLogStreamReader to receive binlog message and forward it to a message queue and a consumer will process it. Here's an example of my code:

```
def binlog_subscribe():
stream = BinLogStreamReader(
connection_settings=MYSQL_SETTINGS,
server_id=2,
only_events=[DeleteRowsEvent, UpdateRowsEvent, WriteRowsEvent],
blocking=True,
enable_logging=False,
only_schemas=set(["test"]),
only_tables=set(["binlog_test"]))

for binlogevent in stream:
for row in binlogevent.rows:
yield row

async def forward_message():
while True:
try:
message = await asyncio.wait_for(memory_queue.get(), timeout=XXX)
await message_queue.push(message)
except asyncio.TimeoutError:
logging.debug("Memory queue is empty")

async def main():
forward_task = asyncio.create_task(forward_message())
for message in binlog_subscribe():
await memory_queue.push(row)

if __name__ == "__main__":
asyncio.run()
```

Due to the fact that BinLogStreamReader `__iter__` method is blocking, it blocks the asyncio event loop and causes some side effects, eg.
1. blocking background health check with message queue server under low qps
2. limit cpu usage under high qps.
3. some of the message in the memory queue is not sent out in time

And I see the examples like https://github.com/julien-duponchelle/python-mysql-replication/blob/main/examples/mysql_to_kafka.py, it's a common use case to have a forward service to listen binlog stream and forward it to somewhere else.

Currently the package does not support asyncio. I think asyncio is a fit for this package because reading binlog is I/O process and asyncio allows parallel processing for multiple I/O process. There's also open source 3p lib that supports asyncio like [aiomysql](https://github.com/aio-libs/aiomysql), [aiokafka](https://github.com/aio-libs/aiokafka) so I think there're use cases to have asyncio support. For order guarantee, as long as we continue to use iterator interface it will not cause message disorder issues.

Let me know your idea!

Contributor guide

Open the contributing guide

Research direction

Start by reading BinLogStreamReader.__iter__ and the forwarding pattern in examples/mysql_to_kafka.py. Define the intended asyncio-facing interface and verify that binlog consumption does not block the event loop, preserves message order, and supports the forwarding use case described in the issue.

Written by the indexing model from the issue text.

Assessment

Tech stack
mysql, python
Domain
backend, databases
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.