Rename CorruptRecordException -> CorruptRecordError

Author: dpkp

Diff for: kafka/consumer/fetcher.py

@@ -341,7 +341,7 @@ def fetched_records(self, max_records=None, update_offsets=True):
 
         Raises:
             OffsetOutOfRangeError: if no subscription offset_reset_strategy
-            CorruptRecordException: if message crc validation fails (check_crcs
+            CorruptRecordError: if message crc validation fails (check_crcs
                 must be set to True)
             RecordTooLargeError: if a message is larger than the currently
                 configured max_partition_fetch_bytes
@@ -925,7 +925,7 @@ def _unpack_records(self, tp, records, key_deserializer, value_deserializer):
                     last_batch = batch
 
                     if self.check_crcs and not batch.validate_crc():
-                        raise Errors.CorruptRecordException(
+                        raise Errors.CorruptRecordError(
                                 "Record batch for partition %s at offset %s failed crc check" % (
                                     self.topic_partition, batch.base_offset))
 
@@ -963,7 +963,7 @@ def _unpack_records(self, tp, records, key_deserializer, value_deserializer):
 
                     for record in batch:
                         if self.check_crcs and not record.validate_crc():
-                            raise Errors.CorruptRecordException(
+                            raise Errors.CorruptRecordError(
                                     "Record for partition %s at offset %s failed crc check" % (
                                         self.topic_partition, record.offset))
                         key_size = len(record.key) if record.key is not None else -1

Diff for: kafka/errors.py

@@ -120,14 +120,14 @@ class OffsetOutOfRangeError(BrokerResponseError):
                    ' maintained by the server for the given topic/partition.')
 
 
-class CorruptRecordException(BrokerResponseError):
+class CorruptRecordError(BrokerResponseError):
     errno = 2
     message = 'CORRUPT_MESSAGE'
     description = ('This message has failed its CRC checksum, exceeds the'
                    ' valid size, or is otherwise corrupt.')
 
 # Backward compatibility
-InvalidMessageError = CorruptRecordException
+CorruptRecordException = CorruptRecordError
 
 
 class UnknownTopicOrPartitionError(BrokerResponseError):

Diff for: kafka/record/default_records.py

@@ -60,7 +60,7 @@
 from kafka.record.util import (
     decode_varint, encode_varint, calc_crc32c, size_of_varint
 )
-from kafka.errors import CorruptRecordException, UnsupportedCodecError
+from kafka.errors import CorruptRecordError, UnsupportedCodecError
 from kafka.codec import (
     gzip_encode, snappy_encode, lz4_encode, zstd_encode,
     gzip_decode, snappy_decode, lz4_decode, zstd_decode
@@ -288,14 +288,14 @@ def _read_msg(
 
         header_count, pos = decode_varint(buffer, pos)
         if header_count < 0:
-            raise CorruptRecordException("Found invalid number of record "
+            raise CorruptRecordError("Found invalid number of record "
                                          "headers {}".format(header_count))
         headers = []
         while header_count:
             # Header key is of type String, that can't be None
             h_key_len, pos = decode_varint(buffer, pos)
             if h_key_len < 0:
-                raise CorruptRecordException(
+                raise CorruptRecordError(
                     "Invalid negative header key size {}".format(h_key_len))
             h_key = buffer[pos: pos + h_key_len].decode("utf-8")
             pos += h_key_len
@@ -313,7 +313,7 @@ def _read_msg(
 
         # validate whether we have read all header bytes in the current record
         if pos - start_pos != length:
-            raise CorruptRecordException(
+            raise CorruptRecordError(
                 "Invalid record size: expected to read {} bytes in record "
                 "payload, but instead read {}".format(length, pos - start_pos))
         self._pos = pos
@@ -332,14 +332,14 @@ def __iter__(self):
     def __next__(self):
         if self._next_record_index >= self._num_records:
             if self._pos != len(self._buffer):
-                raise CorruptRecordException(
+                raise CorruptRecordError(
                     "{} unconsumed bytes after all records consumed".format(
                         len(self._buffer) - self._pos))
             raise StopIteration
         try:
             msg = self._read_msg()
         except (ValueError, IndexError) as err:
-            raise CorruptRecordException(
+            raise CorruptRecordError(
                 "Found invalid record structure: {!r}".format(err))
         else:
             self._next_record_index += 1

Diff for: kafka/record/legacy_records.py

@@ -52,7 +52,7 @@
     gzip_decode, snappy_decode, lz4_decode, lz4_decode_old_kafka,
 )
 import kafka.codec as codecs
-from kafka.errors import CorruptRecordException, UnsupportedCodecError
+from kafka.errors import CorruptRecordError, UnsupportedCodecError
 
 
 class LegacyRecordBase(object):
@@ -191,7 +191,7 @@ def _decompress(self, key_offset):
         value_size = struct.unpack_from(">i", self._buffer, pos)[0]
         pos += self.VALUE_LENGTH
         if value_size == -1:
-            raise CorruptRecordException("Value of compressed message is None")
+            raise CorruptRecordError("Value of compressed message is None")
         else:
             data = self._buffer[pos:pos + value_size]
 

Diff for: kafka/record/memory_records.py

@@ -22,7 +22,7 @@
 
 import struct
 
-from kafka.errors import CorruptRecordException, IllegalStateError, UnsupportedVersionError
+from kafka.errors import CorruptRecordError, IllegalStateError, UnsupportedVersionError
 from kafka.record.abc import ABCRecords
 from kafka.record.legacy_records import LegacyRecordBatch, LegacyRecordBatchBuilder
 from kafka.record.default_records import DefaultRecordBatch, DefaultRecordBatchBuilder
@@ -99,7 +99,7 @@ def next_batch(self, _min_slice=MIN_SLICE,
         if next_slice is None:
             return None
         if len(next_slice) < _min_slice:
-            raise CorruptRecordException(
+            raise CorruptRecordError(
                 "Record size is less than the minimum record overhead "
                 "({})".format(_min_slice - self.LOG_OVERHEAD))
         self._cache_next()

Diff for: test/record/test_records.py

@@ -2,7 +2,7 @@
 from __future__ import unicode_literals
 import pytest
 from kafka.record import MemoryRecords, MemoryRecordsBuilder
-from kafka.errors import CorruptRecordException
+from kafka.errors import CorruptRecordError
 
 from test.testutil import maybe_skip_unsupported_compression
 
@@ -174,7 +174,7 @@ def test_memory_records_corrupt():
         b"\x00\x00\x00\x03"  # Length=3
         b"\xfe\xb0\x1d",  # Some random bytes
     )
-    with pytest.raises(CorruptRecordException):
+    with pytest.raises(CorruptRecordError):
         records.next_batch()