openstack/oslo.messaging
Code Issues Proposed changes

Merge "[kafka] Do not remove kafka_client during reset"

Browse Source
This commit is contained in:
Jenkins 2016-04-10 23:48:36 +00:00 committed by Gerrit Code Review
commit c81aa02832
1 changed files with 3 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
oslo_messaging/_drivers/impl_kafka.py
View File

@ -187,12 +187,8 @@ class Connection(object):
def reset(self): def reset(self):
"""Reset a connection so it can be used again.""" """Reset a connection so it can be used again."""
if self.kafka_client: if self.consumer:
self.kafka_client.close() self.consumer.close()
self.kafka_client = None
if self.producer:
self.producer.stop()
self.producer = None
self.consumer = None self.consumer = None
def close(self): def close(self):
@ -233,6 +229,7 @@ class Connection(object):
*topics, group_id=group, *topics, group_id=group,
bootstrap_servers=["%s:%s" % (self.host, str(self.port))], bootstrap_servers=["%s:%s" % (self.host, str(self.port))],
fetch_message_max_bytes=self.fetch_messages_max_bytes) fetch_message_max_bytes=self.fetch_messages_max_bytes)
self._consume_loop_stopped = False
class OsloKafkaMessage(base.RpcIncomingMessage): class OsloKafkaMessage(base.RpcIncomingMessage):