openstack/oslo.messaging
Code Issues Proposed changes

Merge "Get kafka notifications to work with kafka-python 0.9.5"

Browse Source
This commit is contained in:
Jenkins 2016-02-23 21:40:27 +00:00 committed by Gerrit Code Review
commit 629632bfff
2 changed files with 3 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
oslo_messaging/_drivers/impl_kafka.py
View File

@ -228,8 +228,7 @@ class Connection(object):
def declare_topic_consumer(self, topics, group=None):
self.consumer = kafka.KafkaConsumer(
*topics, group_id=group,
metadata_broker_list=["%s:%s" % (self.host, str(self.port))],
# auto_commit_enable=self.auto_commit,
bootstrap_servers=["%s:%s" % (self.host, str(self.port))],
fetch_message_max_bytes=self.fetch_messages_max_bytes)
@ -262,6 +261,7 @@ class KafkaListener(base.Listener):
messages = self.conn.consume(timeout=timeout)
for msg in messages:
message = msg.value
LOG.debug('poll got message : %s', message)
message = jsonutils.loads(message)
self.incoming_queue.append(OsloKafkaMessage(
ctxt=message['context'], message=message['message']))

2
oslo_messaging/tests/drivers/test_impl_kafka.py
View File

@ -284,4 +284,4 @@ class TestWithRealKafkaBroker(test_utils.BaseTestCase):
deadline = time.time() + 3
received_message = listener.poll(timeout=3)
self.assertEqual(0, int(deadline - time.time()))
self.assertIsNone(received_message)
self.assertEqual([], received_message)