diff --git a/kafka/protocol/types.py b/kafka/protocol/types.py
index 0e3685d73..39cac795a 100644
--- a/kafka/protocol/types.py
+++ b/kafka/protocol/types.py
@@ -2,6 +2,7 @@
 
 import struct
 from struct import error
+import uuid
 
 from kafka.protocol.abstract import AbstractType
 
@@ -90,6 +91,18 @@ def decode(cls, data):
         return _unpack(cls._unpack, data.read(8))
 
 
+class UUID(AbstractType):
+    @classmethod
+    def encode(cls, value):
+        if isinstance(value, uuid.UUID):
+            return value.bytes
+        return uuid.UUID(value).bytes
+
+    @classmethod
+    def decode(cls, data):
+        return uuid.UUID(bytes=data.read(16))
+
+
 class String(AbstractType):
     def __init__(self, encoding='utf-8'):
         self.encoding = encoding
@@ -348,7 +361,6 @@ def encode(cls, value):
 
 
 class CompactArray(Array):
-
     def encode(self, items):
         if items is None:
             return UnsignedVarInt32.encode(0)
@@ -362,4 +374,3 @@ def decode(self, data):
         if length == -1:
             return None
         return [self.array_of.decode(data) for _ in range(length)]
-