Expose selector type as config option

Author: dpkp

kafka/client_async.py

@@ -63,6 +63,7 @@ class KafkaClient(object):
         'ssl_crlfile': None,
         'api_version': None,
         'api_version_auto_timeout_ms': 2000,
+        'selector': selectors.DefaultSelector,
     }
     API_VERSIONS = [
         (0, 10),
@@ -134,6 +135,9 @@ def __init__(self, **configs):
             api_version_auto_timeout_ms (int): number of milliseconds to throw a
                 timeout exception from the constructor when checking the broker
                 api version. Only applies if api_version is None
+            selector (selectors.BaseSelector): Provide a specific selector
+                implementation to use for I/O multiplexing.
+                Default: selectors.DefaultSelector
         """
         self.config = copy.copy(self.DEFAULT_CONFIG)
         for key in self.config:
@@ -149,7 +153,7 @@ def __init__(self, **configs):
         self._topics = set() # empty set will fetch all topic metadata
         self._metadata_refresh_in_progress = False
         self._last_no_node_available_ms = 0
-        self._selector = selectors.DefaultSelector()
+        self._selector = self.config['selector']()
         self._conns = {}
         self._connecting = set()
         self._refresh_on_disconnects = True

kafka/consumer/group.py

@@ -6,7 +6,7 @@
 
 import six
 
-from kafka.client_async import KafkaClient
+from kafka.client_async import KafkaClient, selectors
 from kafka.consumer.fetcher import Fetcher
 from kafka.consumer.subscription_state import SubscriptionState
 from kafka.coordinator.consumer import ConsumerCoordinator
@@ -173,6 +173,9 @@ class KafkaConsumer(six.Iterator):
             metrics. Default: 2
         metrics_sample_window_ms (int): The number of samples maintained to
             compute metrics. Default: 30000
+        selector (selectors.BaseSelector): Provide a specific selector
+            implementation to use for I/O multiplexing.
+            Default: selectors.DefaultSelector
 
     Note:
         Configuration parameters are described in more detail at
@@ -218,6 +221,7 @@ class KafkaConsumer(six.Iterator):
         'metric_reporters': [],
         'metrics_num_samples': 2,
         'metrics_sample_window_ms': 30000,
+        'selector': selectors.DefaultSelector,
     }
 
     def __init__(self, *topics, **configs):

kafka/producer/kafka.py

@@ -8,7 +8,7 @@
 import weakref
 
 from .. import errors as Errors
-from ..client_async import KafkaClient
+from ..client_async import KafkaClient, selectors
 from ..partitioner.default import DefaultPartitioner
 from ..protocol.message import Message, MessageSet
 from ..structs import TopicPartition
@@ -220,6 +220,9 @@ class KafkaProducer(object):
         api_version_auto_timeout_ms (int): number of milliseconds to throw a
             timeout exception from the constructor when checking the broker
             api version. Only applies if api_version set to 'auto'
+        selector (selectors.BaseSelector): Provide a specific selector
+            implementation to use for I/O multiplexing.
+            Default: selectors.DefaultSelector
 
     Note:
         Configuration parameters are described in more detail at
@@ -255,7 +258,8 @@ class KafkaProducer(object):
         'ssl_keyfile': None,
         'ssl_crlfile': None,
         'api_version': None,
-        'api_version_auto_timeout_ms': 2000
+        'api_version_auto_timeout_ms': 2000,
+        'selector': selectors.DefaultSelector,
     }
 
     def __init__(self, **configs):