Fix thread not waking up when there is still data to be sent #34
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
gqmelo
force-pushed
the
gqmelo/fix-sending-thread-wakeup
branch
3 times, most recently
from
880fe0e to
42c6f8e
Compare
gqmelo
commented
Jul 8, 2025
| conn.send(req2, blocking=False) | ||
| # This will send the remaining bytes in the buffer from the first request, but should notice | ||
| # that the second request was queued, therefore it should return False. | ||
| assert conn.send_pending_requests_v2() is False |
Author
There was a problem hiding this comment.
This is where the test would fail before the fix.
juha-aiven
reviewed
Jul 9, 2025
|
I think the PR is fine. But I'd like to somebody like e.g. @aiven-anton to have a look at it too. The PR touches Thus, not approving/merging. |
aiven-anton
reviewed
Aug 12, 2025
aiven-anton
left a comment
aiven-anton
left a comment
There was a problem hiding this comment.
I am not well versed in this code-base, but this seems like a well thought-through fix and I cannot spot any real issues with it. It seems sound to move ahead with this.
We should also notify the upstream(s) of this issue and fix.
I left two minor comments, let me know if you want to address those.
When producing messages quickly without waiting for the future of previous requests, there could be some situations when the last batch was not sent. That seemed to be more frequent with larger messages (~100KiB), but apparently it could happen to any message when `linger_ms` is 0. Not sure if it could happen when it is non-zero though. The reason is that `BrokerConnection.send_pending_requests_v2` would fill the internal buffer with the bytes from a request and try to send it. https://github.com/aiven/kafka-python/blob/e0ab864f7aca3961e729cf03d1caa3899fbee617/kafka/conn.py#L1036 If it couldn't send it completely for some reason, it would try to send again in the next call to `send_pending_requests_v2`. But if between those 2 calls, `BrokerConnection.send` was called, new data would be appended to self._protocol: KafkaProtocol: https://github.com/aiven/kafka-python/blob/b01ffb6a004480635751e325db2ded20bcdc0d2f/kafka/conn.py#L981 but the second call to `send_pending_requests_v2` wouldn't check if any new data was available and would return False: https://github.com/aiven/kafka-python/blob/e0ab864f7aca3961e729cf03d1caa3899fbee617/kafka/conn.py#L1035 This would tell `KafkaClient._poll` that all pending data was sent, which would make the client not listen to socked write readiness anymore: https://github.com/aiven/kafka-python/blob/b01ffb6a004480635751e325db2ded20bcdc0d2f/kafka/client_async.py#L663-L667
gqmelo
force-pushed
the
gqmelo/fix-sending-thread-wakeup
branch
from
42c6f8e to
719cf45
Compare
Author
|
After this is merged, I'll open a PR in the upstream repo |
aiven-anton
approved these changes
Aug 13, 2025
Author
Upstream PR: dpkp#2670 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
When producing messages quickly without waiting for the future of previous requests, there could be some situations when the last batch was not sent.
That seemed to be more frequent with larger messages (~100KiB), but apparently it could happen to any message when
linger_msis 0. Not sure if it could happen when it is non-zero though.The reason is that
BrokerConnection.send_pending_requests_v2would fill the internal buffer with the bytes from a request and try to send it.kafka-python/kafka/conn.py
Line 1036 in e0ab864
If it couldn't send it completely for some reason, it would try to send again in the next call to
send_pending_requests_v2.But if between those 2 calls,
BrokerConnection.sendwas called, new data would be appended to self._protocol: KafkaProtocol:kafka-python/kafka/conn.py
Line 981 in b01ffb6
but the second call to
send_pending_requests_v2wouldn't check if any new data was available and would return False:kafka-python/kafka/conn.py
Line 1035 in e0ab864
This would tell
KafkaClient._pollthat all pending data was sent, which would make the client not listen to socked write readiness anymore:kafka-python/kafka/client_async.py
Lines 663 to 667 in b01ffb6