diff options
| -rw-r--r-- | kafka/coordinator/base.py | 9 | ||||
| -rw-r--r-- | kafka/coordinator/consumer.py | 36 | ||||
| -rw-r--r-- | test/test_coordinator.py | 5 |
3 files changed, 27 insertions, 23 deletions
diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py index 168115a..25dd000 100644 --- a/kafka/coordinator/base.py +++ b/kafka/coordinator/base.py @@ -50,6 +50,7 @@ class BaseCoordinator(object): 'session_timeout_ms': 30000, 'heartbeat_interval_ms': 3000, 'retry_backoff_ms': 100, + 'api_version': (0, 9), } def __init__(self, client, **configs): @@ -194,6 +195,14 @@ class BaseCoordinator(object): """ while self.coordinator_unknown(): + # Prior to 0.8.2 there was no group coordinator + # so we will just pick a node at random and treat + # it as the "coordinator" + if self.config['api_version'] < (0, 8, 2): + self.coordinator_id = self._client.least_loaded_node() + self._client.ready(self.coordinator_id) + continue + future = self._send_group_coordinator_request() self._client.poll(future=future) diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py index 196bcc7..083a36a 100644 --- a/kafka/coordinator/consumer.py +++ b/kafka/coordinator/consumer.py @@ -100,6 +100,12 @@ class ConsumerCoordinator(BaseCoordinator): interval = self.config['auto_commit_interval_ms'] / 1000.0 self._auto_commit_task = AutoCommitTask(weakref.proxy(self), interval) + # When using broker-coordinated consumer groups, auto-commit will + # be automatically enabled on group join (see _on_join_complete) + # Otherwise, we should enable now b/c there will be no group join + if self.config['api_version'] < (0, 9): + self._auto_commit_task.enable() + self._sensors = ConsumerCoordinatorMetrics(metrics, metric_group_prefix, self._subscription) @@ -293,8 +299,7 @@ class ConsumerCoordinator(BaseCoordinator): return {} while True: - if self.config['api_version'] >= (0, 8, 2): - self.ensure_coordinator_known() + self.ensure_coordinator_known() # contact coordinator to fetch committed offsets future = self._send_offset_fetch_request(partitions) @@ -356,8 +361,7 @@ class ConsumerCoordinator(BaseCoordinator): return while True: - if self.config['api_version'] >= (0, 8, 2): - self.ensure_coordinator_known() + self.ensure_coordinator_known() future = self._send_offset_commit_request(offsets) self._client.poll(future=future) @@ -415,14 +419,10 @@ class ConsumerCoordinator(BaseCoordinator): log.debug('No offsets to commit') return Future().success(True) - if self.config['api_version'] >= (0, 8, 2): - if self.coordinator_unknown(): - return Future().failure(Errors.GroupCoordinatorNotAvailableError) - node_id = self.coordinator_id - else: - node_id = self._client.least_loaded_node() - if node_id is None: - return Future().failure(Errors.NoBrokersAvailable) + elif self.coordinator_unknown(): + return Future().failure(Errors.GroupCoordinatorNotAvailableError) + + node_id = self.coordinator_id # create the offset commit request offset_data = collections.defaultdict(dict) @@ -571,14 +571,10 @@ class ConsumerCoordinator(BaseCoordinator): if not partitions: return Future().success({}) - if self.config['api_version'] >= (0, 8, 2): - if self.coordinator_unknown(): - return Future().failure(Errors.GroupCoordinatorNotAvailableError) - node_id = self.coordinator_id - else: - node_id = self._client.least_loaded_node() - if node_id is None: - return Future().failure(Errors.NoBrokersAvailable) + elif self.coordinator_unknown(): + return Future().failure(Errors.GroupCoordinatorNotAvailableError) + + node_id = self.coordinator_id # Verify node is ready if not self._client.ready(node_id): diff --git a/test/test_coordinator.py b/test/test_coordinator.py index 15b915d..735d278 100644 --- a/test/test_coordinator.py +++ b/test/test_coordinator.py @@ -425,8 +425,7 @@ def test_send_offset_commit_request_fail(patched_coord, offsets): ((0, 9), OffsetCommitRequest[2])]) def test_send_offset_commit_request_versions(patched_coord, offsets, api_version, req_type): - # assuming fixture sets coordinator=0, least_loaded_node=1 - expect_node = 0 if api_version >= (0, 8, 2) else 1 + expect_node = 0 patched_coord.config['api_version'] = api_version patched_coord._send_offset_commit_request(offsets) @@ -522,7 +521,7 @@ def test_send_offset_fetch_request_fail(patched_coord, partitions): def test_send_offset_fetch_request_versions(patched_coord, partitions, api_version, req_type): # assuming fixture sets coordinator=0, least_loaded_node=1 - expect_node = 0 if api_version >= (0, 8, 2) else 1 + expect_node = 0 patched_coord.config['api_version'] = api_version patched_coord._send_offset_fetch_request(partitions) |