Update kafka.util.crc32 to unsigned everywhere

kafka/protocol.py

@@ -98,7 +98,7 @@ class KafkaProtocol(object):
             msg += write_int_string(message.key)
             msg += write_int_string(message.value)
             crc = crc32(msg)
-            msg = struct.pack('>i%ds' % len(msg), crc, msg)
+            msg = struct.pack('>I%ds' % len(msg), crc, msg)
         else:
             raise ProtocolError("Unexpected magic number: %d" % message.magic)
         return msg
@@ -148,7 +148,7 @@ class KafkaProtocol(object):
         The offset is actually read from decode_message_set_iter (it is part
         of the MessageSet payload).
         """
-        ((crc, magic, att), cur) = relative_unpack('>iBB', data, 0)
+        ((crc, magic, att), cur) = relative_unpack('>IBB', data, 0)
         if crc != crc32(data[4:]):
             raise ChecksumError("Message checksum failed")
 

kafka/util.py

@@ -1,7 +1,7 @@
+import binascii
 import collections
 import struct
 import sys
-import zlib
 from threading import Thread, Event
 
 import six
@@ -10,16 +10,7 @@ from kafka.common import BufferUnderflowError
 
 
 def crc32(data):
-    """
-    Python 2 returns a value in the range [-2**31, 2**31-1].
-    Python 3 returns a value in the range [0, 2**32-1].
-
-    We want a consistent behavior so let's use python2's.
-    """
-    crc = zlib.crc32(data)
-    if six.PY3 and crc > 2**31:
-        crc -= 2 ** 32
-    return crc
+    return binascii.crc32(data) & 0xffffffff
 
 
 def write_int_string(s):