calling fetch at the last known offset raises protocol error about pykafka HOT 4 CLOSED

also possible to reproduce in ClientIntegrationTestCase.test_fetch

list(self.kafka.fetch(topic, 0, 1, size))

from pykafka.

added more specific issue in #16 but i'll leave this open until that is resolved in case there is more going here

from pykafka.

ignore previous comment, #16 wasn't the issue

from pykafka.

figured out what's happening here: the issue is that the message size is equal to the header + payload, so the offset is not easily derived from the response from client.fetch, since it only responds with the message itself and the starting offset. it's further complicated by the fact that the header can either be 9 or 10 bytes based on format, and the deflated payload size will not equivalent to the compressed payload size for compressed messages. the sizes that the client is returning are correct, but not the most useful.

i'm starting to think it might be more helpful to return a message object (namedtuple?) with start offset, end offset and payload, as well as maybe some of the additional header metadata (checksum, etc). it might not be a bad idea to attach partition/topic information as well if we are going to be chaining streams from multiple partitions together in some cases also.

from pykafka.