Comments (8)
Maybe it is there is weird, on scarce traffic, a batch should still be send at some point (like the next time the sender coroutine execute, that should follow up as there is trigger when data is ready to be drained). Could you provide a more complete example ? (where you have your "stop()" call for example)
What is a bit tricky with send
is that the coroutine itself is just about appending the message to the batch, so it should be quite fast/immediate, i.e await send(...)
go next before anything will be send on the network. Then this coroutine return a future that will be completed when the message is actually sent, so somehow the code of "send_and_await" is a bit like
fut = await send(...)
await fut
from aiokafka.
Kafka is optimized for throughput and the main idea to achieve that with fewer brokers is to delegate a lot of work to the clients (producer and consumer). More or less, you will have the following flow :
- the producer partitions messages, group them in a chunk and compress them
- the producer send the chunk of messages, compressed to the broker with a TCP message
- the broker gets the chunk, and append it to a file "as is" (ideally the broker doesn't run anykind of processing on the chunk, just a simple copy from the socket to the file)
- the consumer polls the broker, sees there is a new chunk, fetches it (usually from broker cache memory as it is a file just written)
- the consumer decompresses the chunk and retrieve individual messages
The behavior you are observing then could be consider normal, as the producer tries to batch as many message as possible (there are some producer options to control that). There is also a "send_and_wait" method for the app to wait for a batch to be send, but then it will tend to have one message per chunk, it is not very efficient, see https://aiokafka.readthedocs.io/en/stable/producer.html#message-buffering
from aiokafka.
@vmaurin The link says "By default, a new batch is sent immediately after the previous one (even if itβs not full)". As linger_ms
is set to 0 by default I expected that messages should be sent immediately. Now I see that send()
method can just wait forever if messages are scarce.
from aiokafka.
@vmaurin example with stop()
:
import asyncio
from time import sleep
from aiokafka import AIOKafkaProducer
async def main():
producer = AIOKafkaProducer(bootstrap_servers='localhost:9092')
await producer.start()
await producer.send('topic', b'test')
sleep(30)
await producer.stop()
asyncio.run(main())
Message gets to Kafka only after stop()
is called. And the example in initial post does not send anything at all.
from aiokafka.
Thank you for the snippet @ant0nk
With a time.sleep, you are blocking the process running the asyncio event loop, so the sender co routine won't be able to run. Could you try the same sample with an await asyncio.sleep(30)
instead ?
from aiokafka.
@vmaurin So when program have infinite cycle it must have asyncio steps to send()
to work.
For example if I have program like this:
import asyncio
import os
from aiokafka import AIOKafkaProducer
async def main():
producer = AIOKafkaProducer(bootstrap_servers='localhost:9092')
await producer.start()
while True:
if os.path.exists('/path/to/trigger.txt'):
await producer.send('topic', b'I saw trigger.txt')
os.remove('/path/to/trigger.txt')
asyncio.run(main())
it will send first message only when file /path/to/trigger.txt
will be created for second time.
from aiokafka.
@ant0nk Yes, this one is a common advice when using asyncio/coroutine in python (but also in similar thing like javascript) "Don't block the event loop". So if you need to lookup for your file to exist, you can either try to find a lib to do file IO in a coroutine way, or use a thread pool to run you IO blocking operation + sync primitive to combine both
https://docs.python.org/3/library/asyncio-task.html#running-in-threads
from aiokafka.
Thanks, closing issue.
from aiokafka.
Related Issues (20)
- UnsupportedCodecError for snappy when producing events without snappy compression HOT 4
- [QUESTION] Perfomance producer with large messages HOT 1
- PY_SSIZE_T_CLEAN macro must be defined for '#' formats HOT 1
- Python 3.12.0 compatibility issue HOT 16
- [QUESTION] How to deal with the background rebalance process
- Snappy package doesn't support Python 3.11
- [QUESTION] I am using AIOkafka, but i have some question. HOT 1
- Inconsistent producer start/stop
- AIOKafkaAdminClient.create_topics fails randomly because it selects a node at random HOT 1
- performance degradation of producer when having many topics HOT 1
- Can I use Azure Event Hubs with `aiokafka`? HOT 1
- 0.9.0 zstd codec depends on cramjam but missing in documentation HOT 1
- Let's put `aiokafka` under the `aio-libs` org on PyPI HOT 2
- Add create_acls function for kafka admin client
- AIOKafkaProducer failed to produce message with headers HOT 3
- I keep getting MessageSizeTooLargeError, error message gives size much bigger than actual message were given to producer. HOT 9
- Add delete_records to the admin client HOT 5
- Can't connect to kafka docker HOT 1
- asyncio.exceptions.CancelledError
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
π Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. πππ
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google β€οΈ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from aiokafka.