Raise UnsupportedVersionError from coordinator (#2579)

Author: dpkp

kafka/consumer/group.py

@@ -972,7 +972,8 @@ def unsubscribe(self):
         # are committed since there will be no following rebalance
         self._coordinator.maybe_auto_commit_offsets_now()
         self._subscription.unsubscribe()
-        self._coordinator.maybe_leave_group()
+        if self.config['api_version'] >= (0, 9):
+            self._coordinator.maybe_leave_group()
         self._client.cluster.need_all_topic_metadata = False
         self._client.set_topics([])
         log.debug("Unsubscribed all topics or patterns and assigned partitions")

kafka/coordinator/base.py

@@ -395,12 +395,16 @@ def ensure_active_group(self, timeout_ms=None):
 
         Raises: KafkaTimeoutError if timeout_ms is not None
         """
+        if self.config['api_version'] < (0, 9):
+            raise Errors.UnsupportedVersionError('Group Coordinator APIs require 0.9+ broker')
         inner_timeout_ms = timeout_ms_fn(timeout_ms, 'Timeout attempting to join consumer group')
         self.ensure_coordinator_ready(timeout_ms=inner_timeout_ms())
         self._start_heartbeat_thread()
         self.join_group(timeout_ms=inner_timeout_ms())
 
     def join_group(self, timeout_ms=None):
+        if self.config['api_version'] < (0, 9):
+            raise Errors.UnsupportedVersionError('Group Coordinator APIs require 0.9+ broker')
         inner_timeout_ms = timeout_ms_fn(timeout_ms, 'Timeout attempting to join consumer group')
         while self.need_rejoin():
             self.ensure_coordinator_ready(timeout_ms=inner_timeout_ms())
@@ -763,6 +767,8 @@ def request_rejoin(self):
         self.rejoin_needed = True
 
     def _start_heartbeat_thread(self):
+        if self.config['api_version'] < (0, 9):
+            raise Errors.UnsupportedVersionError('Heartbeat APIs require 0.9+ broker')
         with self._lock:
             if self._heartbeat_thread is None:
                 log.info('Starting new heartbeat thread')
@@ -794,10 +800,13 @@ def close(self, timeout_ms=None):
         """Close the coordinator, leave the current group,
         and reset local generation / member_id"""
         self._close_heartbeat_thread(timeout_ms=timeout_ms)
-        self.maybe_leave_group(timeout_ms=timeout_ms)
+        if self.config['api_version'] >= (0, 9):
+            self.maybe_leave_group(timeout_ms=timeout_ms)
 
     def maybe_leave_group(self, timeout_ms=None):
         """Leave the current group and reset local generation/memberId."""
+        if self.config['api_version'] < (0, 9):
+            raise Errors.UnsupportedVersionError('Group Coordinator APIs require 0.9+ broker')
         with self._client._lock, self._lock:
             if (not self.coordinator_unknown()
                 and self.state is not MemberState.UNJOINED

kafka/coordinator/consumer.py

@@ -494,7 +494,8 @@ def commit_offsets_async(self, offsets, callback=None):
         return future
 
     def _do_commit_offsets_async(self, offsets, callback=None):
-        assert self.config['api_version'] >= (0, 8, 1), 'Unsupported Broker API'
+        if self.config['api_version'] < (0, 8, 1):
+            raise Errors.UnsupportedVersionError('OffsetCommitRequest requires 0.8.1+ broker')
         assert all(map(lambda k: isinstance(k, TopicPartition), offsets))
         assert all(map(lambda v: isinstance(v, OffsetAndMetadata),
                        offsets.values()))
@@ -516,7 +517,8 @@ def commit_offsets_sync(self, offsets, timeout_ms=None):
 
         Raises error on failure
         """
-        assert self.config['api_version'] >= (0, 8, 1), 'Unsupported Broker API'
+        if self.config['api_version'] < (0, 8, 1):
+            raise Errors.UnsupportedVersionError('OffsetCommitRequest requires 0.8.1+ broker')
         assert all(map(lambda k: isinstance(k, TopicPartition), offsets))
         assert all(map(lambda v: isinstance(v, OffsetAndMetadata),
                        offsets.values()))
@@ -573,7 +575,8 @@ def _send_offset_commit_request(self, offsets):
         Returns:
             Future: indicating whether the commit was successful or not
         """
-        assert self.config['api_version'] >= (0, 8, 1), 'Unsupported Broker API'
+        if self.config['api_version'] < (0, 8, 1):
+            raise Errors.UnsupportedVersionError('OffsetCommitRequest requires 0.8.1+ broker')
         assert all(map(lambda k: isinstance(k, TopicPartition), offsets))
         assert all(map(lambda v: isinstance(v, OffsetAndMetadata),
                        offsets.values()))
@@ -761,7 +764,8 @@ def _send_offset_fetch_request(self, partitions):
         Returns:
             Future: resolves to dict of offsets: {TopicPartition: OffsetAndMetadata}
         """
-        assert self.config['api_version'] >= (0, 8, 1), 'Unsupported Broker API'
+        if self.config['api_version'] < (0, 8, 1):
+            raise Errors.UnsupportedVersionError('OffsetFetchRequest requires 0.8.1+ broker')
         assert all(map(lambda k: isinstance(k, TopicPartition), partitions))
         if not partitions:
             return Future().success({})