Skip to content

Add producer batch send queue size limit #304

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 3 commits into from
Closed

Add producer batch send queue size limit #304

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d9464d9
add producer batch send queue size limit
Jan 23, 2015
b0f5ef7
add producer send batch queue overfilled test
Jan 23, 2015
b660057
async queue: refactored code; add one more test
ediskandarov Jan 23, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions kafka/common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,10 @@ class KafkaConfigurationError(KafkaError):
pass


class BatchQueueOverfilledError(KafkaError):
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about AsyncProducerQueueFull

pass


def _iter_broker_errors():
for name, obj in inspect.getmembers(sys.modules[__name__]):
if inspect.isclass(obj) and issubclass(obj, BrokerResponseError) and obj != BrokerResponseError:
Expand Down
22 changes: 17 additions & 5 deletions kafka/producer/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,15 +4,16 @@
import time

try:
from queue import Empty
from queue import Empty, Full
except ImportError:
from Queue import Empty
from Queue import Empty, Full
from collections import defaultdict
from multiprocessing import Queue, Process

import six

from kafka.common import (
BatchQueueOverfilledError,
ProduceRequest, TopicAndPartition, UnsupportedCodecError
)
from kafka.protocol import CODEC_NONE, ALL_CODECS, create_message_set
Expand All @@ -21,6 +22,8 @@

BATCH_SEND_DEFAULT_INTERVAL = 20
BATCH_SEND_MSG_COUNT = 20
# unlimited
ASYNC_QUEUE_MAXSIZE = 0

STOP_ASYNC_PRODUCER = -1

Expand Down Expand Up @@ -113,12 +116,14 @@ def __init__(self, client, async=False,
codec=None,
batch_send=False,
batch_send_every_n=BATCH_SEND_MSG_COUNT,
batch_send_every_t=BATCH_SEND_DEFAULT_INTERVAL):
batch_send_every_t=BATCH_SEND_DEFAULT_INTERVAL,
async_queue_maxsize=ASYNC_QUEUE_MAXSIZE):

if batch_send:
async = True
assert batch_send_every_n > 0
assert batch_send_every_t > 0
assert async_queue_maxsize >= 0
else:
batch_send_every_n = 1
batch_send_every_t = 3600
Expand All @@ -139,7 +144,8 @@ def __init__(self, client, async=False,
log.warning("async producer does not guarantee message delivery!")
log.warning("Current implementation does not retry Failed messages")
log.warning("Use at your own risk! (or help improve with a PR!)")
self.queue = Queue() # Messages are sent through this queue
# Messages are sent through this queue
self.queue = Queue(maxsize=async_queue_maxsize)
self.proc = Process(target=_send_upstream,
args=(self.queue,
self.client.copy(),
Expand Down Expand Up @@ -188,7 +194,13 @@ def _send_messages(self, topic, partition, *msg, **kwargs):

if self.async:
for m in msg:
self.queue.put((TopicAndPartition(topic, partition), m, key))
try:
item = (TopicAndPartition(topic, partition), m, key)
self.queue.put_nowait(item)
except Full:
raise BatchQueueOverfilledError(
'Producer batch send queue overfilled. '
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

s/batch/async/

'Current queue size %d.' % self.queue.qsize())
resp = []
else:
messages = create_message_set(msg, self.codec, key)
Expand Down
30 changes: 29 additions & 1 deletion test/test_producer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,10 @@

import logging

from mock import MagicMock
from mock import MagicMock, patch
from . import unittest

from kafka.common import BatchQueueOverfilledError
from kafka.producer.base import Producer

class TestKafkaProducer(unittest.TestCase):
Expand All @@ -25,3 +26,30 @@ def test_producer_message_types(self):
# This should not raise an exception
producer.send_messages(topic, partition, m)

@patch('kafka.producer.base.Process')
def test_producer_async_queue_overfilled_batch_send(self, process_mock):
queue_size = 2
producer = Producer(MagicMock(), batch_send=True,
async_queue_maxsize=queue_size)

topic = b'test-topic'
partition = 0
message = b'test-message'

with self.assertRaises(BatchQueueOverfilledError):
message_list = [message] * (queue_size + 1)
producer.send_messages(topic, partition, *message_list)

@patch('kafka.producer.base.Process')
def test_producer_async_queue_overfilled(self, process_mock):
queue_size = 2
producer = Producer(MagicMock(), async=True,
async_queue_maxsize=queue_size)

topic = b'test-topic'
partition = 0
message = b'test-message'

with self.assertRaises(BatchQueueOverfilledError):
message_list = [message] * (queue_size + 1)
producer.send_messages(topic, partition, *message_list)