Add record.validate_crc() for v0/v1 crc checks

Author: dpkp

kafka/record/abc.py

@@ -45,6 +45,11 @@ def checksum(self):
             be the checksum for v0 and v1 and None for v2 and above.
         """
 
+    @abc.abstractmethod
+    def validate_crc(self):
+        """ Return True if v0/v1 record matches checksum. noop/True for v2 records
+        """
+
     @abc.abstractproperty
     def headers(self):
         """ If supported by version list of key-value tuples, or empty list if

kafka/record/default_records.py

@@ -366,6 +366,9 @@ def headers(self):
     def checksum(self):
         return None
 
+    def validate_crc(self):
+        return True
+
     def __repr__(self):
         return (
             "DefaultRecord(offset={!r}, timestamp={!r}, timestamp_type={!r},"

kafka/record/legacy_records.py

@@ -236,6 +236,9 @@ def _read_key_value(self, pos):
             value = self._buffer[pos:pos + value_size].tobytes()
         return key, value
 
+    def _crc_bytes(self, msg_pos, length):
+        return self._buffer[msg_pos + self.MAGIC_OFFSET:msg_pos + self.LOG_OVERHEAD + length]
+
     def __iter__(self):
         if self._magic == 1:
             key_offset = self.KEY_OFFSET_V1
@@ -259,7 +262,7 @@ def __iter__(self):
                 absolute_base_offset = -1
 
             for header, msg_pos in headers:
-                offset, _, crc, _, attrs, timestamp = header
+                offset, length, crc, _, attrs, timestamp = header
                 # There should only ever be a single layer of compression
                 assert not attrs & self.CODEC_MASK, (
                     'MessageSet at offset %d appears double-compressed. This '
@@ -275,29 +278,32 @@ def __iter__(self):
                     offset += absolute_base_offset
 
                 key, value = self._read_key_value(msg_pos + key_offset)
+                crc_bytes = self._crc_bytes(msg_pos, length)
                 yield LegacyRecord(
                     self._magic, offset, timestamp, timestamp_type,
-                    key, value, crc)
+                    key, value, crc, crc_bytes)
         else:
             key, value = self._read_key_value(key_offset)
+            crc_bytes = self._crc_bytes(0, len(self._buffer) - self.LOG_OVERHEAD)
             yield LegacyRecord(
                 self._magic, self._offset, self._timestamp, timestamp_type,
-                key, value, self._crc)
+                key, value, self._crc, crc_bytes)
 
 
 class LegacyRecord(ABCRecord):
 
     __slots__ = ("_magic", "_offset", "_timestamp", "_timestamp_type", "_key", "_value",
-                 "_crc")
+                 "_crc", "_crc_bytes")
 
-    def __init__(self, magic, offset, timestamp, timestamp_type, key, value, crc):
+    def __init__(self, magic, offset, timestamp, timestamp_type, key, value, crc, crc_bytes):
         self._magic = magic
         self._offset = offset
         self._timestamp = timestamp
         self._timestamp_type = timestamp_type
         self._key = key
         self._value = value
         self._crc = crc
+        self._crc_bytes = crc_bytes
 
     @property
     def magic(self):
@@ -339,6 +345,10 @@ def headers(self):
     def checksum(self):
         return self._crc
 
+    def validate_crc(self):
+        crc = calc_crc32(self._crc_bytes)
+        return self._crc == crc
+
     @property
     def size_in_bytes(self):
         return LegacyRecordBatchBuilder.estimate_size_in_bytes(self._magic, None, self._key, self._value)