Refactor MessageSet and Message into LegacyRecordBatch to later support v2 message format

1 files changed, 5 insertions, 6 deletions

diff --git a/kafka/protocol/fetch.py b/kafka/protocol/fetch.py
index 359f197..0b03845 100644
--- a/kafka/protocol/fetch.py
+++ b/kafka/protocol/fetch.py
@@ -1,8 +1,7 @@
 from __future__ import absolute_import
 
 from .api import Request, Response
-from .message import MessageSet
-from .types import Array, Int8, Int16, Int32, Int64, Schema, String
+from .types import Array, Int8, Int16, Int32, Int64, Schema, String, Bytes
 
 
 class FetchResponse_v0(Response):
@@ -15,7 +14,7 @@ class FetchResponse_v0(Response):
                 ('partition', Int32),
                 ('error_code', Int16),
                 ('highwater_offset', Int64),
-                ('message_set', MessageSet)))))
+                ('message_set', Bytes)))))
     )
 
 
@@ -30,7 +29,7 @@ class FetchResponse_v1(Response):
                 ('partition', Int32),
                 ('error_code', Int16),
                 ('highwater_offset', Int64),
-                ('message_set', MessageSet)))))
+                ('message_set', Bytes)))))
     )
 
 
@@ -61,7 +60,7 @@ class FetchResponse_v4(Response):
                 ('aborted_transactions', Array(
                     ('producer_id', Int64),
                     ('first_offset', Int64))),
-                ('message_set', MessageSet)))))
+                ('message_set', Bytes)))))
     )
 
 
@@ -81,7 +80,7 @@ class FetchResponse_v5(Response):
                 ('aborted_transactions', Array(
                     ('producer_id', Int64),
                     ('first_offset', Int64))),
-                ('message_set', MessageSet)))))
+                ('message_set', Bytes)))))
     )