openstack/monasca-common
Code Issues Proposed changes

Confluent Kafka driver broker option bug

Browse Source 
In confluent Kafka driver, brokers option is string type not list type.
Comma separated list of host/port pairs. See
https://github.com/confluentinc/confluent-kafka-python/blob/master/README.md

Story: 2003705
Task: 36199

Change-Id: I5961ee32c91a9a946831a36dc88413e56b251e1e
This commit is contained in:
zhangjianweibj 2019-08-07 16:17:38 +08:00 committed by Witek Bedyk
parent 2a6cc6cc0a
commit 2247194f08
2 changed files with 4 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
monasca_common/confluent_kafka/consumer.py
View File

@ -29,8 +29,8 @@ class KafkaConsumer(object):
""" """
Create new high-level Consumer instance. Create new high-level Consumer instance.
:param list(str) bootstrap_servers: A list of host/port pairs to use :param str bootstrap_servers: Comma separated list of host/port pairs to
for establishing the initial connection to the Kafka cluster. use for establishing the initial connection to the Kafka cluster.
:param str group_id: A unique string that identifies the consumer group :param str group_id: A unique string that identifies the consumer group
this consumer belongs to. this consumer belongs to.
:param str topic: Topic to subscribe to. :param str topic: Topic to subscribe to.

6
monasca_common/tests/test_confluent_kafka.py
View File

@ -101,8 +101,7 @@ class TestConfluentKafkaConsumer(base.BaseTestCase):
def setUp(self, mock_confluent_consumer): def setUp(self, mock_confluent_consumer):
super(TestConfluentKafkaConsumer, self).setUp() super(TestConfluentKafkaConsumer, self).setUp()
self.mock_confluent_consumer = mock_confluent_consumer self.mock_confluent_consumer = mock_confluent_consumer
self.consumer = consumer.KafkaConsumer(['fake_server1', self.consumer = consumer.KafkaConsumer('fake_server1,fake_server2',
'fake_server2'],
'fake_group', 'fake_group',
FAKE_KAFKA_TOPIC, 128, FAKE_KAFKA_TOPIC, 128,
'test_client', 'test_client',
@ -124,8 +123,7 @@ class TestConfluentKafkaConsumer(base.BaseTestCase):
def test_kafka_consumer_init(self): def test_kafka_consumer_init(self):
expected_config = {'group.id': 'fake_group', expected_config = {'group.id': 'fake_group',
'session.timeout.ms': 10000, 'session.timeout.ms': 10000,
'bootstrap.servers': ['fake_server1', 'bootstrap.servers': 'fake_server1,fake_server2',
'fake_server2'],
'fetch.min.bytes': 128, 'fetch.min.bytes': 128,
'client.id': 'test_client', 'client.id': 'test_client',
'enable.auto.commit': False, 'enable.auto.commit': False,