diff --git a/kafka/protocol/schemas/fields/simple.py b/kafka/protocol/schemas/fields/simple.py
index c8982af84..8d3a7416d 100644
--- a/kafka/protocol/schemas/fields/simple.py
+++ b/kafka/protocol/schemas/fields/simple.py
@@ -110,11 +110,14 @@ def to_json(self, val):
         if self._type is UUID:
             return str(val)
         elif self._type is Bytes:
-            if not isinstance(val, (bytes, bytearray, memoryview)):
-                val = val.encode()
-            if not isinstance(val, memoryview):
+            if isinstance(val, memoryview):
                 val = val.tobytes()
+            if hasattr(val, 'to_dict'):
+                return val.to_dict()
+            elif not isinstance(val, (bytes, bytearray)):
+                val = val.encode()
             return val.decode(errors='backslashreplace')
+
         elif self._type is BitField:
             return list(val)
         else: