From 799cd6fa8f5435e3b4473e4e224df099380919c1 Mon Sep 17 00:00:00 2001 From: Mehdi Abaakouk Date: Wed, 7 Dec 2016 19:30:40 +0100 Subject: kafka: disable batch for functional tests Change-Id: I09a3049ca5f4647d0f6b002b3732a4c0edd43986 --- oslo_messaging/_drivers/impl_kafka.py | 8 +------- oslo_messaging/_drivers/kafka_options.py | 7 +++++++ oslo_messaging/tests/functional/utils.py | 21 +++++++++++++-------- 3 files changed, 21 insertions(+), 15 deletions(-) diff --git a/oslo_messaging/_drivers/impl_kafka.py b/oslo_messaging/_drivers/impl_kafka.py index 6c90d30..4c8dc29 100644 --- a/oslo_messaging/_drivers/impl_kafka.py +++ b/oslo_messaging/_drivers/impl_kafka.py @@ -26,7 +26,6 @@ import threading import kafka from kafka.client_async import selectors import kafka.errors -from oslo_config import cfg from oslo_log import log as logging from oslo_utils import eventletutils import tenacity @@ -319,15 +318,10 @@ class KafkaDriver(base.BaseDriver): def __init__(self, conf, url, default_exchange=None, allowed_remote_exmods=None): - - opt_group = cfg.OptGroup(name='oslo_messaging_kafka', - title='Kafka driver options') - conf.register_group(opt_group) - conf.register_opts(kafka_options.KAFKA_OPTS, group=opt_group) - super(KafkaDriver, self).__init__( conf, url, default_exchange, allowed_remote_exmods) + kafka_options.register_opts(conf) # the pool configuration properties max_size = self.conf.oslo_messaging_kafka.pool_size min_size = self.conf.oslo_messaging_kafka.conn_pool_min_size diff --git a/oslo_messaging/_drivers/kafka_options.py b/oslo_messaging/_drivers/kafka_options.py index c733a0a..7989288 100644 --- a/oslo_messaging/_drivers/kafka_options.py +++ b/oslo_messaging/_drivers/kafka_options.py @@ -50,3 +50,10 @@ KAFKA_OPTS = [ cfg.IntOpt('producer_batch_size', default=16384, help='Size of batch for the producer async send') ] + + +def register_opts(conf): + opt_group = cfg.OptGroup(name='oslo_messaging_kafka', + title='Kafka driver options') + conf.register_group(opt_group) + conf.register_opts(KAFKA_OPTS, group=opt_group) diff --git a/oslo_messaging/tests/functional/utils.py b/oslo_messaging/tests/functional/utils.py index ebdc725..b47ac5f 100644 --- a/oslo_messaging/tests/functional/utils.py +++ b/oslo_messaging/tests/functional/utils.py @@ -20,6 +20,7 @@ from oslo_config import cfg from six import moves import oslo_messaging +from oslo_messaging._drivers import kafka_options from oslo_messaging._drivers.zmq_driver import zmq_options from oslo_messaging.notify import notifier from oslo_messaging.tests import utils as test_utils @@ -305,23 +306,27 @@ class SkipIfNoTransportURL(test_utils.BaseTestCase): rpc_zmq_ipc_dir=zmq_ipc_dir) zmq_redis_port = os.environ.get('ZMQ_REDIS_PORT') if zmq_redis_port: - self.config(port=zmq_redis_port, group="matchmaker_redis") - self.config(check_timeout=10000, group="matchmaker_redis") - self.config(wait_timeout=1000, group="matchmaker_redis") + self.config(port=zmq_redis_port, + check_timeout=10000, + wait_timeout=1000, + group="matchmaker_redis") zmq_use_pub_sub = os.environ.get('ZMQ_USE_PUB_SUB') - self.config(use_pub_sub=zmq_use_pub_sub, - group='oslo_messaging_zmq') zmq_use_router_proxy = os.environ.get('ZMQ_USE_ROUTER_PROXY') - self.config(use_router_proxy=zmq_use_router_proxy, - group='oslo_messaging_zmq') zmq_use_acks = os.environ.get('ZMQ_USE_ACKS') - self.config(rpc_use_acks=zmq_use_acks, + self.config(use_pub_sub=zmq_use_pub_sub, + use_router_proxy=zmq_use_router_proxy, + rpc_use_acks=zmq_use_acks, group='oslo_messaging_zmq') zmq_use_dynamic_connections = \ os.environ.get('ZMQ_USE_DYNAMIC_CONNECTIONS') self.config(use_dynamic_connections=zmq_use_dynamic_connections, group='oslo_messaging_zmq') + kafka_options.register_opts(conf) + + self.config(producer_batch_size=0, + group='oslo_messaging_kafka') + class NotificationFixture(fixtures.Fixture): def __init__(self, conf, url, topics, batch=None): -- cgit v1.2.1