diff --git a/kafka/protocol/new/schemas/fields/simple.py b/kafka/protocol/new/schemas/fields/simple.py
index 0dc288790..a3c1b5398 100644
--- a/kafka/protocol/new/schemas/fields/simple.py
+++ b/kafka/protocol/new/schemas/fields/simple.py
@@ -60,7 +60,7 @@ def _calculate_default(self, default):
             return int(default)
         elif self._type is UUID:
             if not default:
-                return UUID.ZERO_UUID
+                return None
             else:
                 return uuid.UUID(default)
         elif self._type is Float64:
diff --git a/kafka/protocol/types.py b/kafka/protocol/types.py
index 293c3266e..28369f29e 100644
--- a/kafka/protocol/types.py
+++ b/kafka/protocol/types.py
@@ -102,13 +102,18 @@ class UUID(AbstractType):
 
     @classmethod
     def encode(cls, value):
+        if value is None:
+            value = cls.ZERO_UUID
         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))
+        val = uuid.UUID(bytes=data.read(16))
+        if val == cls.ZERO_UUID:
+            return None
+        return val
 
 
 class String(AbstractType):
diff --git a/test/protocol/new/admin/test_new_admin.py b/test/protocol/new/admin/test_new_admin.py
index 5fb0b1c31..df8a9ae51 100644
--- a/test/protocol/new/admin/test_new_admin.py
+++ b/test/protocol/new/admin/test_new_admin.py
@@ -60,7 +60,7 @@ def test_create_topics_response_roundtrip(version):
         topics=[
             Topic(
                 name="test-topic",
-                topic_id=uuid.uuid4() if version >= 7 else uuid.UUID(int=0),
+                topic_id=uuid.uuid4() if version >= 7 else None,
                 error_code=13,
                 error_message='foo' if version >= 1 else '',
                 topic_config_error_code=2 if version >= 5 else 0,