diff options
| -rw-r--r-- | kafka/coordinator/base.py | 9 | ||||
| -rw-r--r-- | kafka/coordinator/consumer.py | 30 | ||||
| -rw-r--r-- | test/test_coordinator.py | 5 |
3 files changed, 21 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 44d4c6c..083a36a 100644 --- a/kafka/coordinator/consumer.py +++ b/kafka/coordinator/consumer.py @@ -299,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) @@ -362,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) @@ -421,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) @@ -577,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) |