openstack/deb-python-kafka
Code Issues Proposed changes

PR 331 fixup: do not attempt to get new messages if there are pending retries

Browse Source
This commit is contained in:
Dana Powers
2015-06-06 16:09:30 -07:00
parent 794ab5bba4
commit 9712f613c9
1 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
kafka/producer/base.py
View File

@@ -52,13 +52,18 @@ def _send_upstream(queue, client, codec, batch_time, batch_size,
while not stop_event.is_set(): while not stop_event.is_set():
timeout = batch_time timeout = batch_time
count = batch_size
# it's a simplification: we're comparing message sets and
# messages: each set can contain [1..batch_size] messages
count = batch_size - len(request_tries)
send_at = time.time() + timeout send_at = time.time() + timeout
msgset = defaultdict(list) msgset = defaultdict(list)
# Merging messages will require a bit more work to manage correctly
# for now, dont look for new batches if we have old ones to retry
if request_tries:
count = 0
log.debug('Skipping new batch collection to handle retries')
else:
log.debug('Batching size: {0}, timeout: {1}'.format(count, timeout))
# Keep fetching till we gather enough messages or a # Keep fetching till we gather enough messages or a
# timeout is reached # timeout is reached
while count > 0 and timeout >= 0: while count > 0 and timeout >= 0: