openstack/deb-python-kafka
Code Issues Proposed changes

Moving to **kwargs for MPConsumer options

Browse Source
This commit is contained in:
Viktor Shlapakov
2015-03-24 12:49:44 +03:00
parent e43f405a0c
commit 9641e9fa29
2 changed files with 5 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
kafka/consumer/multiprocess.py
View File

@@ -109,7 +109,7 @@ class MultiProcessConsumer(Consumer):
auto_commit_every_n=AUTO_COMMIT_MSG_COUNT,
auto_commit_every_t=AUTO_COMMIT_INTERVAL,
num_procs=1, partitions_per_proc=0,
simple_consumer_options=None):
**simple_consumer_options):
# Initiate the base consumer class
super(MultiProcessConsumer, self).__init__(
@@ -150,6 +150,7 @@ class MultiProcessConsumer(Consumer):
for chunk in chunks:
options = {'partitions': list(chunk)}
if simple_consumer_options:
simple_consumer_options.pop('partitions', None)
options.update(simple_consumer_options)
args = (client.copy(), group, topic, self.queue,

8
test/test_consumer_integration.py
View File

@@ -61,10 +61,8 @@ class TestConsumerIntegration(KafkaIntegrationTestCase):
group = kwargs.pop('group', self.id().encode('utf-8'))
topic = kwargs.pop('topic', self.topic)
if consumer_class == SimpleConsumer:
if consumer_class in [SimpleConsumer, MultiProcessConsumer]:
kwargs.setdefault('iter_timeout', 0)
elif consumer_class == MultiProcessConsumer:
kwargs.setdefault('simple_consumer_options', {'iter_timeout': 0})
return consumer_class(self.client, group, topic, **kwargs)
@@ -245,8 +243,8 @@ class TestConsumerIntegration(KafkaIntegrationTestCase):
self.send_messages(0, range(0, 10))
self.send_messages(1, range(10, 20))
consumer = MultiProcessConsumer(self.client, "group1", self.topic, auto_commit=False,
simple_consumer_options={'iter_timeout': 0})
consumer = MultiProcessConsumer(self.client, "group1", self.topic,
auto_commit=False, iter_timeout=0)
self.assertEqual(consumer.pending(), 20)
self.assertEqual(consumer.pending(partitions=[0]), 10)