Repository: Parsely/pykafka
Pull request #693 (2017-06-12 23:20:29)

message timestamp tests
This pull request adds tests for message timestamps in the protocol.


Patch:
diff --git a/pykafka/broker.py b/pykafka/broker.py
index dda2f0c..012b52d 100644
--- a/pykafka/broker.py
+++ b/pykafka/broker.py
@@ -59,7 +59,8 @@ class Broker(object):
                  buffer_size=1024 * 1024,
                  source_host='',
                  source_port=0,
-                 ssl_config=None):
+                 ssl_config=None,
+                 broker_version="0.9.0"):
         """Create a Broker instance.
 
         :param id_: The id number of this broker
@@ -104,6 +105,7 @@ class Broker(object):
         self._offsets_channel_socket_timeout_ms = offsets_channel_socket_timeout_ms
         self._buffer_size = buffer_size
         self._req_handlers = {}
+        self._broker_version = broker_version
         try:
             self.connect()
         except SocketDisconnectedError:
@@ -129,7 +131,8 @@ class Broker(object):
                       buffer_size=64 * 1024,
                       source_host='',
                       source_port=0,
-                      ssl_config=None):
+                      ssl_config=None,
+                      broker_version="0.9.0"):
         """Create a Broker using BrokerMetadata
 
         :param metadata: Metadata that describes the broker.
@@ -160,7 +163,8 @@ class Broker(object):
                    buffer_size=buffer_size,
                    source_host=source_host,
                    source_port=source_port,
-                   ssl_config=ssl_config)
+                   ssl_config=ssl_config,
+                   broker_version=broker_version)
 
     @property
     def connected(self):
@@ -289,13 +293,15 @@ class Broker(object):
             block for up to `timeout` milliseconds.
         :type min_bytes: int
         """
+        response_class = FetchResponse.get_subclass(self._broker_version)
         future = self._req_handler.request(FetchRequest(
             partition_requests=partition_requests,
             timeout=timeout,
             min_bytes=min_bytes,
+            api_version=response_class.api_version,
         ))
         # XXX - this call returns even with less than min_bytes of messages?
-        return future.get(FetchResponse)
+        return future.get(response_class)
 
     @_check_handler
     def produce_messages(self, produce_request):
diff --git a/pykafka/cluster.py b/pykafka/cluster.py
index 6c0aa7f..de3a9ba 100644
--- a/pykafka/cluster.py
+++ b/pykafka/cluster.py
@@ -246,7 +246,8 @@ class Cluster(object):
                                     buffer_size=1024 * 1024,
                                     source_host=self._source_host,
                                     source_port=self._source_port,
-                                    ssl_config=self._ssl_config)
+                                    ssl_config=self._ssl_config,
+                                    broker_version=self._broker_version)
                     response = broker.request_metadata(topics)
                     if response is not None:
                         return response
@@ -355,7 +356,8 @@ class Cluster(object):
                     buffer_size=1024 * 1024,
                     source_host=self._source_host,
                     source_port=self._source_port,
-                    ssl_config=self._ssl_config)
+                    ssl_config=self._ssl_config,
+                    broker_version=self._broker_version)
             elif not self._brokers[id_].connected:
                 log.info('Reconnecting to broker id %s: %s:%s', id_, meta.host, meta.port)
                 try:
diff --git a/pykafka/producer.py b/pykafka/producer.py
index 781a89e..7c73b24 100644
--- a/pykafka/producer.py
+++ b/pykafka/producer.py
@@ -320,6 +320,8 @@ class Producer(object):
         if not (isinstance(message, bytes) or message is None):
             raise TypeError("Producer.produce accepts a bytes object as message, but it "
                             "got '%s'", type(message))
+        if self._protocol_version < 1 and timestamp:
+            raise RuntimeError("Producer.produce got a timestamp with protocol 0")
         if not self._running:
             raise ProducerStoppedException()
         partitions = list(self._topic.partitions.values())
diff --git a/pykafka/protocol.py b/pykafka/protocol.py
index e0ad8ae..61d3eb9 100644
--- a/pykafka/protocol.py
+++ b/pykafka/protocol.py
@@ -62,6 +62,8 @@ from zlib import crc32
 from datetime import datetime
 import types
 from six import integer_types
+from pkg_resources import parse_version
+
 
 from .common import CompressionType, Message
 from .exceptions import ERROR_CODES, MessageSizeTooLarge
@@ -212,13 +214,13 @@ class Message(Message, Serializable):
 
     @classmethod
     def decode(self, buff, msg_offset=-1, partition_id=-1):
-        (crc, protocol_version) = struct_helpers.unpack_from('iB', buff, 0)
-        offset = 5
+        (crc, protocol_version, attr) = struct_helpers.unpack_from('iBB', buff, 0)
+        offset = 6
         timestamp = 0
         if protocol_version > 0:
             (timestamp,) = struct_helpers.unpack_from('Q', buff, offset)
             offset += 8
-        (attr, key, val) = struct_helpers.unpack_from('BYY', buff, offset)
+        (key, val) = struct_helpers.unpack_from('YY', buff, offset)
         # TODO: Handle CRC failure
         return Message(val,
                        partition_key=key,
@@ -677,7 +679,8 @@ class FetchRequest(Request):
           FetchOffset => int64
           MaxBytes => int32
     """
