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):

test/record/test_default_records.py

@@ -20,7 +20,7 @@ def test_read_write_serde_v2(compression_type):
     headers = []  # [("header1", b"aaa"), ("header2", b"bbb")]
     for offset in range(10):
         builder.append(
-            offset, timestamp=9999999, key=b"test", value=b"Super",
+            offset, timestamp=9999999, key_bytes=b"test", value_bytes=b"Super",
             headers=headers)
     buffer = builder.build()
     reader = DefaultRecordBatch(bytes(buffer))
@@ -40,20 +40,20 @@ def test_read_write_serde_v2(compression_type):
 
 
 def test_written_bytes_equals_size_in_bytes_v2():
-    key = b"test"
-    value = b"Super"
+    key_bytes = b"test"
+    value_bytes = b"Super"
     headers = [("header1", b"aaa"), ("header2", b"bbb"), ("xx", None)]
     builder = DefaultRecordBatchBuilder(
         magic=2, compression_type=0, is_transactional=0,
         producer_id=-1, producer_epoch=-1, base_sequence=-1,
         batch_size=999999)
 
     size_in_bytes = builder.size_in_bytes(
-        0, timestamp=9999999, key=key, value=value, headers=headers)
+        0, timestamp=9999999, key_bytes=key_bytes, value_bytes=value_bytes, headers=headers)
 
     pos = builder.size()
     meta = builder.append(
-        0, timestamp=9999999, key=key, value=value, headers=headers)
+        0, timestamp=9999999, key_bytes=key_bytes, value_bytes=value_bytes, headers=headers)
 
     assert builder.size() - pos == size_in_bytes
     assert meta.size == size_in_bytes
@@ -71,7 +71,7 @@ def test_estimate_size_in_bytes_bigger_than_batch_v2():
         producer_id=-1, producer_epoch=-1, base_sequence=-1,
         batch_size=999999)
     builder.append(
-        0, timestamp=9999999, key=key, value=value, headers=headers)
+        0, timestamp=9999999, key_bytes=key, value_bytes=value, headers=headers)
     buf = builder.build()
     assert len(buf) <= estimate_size, \
         "Estimate should always be upper bound"
@@ -86,35 +86,35 @@ def test_default_batch_builder_validates_arguments():
     # Key should not be str
     with pytest.raises(TypeError):
         builder.append(
-            0, timestamp=9999999, key="some string", value=None, headers=[])
+            0, timestamp=9999999, key_bytes="some string", value_bytes=None, headers=[])
 
     # Value should not be str
     with pytest.raises(TypeError):
         builder.append(
-            0, timestamp=9999999, key=None, value="some string", headers=[])
+            0, timestamp=9999999, key_bytes=None, value_bytes="some string", headers=[])
 
     # Timestamp should be of proper type
     with pytest.raises(TypeError):
         builder.append(
-            0, timestamp="1243812793", key=None, value=b"some string",
+            0, timestamp="1243812793", key_bytes=None, value_bytes=b"some string",
             headers=[])
 
     # Offset of invalid type
     with pytest.raises(TypeError):
         builder.append(
-            "0", timestamp=9999999, key=None, value=b"some string", headers=[])
+            "0", timestamp=9999999, key_bytes=None, value_bytes=b"some string", headers=[])
 
     # Ok to pass value as None
     builder.append(
-        0, timestamp=9999999, key=b"123", value=None, headers=[])
+        0, timestamp=9999999, key_bytes=b"123", value_bytes=None, headers=[])
 
     # Timestamp can be None
     builder.append(
-        1, timestamp=None, key=None, value=b"some string", headers=[])
+        1, timestamp=None, key_bytes=None, value_bytes=b"some string", headers=[])
 
     # Ok to pass offsets in not incremental order. This should not happen thou
     builder.append(
-        5, timestamp=9999999, key=b"123", value=None, headers=[])
+        5, timestamp=9999999, key_bytes=b"123", value_bytes=None, headers=[])
 
     # in case error handling code fails to fix inner buffer in builder
     assert len(builder.build()) == 104
@@ -126,7 +126,7 @@ def test_default_correct_metadata_response():
         producer_id=-1, producer_epoch=-1, base_sequence=-1,
         batch_size=1024 * 1024)
     meta = builder.append(
-        0, timestamp=9999999, key=b"test", value=b"Super", headers=[])
+        0, timestamp=9999999, key_bytes=b"test", value_bytes=b"Super", headers=[])
 
     assert meta.offset == 0
     assert meta.timestamp == 9999999
@@ -146,7 +146,7 @@ def test_default_batch_size_limit():
         batch_size=1024)
 
     meta = builder.append(
-        0, timestamp=None, key=None, value=b"M" * 2000, headers=[])
+        0, timestamp=None, key_bytes=None, value_bytes=b"M" * 2000, headers=[])
     assert meta.size > 0
     assert meta.crc is None
     assert meta.offset == 0
@@ -158,12 +158,12 @@ def test_default_batch_size_limit():
         producer_id=-1, producer_epoch=-1, base_sequence=-1,
         batch_size=1024)
     meta = builder.append(
-        0, timestamp=None, key=None, value=b"M" * 700, headers=[])
+        0, timestamp=None, key_bytes=None, value_bytes=b"M" * 700, headers=[])
     assert meta is not None
     meta = builder.append(
-        1, timestamp=None, key=None, value=b"M" * 700, headers=[])
+        1, timestamp=None, key_bytes=None, value_bytes=b"M" * 700, headers=[])
     assert meta is None
     meta = builder.append(
-        2, timestamp=None, key=None, value=b"M" * 700, headers=[])
+        2, timestamp=None, key_bytes=None, value_bytes=b"M" * 700, headers=[])
     assert meta is None
     assert len(builder.build()) < 1000