Always encode size with MessageSet

Author: dpkp

kafka/protocol/message.py

@@ -143,7 +143,7 @@ class MessageSet(AbstractType):
     HEADER_SIZE = 12 # offset + message_size
 
     @classmethod
-    def encode(cls, items, size=True, recalc_message_size=True):
+    def encode(cls, items):
         # RecordAccumulator encodes messagesets internally
         if isinstance(items, io.BytesIO):
             size = Int32.decode(items)
@@ -156,8 +156,6 @@ def encode(cls, items, size=True, recalc_message_size=True):
             encoded_values.append(Int64.encode(offset))
             encoded_values.append(Bytes.encode(message))
         encoded = b''.join(encoded_values)
-        if not size:
-            return encoded
         return Bytes.encode(encoded)
 
     @classmethod

test/test_protocol.py

@@ -72,8 +72,7 @@ def test_encode_message_set():
         Message(b'v2', key=b'k2')
     ]
     encoded = MessageSet.encode([(0, msg.encode())
-                                 for msg in messages],
-                                size=False)
+                                 for msg in messages])
     expect = b''.join([
         struct.pack('>q', 0),          # MsgSet Offset
         struct.pack('>i', 18),         # Msg Size
@@ -93,6 +92,7 @@ def test_encode_message_set():
         struct.pack('>i', 2),          # Length of value
         b'v2',                          # Value
     ])
+    expect = struct.pack('>i', len(expect)) + expect
     assert encoded == expect