-    def __init__(self, partition_requests=[], timeout=1000, min_bytes=1024):
+    def __init__(self, partition_requests=[], timeout=1000, min_bytes=1024,
+                 api_version=0):
         """Create a new fetch request
 
         Kafka 0.8 uses long polling for fetch requests, which is different
@@ -695,6 +698,7 @@ class FetchRequest(Request):
         self.timeout = timeout
         self.min_bytes = min_bytes
         self._reqs = defaultdict(dict)
+        self.api_version = api_version
         for req in partition_requests:
             self.add_request(req)
 
@@ -732,7 +736,7 @@ class FetchRequest(Request):
         :rtype: :class:`bytearray`
         """
         output = bytearray(len(self))
-        self._write_header(output)
+        self._write_header(output, api_version=self.api_version)
         offset = self.HEADER_LEN
         struct.pack_into('!iiii', output, offset,
                          -1, self.timeout, self.min_bytes, len(self._reqs))
@@ -769,14 +773,30 @@ class FetchResponse(Response):
           HighwaterMarkOffset => int64
           MessageSetSize => int32
     """
-    def __init__(self, buff):
+    api_version = 0
+
+    @staticmethod
+    def get_subclass(broker_protocol):
+        """Choose which subclass of response to demand and expect. Cf.
+        https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol"""
+        target_version = parse_version(broker_protocol)
+        if target_version >= parse_version("0.10.0"):
+            return FetchResponseV2
+        elif target_version >= parse_version("0.9.0"):
+            return FetchResponseV1
+        else:
+            return FetchResponse
+
+    def __init__(self, buff, offset=0):
         """Deserialize into a new Response
 
         :param buff: Serialized message
         :type buff: :class:`bytearray`
+        :param offset: Offset into the message
+        :type offset: int
         """
         fmt = '[S [ihqY] ]'
-        response = struct_helpers.unpack_from(fmt, buff, 0)
+        response = struct_helpers.unpack_from(fmt, buff, offset)
         self.topics = defaultdict(dict)
         for (topic, partitions) in response:
             for partition in partitions:
@@ -794,17 +814,45 @@ class FetchResponse(Response):
         for message in message_set.messages:
             if message.compression_type == CompressionType.NONE:
                 output.append(message)
+                continue
             elif message.compression_type == CompressionType.GZIP:
                 decompressed = compression.decode_gzip(message.value)
-                output += self._unpack_message_set(decompressed,
-                                                   partition_id=partition_id)
+                messages = self._unpack_message_set(decompressed,
+                                                    partition_id=partition_id)
             elif message.compression_type == CompressionType.SNAPPY:
                 decompressed = compression.decode_snappy(message.value)
-                output += self._unpack_message_set(decompressed,
-                                                   partition_id=partition_id)
+                messages = self._unpack_message_set(decompressed,
+                                                    partition_id=partition_id)
+            if messages[-1].offset < message.offset:
+                # With protocol 1, offsets from compressed messages start at 0
+                assert messages[0].offset == 0
+                delta = message.offset - len(messages) + 1
+                for msg in messages:
+                    msg.offset += delta
+            output += messages
         return output
 
 
+class FetchResponseV1(FetchResponse):
+    api_version = 1
+
+    def __init__(self, buff, offset=0):
+        """Deserialize into a new Response
+
+        :param buff: Serialized message
+        :type buff: :class:`bytearray`
+        :param offset: Offset into the message
+        :type offset: int
+        """
+        # TODO: Use throttle_time
+        self.throttle_time = struct_helpers.unpack_from("i", buff, offset)
+        super(FetchResponseV1, self).__init__(buff, offset + 4)
+
+
+class FetchResponseV2(FetchResponseV1):
+    api_version = 2
+
+
 ##
 # Offset API
 ##
diff --git a/pykafka/rdkafka/_rd_kafkamodule.c b/pykafka/rdkafka/_rd_kafkamodule.c
index 97255a4..69c4bd0 100644
--- a/pykafka/rdkafka/_rd_kafkamodule.c
+++ b/pykafka/rdkafka/_rd_kafkamodule.c
@@ -957,6 +957,7 @@ Consumer_consume(RdkHandle *self, PyObject *args)
     PyObject *empty_args = NULL;
     PyObject *kwargs = NULL;
     rd_kafka_message_t *rkmessage;
+    int protocol_version = 0;
 
     if (RdkHandle_safe_lock(self, /* check_running= */ 1)) return NULL;
     Py_BEGIN_ALLOW_THREADS  /* avoid callbacks deadlocking */
@@ -978,11 +979,15 @@ Consumer_consume(RdkHandle *self, PyObject *args)
         int64_t timestamp = rd_kafka_message_timestamp(rkmessage, &timestamp_type);
         if (timestamp_type == RD_KAFKA_TIMESTAMP_NOT_AVAILABLE) {
             timestamp = 0;
+        } else {
+            // infer protocol from presence of timestamp. Fragile, but protocol
+            // not transmitted by librdkafka
+            protocol_version = 1;
         }
 #if PY_MAJOR_VERSION >= 3
-        const char *format = "{s:y#,s:y#,s:l,s:L,s:L}";
+        const char *format = "{s:y#,s:y#,s:l,s:L,s:i,s:L}";
 #else
-        const char *format = "{s:s#,s:s#,s:l,s:L,s:L}";
+        const char *format = "{s:s#,s:s#,s:l,s:L,s:i,s:L}";
 #endif
         kwargs = Py_BuildValue(
             format,
@@ -990,6 +995,7 @@ Consumer_consume(RdkHandle *self, PyObject *args)
             "partition_key", rkmessage->key, rkmessage->key_len,
             "partition_id", (long)rkmessage->partition,
             "offset", (PY_LONG_LONG)rkmessage->offset,
+            "protocol_version", protocol_version,
             "timestamp", (PY_LONG_LONG)timestamp);
         if (! kwargs) goto cleanup;
         empty_args = PyTuple_New(0);


Test patch:
diff --git a/tests/pykafka/test_protocol.py b/tests/pykafka/test_protocol.py
index 8ac3e19..8f1ef44 100644
--- a/tests/pykafka/test_protocol.py
+++ b/tests/pykafka/test_protocol.py
@@ -136,6 +136,8 @@ class TestProduceAPI(unittest2.TestCase):
 
     test_messages = [
         protocol.Message(b'this is a test message', partition_key=b'asdf'),
+        protocol.Message(b'this is a test message', partition_key=b'asdf',
+                         timestamp=1497302164, protocol_version=1),
         protocol.Message(b'this is also a test message', partition_key=b'test_key'),
         protocol.Message(b"this doesn't have a partition key"),
     ]
@@ -169,17 +171,47 @@ class TestProduceAPI(unittest2.TestCase):
             )
         )
 
+    def test_request_message_timestamp(self):
+        message = self.test_messages[1]
+        req = protocol.ProduceRequest()
+        req.add_message(message, b'test', 0)
+        msg = req.get_bytes()
+        self.assertEqual(
+            msg,
+            bytearray(
+                b'\x00\x00\x00i\x00\x00\x00\x00\x00\x00\x00\x00\x00\x07pykafka'  # header
+                b'\x00\x01'  # required acks
+                b"\x00\x00\'\x10"  # timeout
+                b"\x00\x00\x00\x01"  # len(topics)
+                    b"\x00\x04"  # len(topic name)
+                        b"test"  # topic name
+                    b"\x00\x00\x00\x01"  # len(partitions)
+                        b"\x00\x00\x00\x00"  # partition
+                        b"\x00\x00\x00<"  # message set size
+                            b"\xff\xff\xff\xff\xff\xff\xff\xff"  # offset
+                            b"\x00\x00\x000"  # message size
+                                b"Z\x92\x80\t"  # crc
+                                b"\x01"  # magic byte (protocol version)
+                                b"\x00"  # attributes
+                                b"\x00\x00\x00\x00Y?\x04\x94"  # timestamp
+                                b"\x00\x00\x00\x04"  # len(key)
+                                    b"asdf"  # key
+                                b"\x00\x00\x00\x16"  # len(value)
+                                    b"this is a test message"  # value
+            )
+        )
+
     def test_gzip_compression(self):
         req = protocol.ProduceRequest(compression_type=CompressionType.GZIP)
         [req.add_message(m, b'test_gzip', 0) for m in self.test_messages]
         msg = req.get_bytes()
-        self.assertEqual(len(msg), 207)  # this isn't a good test
+        self.assertEqual(len(msg), 230)  # this isn't a good test
 
     def test_snappy_compression(self):
         req = protocol.ProduceRequest(compression_type=CompressionType.SNAPPY)
         [req.add_message(m, b'test_snappy', 0) for m in self.test_messages]
         msg = req.get_bytes()
-        self.assertEqual(len(msg), 212)  # this isn't a good test
+        self.assertEqual(len(msg), 240)  # this isn't a good test
 
     def test_partition_error(self):
         # Response has a UnknownTopicOrPartition error for test/0
@@ -226,8 +258,8 @@ class TestFetchAPI(unittest2.TestCase):
             'partition_id': 0,
             'produce_attempt': 0,
             'delivery_report_q': None,
-            'timestamp': 0,
-            'protocol_version': 0,
+            'timestamp': 1497393304998,
+            'protocol_version': 1,
             'partition': None
         }, {
             'partition_key': b'test_key',
@@ -237,8 +269,8 @@ class TestFetchAPI(unittest2.TestCase):
             'partition_id': 0,
             'produce_attempt': 0,
             'delivery_report_q': None,
-            'timestamp': 0,
-            'protocol_version': 0,
+            'timestamp': 1497393305005,
+            'protocol_version': 1,
             'partition': None
         }, {
             'partition_key': None,
@@ -248,8 +280,30 @@ class TestFetchAPI(unittest2.TestCase):
             'partition_id': 0,
             'produce_attempt': 0,
             'delivery_report_q': None,
-            'timestamp': 0,
-            'protocol_version': 0,
+            'timestamp': 1497393305013,
+            'protocol_version': 1,
+            'partition': None
+        }, {
+            'partition_key': b"test_key",
+            'compression_type': 0,
+            'value': b"this has a partition key and a timestamp",
+            'offset': 3,
+            'partition_id': 0,
+            'produce_attempt': 0,
+            'delivery_report_q': None,
+            'timestamp': 1497302164,
+            'protocol_version': 1,
+            'partition': None
+        }, {
+            'partition_key': None,
+            'compression_type': 0,
+            'value': b"this has a timestamp",
+            'offset': 4,
+            'partition_id': 0,
+            'produce_attempt': 0,
+            'delivery_report_q': None,
+            'timestamp': 1497302164,
+            'protocol_version': 1,
             'partition': None
         }]
 
@@ -337,53 +391,27 @@ class TestFetchAPI(unittest2.TestCase):
 
     def test_gzip_decompression(self):
         msg = b''.join([
-            b'\x00\x00\x00\x01'  # len(topics)
-            b'\x00\t'  # len(topic name)
-                b'test_gzip'  # topic name
-            b'\x00\x00\x00\x01'  # len(partitions)
-                b'\x00\x00\x00\x00'  # partition
-                    b'\x00\x00'  # error code
-                    b'\x00\x00\x00\x00\x00\x00\x00\x03'  # highwater mark offset
-                    b'\x00\x00\x00\x9b'  # message set size
-                        b'\x00\x00\x00\x00\x00\x00\x00\x02'  # offset
-                        b'\x00\x00\x00\x8f'  # message size
-                            b'\xbb\xe7\x1f\xb8'  # crc
-                            b'\x00'  # magic byte
-                            b'\x01'  # attributes
-                            b'\xff\xff\xff\xff'  # len(key)
-                            b'\x00\x00\x00\x81'  # len(value)
-                                b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x00c`\x80\x03\r\xbe.I\x7f0\x8b%\xb18%\rH\x8b\x95dd\x16+\x00Q\xa2BIjq\x89Bnjqqbz*T=#\x10\x1b\xb2\xf3\xcb\xf4\x81y\x1c \x15\xf1\xd9\xa9\x95@\xb64\\_Nq>v\xcdL@\xac\x7f\xb5(\xd9\x98\x81\xe1?\x10\x00y\x8a`M)\xf9\xa9\xc5y\xea%\n\x19\x89e\xa9@\x9d\x05\x89E%\x99%\x99\xf9y\n@\x93\x01N1\x9f[\xac\x00\x00\x00'  # value
-        ])
-        response = protocol.FetchResponse(msg)
+b'\x00\x00\x00\x00\x00\x00\x00\x01\x00\x0btest_gzip_5\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x05\x00\x00\x02\x17\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00Z\xb1Z\xf4\xc3\x01\x01\x00\x00\x01\\\xa3\x98\x95\xa6\xff\xff\xff\xff\x00\x00\x00D\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x00c`\x80\x03\x03\x97?.{\x19\x81\x0c\xc6\x98\xc53\xa6.\x032X\x12\x8bS\xd2\x80\xb4XIFf\xb1\x02\x10%*\x94\xa4\x16\x97(\xe4\xa6\x16\x17\'\xa6\xa7\x02\x00N\xddm\x92<\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00`\xcdX\t\xed\x01\x01\x00\x00\x01\\\xa3\x98\x95\xad\xff\xff\xff\xff\x00\x00\x00J\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x00c`\x80\x03\xcb%\xf2\x01\x99\x8c@\x06c\xcc\xe2\x19S\xd7\x02\x19\x1c%\xa9\xc5%\xf1\xd9\xa9\x95@\xb6tIFf\xb1\x02\x10%\xe6\x14\xe7+$*\x80\xa4\x14rS\x8b\x8b\x13\xd3S\x01\xfe<~BE\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00b\xba\xb0lN\x01\x01\x00\x00\x01\\\xa3\x98\x95\xb5\xff\xff\xff\xff\x00\x00\x00L\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x00c`\x80\x03s\xfb\xad2\x07\x19\x81\x0c\xc6\x98\xc53\xa6n\xfd\x0f\x04@\x8ebIFf\xb1BJ~jq\x9ez\x89BFbY\xaaB\xa2BAbQIfIf~\x9eBvj%\x00\xc8f?\xe3C\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x03\x00\x00\x00l\x9f\xd7)\xa0\x01\x01\x00\x00\x00\x00Y?\x04\x94\xff\xff\xff\xff\x00\x00\x00V\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x00c`\x80\x037\x86wK\x8f0\x82\x99\x91\xf6,S\x80\x14GIjqI|vj%\x90\xadQ\x92\x91Y\xac\x90\x91X\xac\x90\xa8P\x90XT\x92Y\x92\x99\x9f\xa7\x00\x94SH\xccK\x01\x8a\x95d\xe6\x02\x15\'\xe6\x16\x00\x00\xac\xc0O.R\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\x00\x00\x00S\xd8"\xff7\x01\x01\x00\x00\x00\x00Y?\x04\x94\xff\xff\xff\xff\x00\x00\x00=\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x00c`\x80\x03\xad\x0f\xae\x97^2\x82\x99\x91\xf6,S\xfe\x03\x01\x90)R\x92\x91Y\xac\x90\x91X\xac\x90\xa8P\x92\x99\x9bZ\\\x92\x98[\x00\x00\x83\x0f\xe5\xc16\x00\x00\x00\x00\x00\x00\x00'
+            ])
+        response = protocol.FetchResponse.get_subclass("0.10.1.0")(msg)
         for i in range(len(self.expected_data)):
             self.assertDictEqual(
-                self.msg_to_dict(response.topics[b'test_gzip'][0].messages[i]),
+                self.msg_to_dict(response.topics[b'test_gzip_5'][0].messages[i]),
                 self.expected_data[i])
 
     def test_snappy_decompression(self):
         msg = b''.join([
-            b'\x00\x00\x00\x01'  # len(topics)
-            b'\x00\x0b'  # len(topic name)
-                b'test_snappy'  # topic name
-            b'\x00\x00\x00\x01'  # len(partitions)
-                b'\x00\x00\x00\x00'  # partition
-                    b'\x00\x00'  # error code
-                    b'\x00\x00\x00\x00\x00\x00\x00\x03'  # highwater mark offset
-                    b'\x00\x00\x00\xb5'  # message set size
-                        b'\x00\x00\x00\x00\x00\x00\x00\x02'  # offset
-                        b'\x00\x00\x00\xa9'  # message size
-                            b'\xc1\xf2\xa3\xe1'  # crc
-                            b'\x00'  # magic bytes
-                            b'\x02'  # attributes
-                            b'\xff\xff\xff\xff'  # len(key)
-                            b'\x00\x00\x00\x9b'  # len(value)
-                                b'\x82SNAPPY\x00\x00\x00\x00\x01\x00\x00\x00\x01\x00\x00\x00\x87\xac\x01\x00\x00\x19\x01\x10(\x0e\x8a\x19O\x05\x0fx\x04asdf\x00\x00\x00\x16this is a test message\x05$(\x00\x00\x01\x00\x00\x001\x07\x0f\x1c\x8e\x05\x10\x00\x08\x01"\x1c_key\x00\x00\x00\x1b\x158\x08lsoV=\x00H\x02\x00\x00\x00/\xd5rc3\x00\x00\xff\xff\xff\xff\x00\x00\x00!\x055ldoesn\'t have a partition key'  # value
+b"\x00\x00\x00\x00\x00\x00\x00\x01\x00\x0btest_snappy\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x05\x00\x00\x02=\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00a\xaa\x92\x19\xe2\x01\x02\x00\x00\x01\\\xa3\xa5\xab/\xff\xff\xff\xff\x00\x00\x00K\x82SNAPPY\x00\x00\x00\x00\x01\x00\x00\x00\x01\x00\x00\x007<\x00\x00\x19\x01\xc00\x86\xf55\x85\x01\x00\x00\x00\x01\\\xa3\xa5\xab/\x00\x00\x00\x04asdf\x00\x00\x00\x16this is a test message\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00jI\xfc\xe7\xf6\x01\x02\x00\x00\x01\\\xa3\xa5\xab\xa1\xff\xff\xff\xff\x00\x00\x00T\x82SNAPPY\x00\x00\x00\x00\x01\x00\x00\x00\x01\x00\x00\x00@E\x00\x00\x19\x01\xe49\xd4\r\xf8v\x01\x00\x00\x00\x01\\\xa3\xa5\xab\xa1\x00\x00\x00\x08test_key\x00\x00\x00\x1bthis is also a test message\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00h\x80\t0\x95\x01\x02\x00\x00\x01\\\xa3\xa5\xab\xe1\xff\xff\xff\xff\x00\x00\x00R\x82SNAPPY\x00\x00\x00\x00\x01\x00\x00\x00\x01\x00\x00\x00>C\x00\x00\x19\x01\xdc72\x98C\xe8\x01\x00\x00\x00\x01\\\xa3\xa5\xab\xe1\xff\xff\xff\xff\x00\x00\x00!this doesn't have a partition key\x00\x00\x00\x00\x00\x00\x00\x03\x00\x00\x00uJ\xab\xd0\xdf\x01\x02\x00\x00\x00\x00Y?\x04\x94\xff\xff\xff\xff\x00\x00\x00_\x82SNAPPY\x00\x00\x00\x00\x01\x00\x00\x00\x01\x00\x00\x00KR\x00\x00\x19\x01\x14F\x00\xee\xa5\xc4\x01\x05\x10\xecY?\x04\x94\x00\x00\x00\x08test_key\x00\x00\x00(this has a partition key and a timestamp\x00\x00\x00\x00\x00\x00\x00\x04\x00\x00\x00Y\xfb!\x15\xce\x01\x02\x00\x00\x00\x00Y?\x04\x94\xff\xff\xff\xff\x00\x00\x00C\x82SNAPPY\x00\x00\x00\x00\x01\x00\x00\x00\x01\x00\x00\x00/6\x00\x00\x19\x01\x14*\xf0E\xd2\xe9\x01\x05\x10|Y?\x04\x94\xff\xff\xff\xff\x00\x00\x00\x14this has a timestamp\x00\x00\x00\x00"
         ])
-        response = protocol.FetchResponse(msg)
+        response = protocol.FetchResponse.get_subclass("0.10.1.0")(msg)
         for i in range(len(self.expected_data)):
-            self.assertDictEqual(
-                self.msg_to_dict(response.topics[b'test_snappy'][0].messages[i]),
-                self.expected_data[i])
+            returned = self.msg_to_dict(response.topics[b'test_snappy'][0].messages[i])
+            expected = self.expected_data[i]
+            # ignore timestamps if they were auto-set
+            if i <= 2:
+                returned.pop("timestamp")
+                expected.pop("timestamp")
+            self.assertDictEqual(returned, expected)
 
 
 class TestOffsetAPI(unittest2.TestCase):