Repository navigation
Track unacked Kafka messages per QoS instance - #2680
Merged
auvipy merged 3 commits intoSep 30, 2026
Merged
Conversation
QoS._not_yet_acked was a class attribute, so every channel on every connection shared one dict. can_consume() compared that global count against each channel's own prefetch_count, which let one channel's unacked messages stop an unrelated channel from receiving anything. Initialise the dict in QoS.__init__ so each channel counts only its own deliveries.
auvipy
self-requested a review
September 27, 2026 10:52
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #2680 +/- ##
==========================================
+ Coverage 84.35% 85.02% +0.66%
==========================================
Files 80 80
Lines 11315 11317 +2
Branches 1354 1354
==========================================
+ Hits 9545 9622 +77
+ Misses 1552 1475 -77
- Partials 218 220 +2 ☔ View full report in Codecov by Harness. |
auvipy
approved these changes
Sep 30, 2026
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
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.
QoS._not_yet_ackedin the confluentkafka transport is a class attribute, so every channel on every connection shares one dict.can_consume()andcan_consume_max_estimate()compare that shared count against the channel's ownprefetch_count, which means unacked messages on one channel can stop an unrelated channel from receiving anything.I reproduced this against a local Kafka broker (apache/kafka 3.8) with two connections, each consuming its own topic with
prefetch_count=1. Once connection A held one unacked message, connection B never got its message delivered. With this change both are delivered.The fix moves the dict into
QoS.__init__, the same per-instance scoping used for the SQS and Azure Storage Queues caches in #2676 and #2679. There were no unit tests for this transport yet, so I addedt/unit/transport/test_confluentkafka.pywith a test that fails on main and one covering ack. The unit suite andt/integration/test_kafka.py(36 passed, 1 skipped) pass locally.