Prefer raised Exceptions to assert / AssertionError (#3042)

Author: dpkp

kafka/cli/admin/partitions/list_offsets.py

@@ -42,10 +42,12 @@ def command(cls, client, args):
     @classmethod
     def _parse_partition_specs(cls, client, args):
         if args.partitions:
-            assert not args.topic and not args.spec, "Either --partition or (--topic and --spec) is supported, but not both."
+            if args.topic or args.spec:
+                raise ValueError("Either --partition or (--topic and --spec) is supported, but not both.")
             partitions = args.partitions
         else:
-            assert args.topic and args.spec, "Both --topic and --spec must be provided."
+            if not args.topic or not args.spec:
+                raise ValueError("Both --topic and --spec must be provided.")
             partitions = [f'{args.topic}:*:{args.spec}']
         tp_offsets = {}
         for entry in partitions:

kafka/codec.py

@@ -258,7 +258,8 @@ def lz4f_decode(payload):
 
 def lz4_encode_old_kafka(payload):
     """Encode payload for 0.8/0.9 brokers -- requires an incorrect header checksum."""
-    assert xxhash is not None
+    if xxhash is None:
+        raise RuntimeError('pip install xxhash for lz4 encoding with 0.8/0.9 brokers')
     data = lz4_encode(payload)
     header_size = 7
     flg = data[4]
@@ -288,7 +289,8 @@ def lz4_encode_old_kafka(payload):
 
 
 def lz4_decode_old_kafka(payload):
-    assert xxhash is not None
+    if xxhash is None:
+        raise RuntimeError('pip install xxhash for lz4 encoding with 0.8/0.9 brokers')
     # Kafka's LZ4 code has a bug in its header checksum implementation
     header_size = 7
     if isinstance(payload[4], int):

kafka/consumer/fetcher.py

@@ -460,6 +460,7 @@ def fetched_records(self, max_records=None, update_offsets=True):
                 configured max_partition_fetch_bytes
             TopicAuthorizationError: if consumer is not authorized to fetch
                 messages from the topic
+            ValueError: if max_records is <= 0
 
         Returns: (records (dict), partial (bool))
             records: {TopicPartition: [messages]}
@@ -469,7 +470,8 @@ def fetched_records(self, max_records=None, update_offsets=True):
         """
         if max_records is None:
             max_records = self.config['max_poll_records']
-        assert max_records > 0
+        if max_records <= 0:
+            raise ValueError('max_records must be > 0')
 
         if self._next_in_line_exception_metadata is not None:
             exc_meta = self._next_in_line_exception_metadata
@@ -783,6 +785,7 @@ def _handle_list_offsets_response(self, response):
 
         Raises:
             TopicAuthorizationFailedError: if any topic returned an auth error
+            ValueError: if ListOffsetsResponse v0 and > 1 offset returned
         """
         fetched_offsets = dict()
         partitions_to_retry = set()
@@ -795,7 +798,8 @@ def _handle_list_offsets_response(self, response):
                 if error_type is Errors.NoError:
                     if response.API_VERSION == 0:
                         offsets = partition_info.old_style_offsets
-                        assert len(offsets) <= 1, 'Expected ListOffsetsResponse with one offset'
+                        if len(offsets) > 1:
+                            raise ValueError('Expected ListOffsetsResponse with one offset')
                         offset = offsets[0] if offsets else UNKNOWN_OFFSET
                     else:
                         offset = partition_info.offset

kafka/consumer/group.py

@@ -608,11 +608,17 @@ def commit_async(self, offsets=None, callback=None):
                 struct. This callback can be used to trigger custom actions when
                 a commit request completes.
 
+        Raises:
+            IncompatibleBrokerVersion: if broker version < 0.8.1
+            IllegalStateError: if group_id is None
+
         Returns:
             kafka.future.Future
         """
-        assert self.config['api_version'] >= (0, 8, 1), 'Requires >= Kafka 0.8.1'
-        assert self.config['group_id'] is not None, 'Requires group_id'
+        if self.config['api_version'] < (0, 8, 1):
+            raise Errors.IncompatibleBrokerVersion('Requires >= Kafka 0.8.1')
+        if self.config['group_id'] is None:
+            raise Errors.IllegalStateError('Requires group_id')
         if offsets is None:
             offsets = self._subscription.all_consumed_offsets()
         log.debug("Committing offsets: %s", offsets)
@@ -639,9 +645,15 @@ def commit(self, offsets=None, timeout_ms=None):
             offsets (dict, optional): {TopicPartition: OffsetAndMetadata} dict
                 to commit with the configured group_id. Defaults to currently
                 consumed offsets for all subscribed partitions.
+
+        Raises:
+            IncompatibleBrokerVersion: if broker version < 0.8.1
+            IllegalStateError: if group_id is None
         """
-        assert self.config['api_version'] >= (0, 8, 1), 'Requires >= Kafka 0.8.1'
-        assert self.config['group_id'] is not None, 'Requires group_id'
+        if self.config['api_version'] < (0, 8, 1):
+            raise Errors.IncompatibleBrokerVersion('Requires >= Kafka 0.8.1')
+        if self.config['group_id'] is None:
+            raise Errors.IllegalStateError('Requires group_id')
         if offsets is None:
             offsets = self._subscription.all_consumed_offsets()
         self._coordinator.commit_offsets_sync(offsets, timeout_ms=timeout_ms)
@@ -680,11 +692,16 @@ def committed(self, partition, metadata=False, timeout_ms=None):
             The last committed offset (int or OffsetAndMetadata), or None if there was no prior commit.
 
         Raises:
-            KafkaTimeoutError if timeout_ms provided
-            BrokerResponseErrors if OffsetFetchRequest raises an error.
+            IncompatibleBrokerVersion: if broker version < 0.8.1
+            IllegalStateError: if group_id is None
+            TypeError: if partition is not TopicPartition
+            KafkaTimeoutError: if timeout_ms provided
+            BrokerResponseError: if OffsetFetchRequest raises an error.
         """
-        assert self.config['api_version'] >= (0, 8, 1), 'Requires >= Kafka 0.8.1'
-        assert self.config['group_id'] is not None, 'Requires group_id'
+        if self.config['api_version'] < (0, 8, 1):
+            raise Errors.IncompatibleBrokerVersion('Requires >= Kafka 0.8.1')
+        if self.config['group_id'] is None:
+            raise Errors.IllegalStateError('Requires group_id')
         if not isinstance(partition, TopicPartition):
             raise TypeError('partition must be a TopicPartition namedtuple')
         committed = self._coordinator.fetch_committed_offsets([partition], timeout_ms=timeout_ms)
@@ -755,6 +772,11 @@ def poll(self, timeout_ms=0, max_records=None, update_offsets=True):
                 in a single call to :meth:`~kafka.KafkaConsumer.poll`.
                 Default: Inherit value from max_poll_records.
 
+        Raises:
+            ValueError: if timeout is < 0 or max_records <= 0.
+            TypeError: if max_records is not int.
+            IllegalStateError: if consumer already closed.
+
         Returns:
             dict[TopicPartition, list[ConsumerRecord]]: records since the last
                 fetch for the subscribed list of topics and partitions.
@@ -765,12 +787,16 @@ def poll(self, timeout_ms=0, max_records=None, update_offsets=True):
         # updated until the iterator returns each record to the user. As such,
         # the argument is not documented and should not be relied on by library
         # users to not break in the future.
-        assert timeout_ms >= 0, 'Timeout must not be negative'
+        if timeout_ms < 0:
+            raise ValueError('Timeout must not be negative')
         if max_records is None:
             max_records = self.config['max_poll_records']
-        assert isinstance(max_records, int), 'max_records must be an integer'
-        assert max_records > 0, 'max_records must be positive'
-        assert not self._closed, 'KafkaConsumer is closed'
+        if not isinstance(max_records, int):
+            raise TypeError('max_records must be an integer')
+        if max_records <= 0:
+            raise ValueError('max_records must be positive')
+        if self._closed:
+            raise Errors.IllegalStateError('KafkaConsumer is closed')
 
         # Poll for new data until the timeout expires
         timer = Timer(timeout_ms)
@@ -830,12 +856,17 @@ def position(self, partition, timeout_ms=None):
         Arguments:
             partition (TopicPartition): Partition to check
 
+        Raises:
+            TypeError: if partition is not a TopicPartition.
+            IllegalStateError: if partition is not assigned.
+
         Returns:
             int: Offset or None
         """
         if not isinstance(partition, TopicPartition):
             raise TypeError('partition must be a TopicPartition namedtuple')
-        assert self._subscription.is_assigned(partition), 'Partition is not assigned'
+        if not self._subscription.is_assigned(partition):
+            raise Errors.IllegalStateError('Partition is not assigned')
 
         timer = Timer(timeout_ms)
         # Phase 1: blocking refresh of committed offsets (network round-trip
@@ -883,12 +914,17 @@ def highwater(self, partition):
         Arguments:
             partition (TopicPartition): Partition to check
 
+        Raises:
+            TypeError: if partition is not a TopicPartition.
+            IllegalStateError: if partition is not assigned.
+
         Returns:
             int or None: Offset if available
         """
         if not isinstance(partition, TopicPartition):
             raise TypeError('partition must be a TopicPartition namedtuple')
-        assert self._subscription.is_assigned(partition), 'Partition is not assigned'
+        if not self._subscription.is_assigned(partition):
+            raise Errors.IllegalStateError('Partition is not assigned')
         return self._subscription.assignment[partition].highwater
 
     def pause(self, *partitions):
@@ -950,13 +986,15 @@ def seek(self, partition, offset):
             offset (int): Message offset in partition
 
         Raises:
-            AssertionError: If offset is not an int >= 0; or if partition is not
+            ValueError: If offset is not an int >= 0; or if partition is not
                 currently assigned.
         """
         if not isinstance(partition, TopicPartition):
             raise TypeError('partition must be a TopicPartition namedtuple')
-        assert isinstance(offset, int) and offset >= 0, 'Offset must be >= 0'
-        assert partition in self._subscription.assigned_partitions(), 'Unassigned partition'
+        if not isinstance(offset, int) or offset < 0:
+            raise ValueError('Offset must be int >= 0')
+        if partition not in self._subscription.assigned_partitions():
+            raise ValueError('Unassigned partition')
         log.debug("Seeking to offset %s for partition %s", offset, partition)
         self._subscription.assignment[partition].seek(offset)
         self._iterator = None
@@ -969,17 +1007,19 @@ def seek_to_beginning(self, *partitions):
                 default to all assigned partitions.
 
         Raises:
-            AssertionError: If any partition is not currently assigned, or if
+            ValueError: If any partition is not currently assigned, or if
                 no partitions are assigned.
         """
         if not all([isinstance(p, TopicPartition) for p in partitions]):
             raise TypeError('partitions must be TopicPartition namedtuples')
         if not partitions:
             partitions = self._subscription.assigned_partitions()
-            assert partitions, 'No partitions are currently assigned'
+            if not partitions:
+                raise ValueError('No partitions are currently assigned')
         else:
             for p in partitions:
-                assert p in self._subscription.assigned_partitions(), 'Unassigned partition'
+                if p not in self._subscription.assigned_partitions():
+                    raise ValueError('Unassigned partition: %s' % (p,))
 
         for tp in partitions:
             log.debug("Seeking to beginning of partition %s", tp)
@@ -994,17 +1034,19 @@ def seek_to_end(self, *partitions):
                 default to all assigned partitions.
 
         Raises:
-            AssertionError: If any partition is not currently assigned, or if
+            ValueError: If any partition is not currently assigned, or if
                 no partitions are assigned.
         """
         if not all([isinstance(p, TopicPartition) for p in partitions]):
             raise TypeError('partitions must be TopicPartition namedtuples')
         if not partitions:
             partitions = self._subscription.assigned_partitions()
-            assert partitions, 'No partitions are currently assigned'
+            if not partitions:
+                raise ValueError('No partitions are currently assigned')
         else:
             for p in partitions:
-                assert p in self._subscription.assigned_partitions(), 'Unassigned partition'
+                if p not in self._subscription.assigned_partitions():
+                    raise ValueError('Unassigned partition: %s' % (p,))
 
         for tp in partitions:
             log.debug("Seeking to end of partition %s", tp)
@@ -1049,7 +1091,7 @@ def subscribe(self, topics=(), pattern=None, listener=None):
         Raises:
             IllegalStateError: If called after previously calling
                 :meth:`~kafka.KafkaConsumer.assign`.
-            AssertionError: If neither topics or pattern is provided.
+            ValueError: If neither topics or pattern is provided.
             TypeError: If listener is not a ConsumerRebalanceListener.
         """
         # SubscriptionState handles error checking

kafka/consumer/subscription_state.py

@@ -118,8 +118,15 @@ def subscribe(self, topics=(), pattern=None, listener=None):
                 any listener set in a previous call to subscribe. It is
                 guaranteed, however, that the partitions revoked/assigned
                 through this interface are from topics subscribed in this call.
+
+        Raises:
+            ValueError: if neither topics nor pattern provided.
+            IllegalStateError: if both topics and pattern provided.
+            TypeError: if topics is not a list/sequence, or listener is not
+                a AsyncConsumerRebalanceListener or ConsumerRebalanceListener.
         """
-        assert topics or pattern, 'Must provide topics or pattern'
+        if not topics and not pattern:
+            raise ValueError('Must provide topics or pattern')
         if (topics and pattern):
             raise Errors.IllegalStateError(self._SUBSCRIPTION_EXCEPTION_MESSAGE)
 
@@ -191,7 +198,8 @@ def reset_group_subscription(self):
         """Reset the group's subscription to only contain topics subscribed by this consumer."""
         if not self.partitions_auto_assigned():
             raise Errors.IllegalStateError(self._SUBSCRIPTION_EXCEPTION_MESSAGE)
-        assert self.subscription is not None, 'Subscription required'
+        if self.subscription is None:
+            raise Errors.IllegalStateError('Subscription required')
         self._group_subscription.intersection_update(self.subscription)
 
     @synchronized
@@ -571,8 +579,10 @@ def __init__(self):
         self._pending_revocation = False
 
     def _set_position(self, offset):
-        assert self.has_valid_position, 'Valid position required'
-        assert isinstance(offset, OffsetAndMetadata)
+        if not self.has_valid_position:
+            raise Errors.IllegalStateError('Valid position required')
+        if not isinstance(offset, OffsetAndMetadata):
+            raise TypeError('offset must be OffsetAndMetadata')
         self._position = offset
 
     def _get_position(self):
@@ -581,7 +591,8 @@ def _get_position(self):
     position = property(_get_position, _set_position, None, "last position")
 
     def reset(self, strategy):
-        assert strategy is not None
+        if strategy is None:
+            raise ValueError('strategy cannot be None')
         self.reset_strategy = strategy
         self._position = None
         self.next_allowed_retry_time = None

kafka/coordinator/assignors/sticky/partition_movements.py

@@ -49,7 +49,8 @@ def move_partition(self, partition, old_consumer, new_consumer):
         if partition in self.partition_movements:
             # this partition has previously moved
             existing_pair = self._remove_movement_record_of_partition(partition)
-            assert existing_pair.dst_member_id == old_consumer
+            if existing_pair.dst_member_id != old_consumer:
+                raise ValueError()
             if existing_pair.src_member_id != new_consumer:
                 # the partition is not moving back to its previous consumer
                 self._add_partition_movement_record(
@@ -63,7 +64,8 @@ def get_partition_to_be_moved(self, partition, old_consumer, new_consumer):
             return partition
         if partition in self.partition_movements:
             # this partition has previously moved
-            assert old_consumer == self.partition_movements[partition].dst_member_id
+            if old_consumer != self.partition_movements[partition].dst_member_id:
+                raise ValueError()
             old_consumer = self.partition_movements[partition].src_member_id
         reverse_pair = ConsumerPair(src_member_id=new_consumer, dst_member_id=old_consumer)
         if reverse_pair not in self.partition_movements_by_topic[partition.topic]: