Suffix more places with _bytes

Author: blakeembrey

kafka/record/default_records.py

@@ -512,7 +512,7 @@ def size(self):
         """
         return len(self._buffer)
 
-    def size_in_bytes(self, offset, timestamp, key, value, headers):
+    def size_in_bytes(self, offset, timestamp, key_bytes, value_bytes, headers):
         if self._first_timestamp is not None:
             timestamp_delta = timestamp - self._first_timestamp
         else:
@@ -521,24 +521,24 @@ def size_in_bytes(self, offset, timestamp, key, value, headers):
             1 +  # Attrs
             size_of_varint(offset) +
             size_of_varint(timestamp_delta) +
-            self.size_of(key, value, headers)
+            self.size_of(key_bytes, value_bytes, headers)
         )
         return size_of_body + size_of_varint(size_of_body)
 
     @classmethod
-    def size_of(cls, key, value, headers):
+    def size_of(cls, key_bytes, value_bytes, headers):
         size = 0
         # Key size
-        if key is None:
+        if key_bytes is None:
             size += 1
         else:
-            key_len = len(key)
+            key_len = len(key_bytes)
             size += size_of_varint(key_len) + key_len
         # Value size
-        if value is None:
+        if value_bytes is None:
             size += 1
         else:
-            value_len = len(value)
+            value_len = len(value_bytes)
             size += size_of_varint(value_len) + value_len
         # Header size
         size += size_of_varint(len(headers))
@@ -554,12 +554,12 @@ def size_of(cls, key, value, headers):
         return size
 
     @classmethod
-    def estimate_size_in_bytes(cls, key, value, headers):
+    def estimate_size_in_bytes(cls, key_bytes, value_bytes, headers):
         """ Get the upper bound estimate on the size of record
         """
         return (
             cls.HEADER_STRUCT.size + cls.MAX_RECORD_OVERHEAD +
-            cls.size_of(key, value, headers)
+            cls.size_of(key_bytes, value_bytes, headers)
         )
 
 

kafka/record/legacy_records.py

@@ -329,7 +329,7 @@ def __init__(self, magic, compression_type, batch_size):
         self._batch_size = batch_size
         self._buffer = bytearray()
 
-    def append(self, offset, timestamp, key, value, headers=None):
+    def append(self, offset, timestamp, key_bytes, value_bytes, headers=None):
         """ Append message to batch.
         """
         assert not headers, "Headers not supported in v0/v1"
@@ -344,18 +344,18 @@ def append(self, offset, timestamp, key, value, headers=None):
             raise TypeError(
                 "`timestamp` should be int, but {} provided".format(
                     type(timestamp)))
-        if not (key is None or
-                isinstance(key, (bytes, bytearray, memoryview))):
+        if not (key_bytes is None or
+                isinstance(key_bytes, (bytes, bytearray, memoryview))):
             raise TypeError(
-                "Not supported type for key: {}".format(type(key)))
-        if not (value is None or
-                isinstance(value, (bytes, bytearray, memoryview))):
+                "Not supported type for key: {}".format(type(key_bytes)))
+        if not (value_bytes is None or
+                isinstance(value_bytes, (bytes, bytearray, memoryview))):
             raise TypeError(
-                "Not supported type for value: {}".format(type(value)))
+                "Not supported type for value: {}".format(type(value_bytes)))
 
         # Check if we have room for another message
         pos = len(self._buffer)
-        size = self.size_in_bytes(offset, timestamp, key, value)
+        size = self.size_in_bytes(offset, timestamp, key_bytes, value_bytes)
         # We always allow at least one record to be appended
         if offset != 0 and pos + size >= self._batch_size:
             return None
@@ -364,11 +364,11 @@ def append(self, offset, timestamp, key, value, headers=None):
         self._buffer.extend(bytearray(size))
 
         # Encode message
-        crc = self._encode_msg(pos, offset, timestamp, key, value)
+        crc = self._encode_msg(pos, offset, timestamp, key_bytes, value_bytes)
 
         return LegacyRecordMetadata(offset, crc, size, timestamp)
 
-    def _encode_msg(self, start_pos, offset, timestamp, key, value,
+    def _encode_msg(self, start_pos, offset, timestamp, key_bytes, value_bytes,
                     attributes=0):
         """ Encode msg data into the `msg_buffer`, which should be allocated
             to at least the size of this message.
@@ -380,24 +380,24 @@ def _encode_msg(self, start_pos, offset, timestamp, key, value,
         # Write key and value
         pos += self.KEY_OFFSET_V0 if magic == 0 else self.KEY_OFFSET_V1
 
-        if key is None:
+        if key_bytes is None:
             struct.pack_into(">i", buf, pos, -1)
             pos += self.KEY_LENGTH
         else:
-            key_size = len(key)
+            key_size = len(key_bytes)
             struct.pack_into(">i", buf, pos, key_size)
             pos += self.KEY_LENGTH
-            buf[pos: pos + key_size] = key
+            buf[pos: pos + key_size] = key_bytes
             pos += key_size
 
-        if value is None:
+        if value_bytes is None:
             struct.pack_into(">i", buf, pos, -1)
             pos += self.VALUE_LENGTH
         else:
-            value_size = len(value)
+            value_size = len(value_bytes)
             struct.pack_into(">i", buf, pos, value_size)
             pos += self.VALUE_LENGTH
-            buf[pos: pos + value_size] = value
+            buf[pos: pos + value_size] = value_bytes
             pos += value_size
         length = (pos - start_pos) - self.LOG_OVERHEAD
 
@@ -430,15 +430,15 @@ def _maybe_compress(self):
                 else:
                     compressed = lz4_encode(data)
             size = self.size_in_bytes(
-                0, timestamp=0, key=None, value=compressed)
+                0, timestamp=0, key_bytes=None, value_bytes=compressed)
             # We will try to reuse the same buffer if we have enough space
             if size > len(self._buffer):
                 self._buffer = bytearray(size)
             else:
                 del self._buffer[size:]
             self._encode_msg(
                 start_pos=0,
-                offset=0, timestamp=0, key=None, value=compressed,
+                offset=0, timestamp=0, key_bytes=None, value_bytes=compressed,
                 attributes=self._compression_type)
             return True
         return False
@@ -455,20 +455,20 @@ def size(self):
 
     # Size calculations. Just copied Java's implementation
 
-    def size_in_bytes(self, offset, timestamp, key, value, headers=None):
+    def size_in_bytes(self, offset, timestamp, key_bytes, value_bytes, headers=None):
         """ Actual size of message to add
         """
         assert not headers, "Headers not supported in v0/v1"
         magic = self._magic
-        return self.LOG_OVERHEAD + self.record_size(magic, key, value)
+        return self.LOG_OVERHEAD + self.record_size(magic, key_bytes, value_bytes)
 
     @classmethod
-    def record_size(cls, magic, key, value):
+    def record_size(cls, magic, key_bytes, value_bytes):
         message_size = cls.record_overhead(magic)
-        if key is not None:
-            message_size += len(key)
-        if value is not None:
-            message_size += len(value)
+        if key_bytes is not None:
+            message_size += len(key_bytes)
+        if value_bytes is not None:
+            message_size += len(value_bytes)
         return message_size
 
     @classmethod
@@ -480,17 +480,17 @@ def record_overhead(cls, magic):
             return cls.RECORD_OVERHEAD_V1
 
     @classmethod
-    def estimate_size_in_bytes(cls, magic, compression_type, key, value):
+    def estimate_size_in_bytes(cls, magic, compression_type, key_bytes, value_bytes):
         """ Upper bound estimate of record size.
         """
         assert magic in [0, 1], "Not supported magic"
         # In case of compression we may need another overhead for inner msg
         if compression_type:
             return (
                 cls.LOG_OVERHEAD + cls.record_overhead(magic) +
-                cls.record_size(magic, key, value)
+                cls.record_size(magic, key_bytes, value_bytes)
             )
-        return cls.LOG_OVERHEAD + cls.record_size(magic, key, value)
+        return cls.LOG_OVERHEAD + cls.record_size(magic, key_bytes, value_bytes)
 
 
 class LegacyRecordMetadata(object):