Spurious ConsumerFetchSizeTooSmall error?

[clofresh]

I'm making a batch fetch request on 16 partitions and getting ConsumerFetchSizeTooSmall() errors. I tried raising the max buffer size to a something extremely large like 1gb, but it still gives the error.

2014-02-27 21:00:28,728 WARNING [kafka] Fetch size too small, increase to 104857600 (2x) and retry
2014-02-27 21:00:28,733 ERROR [kafka] Max fetch size 1073741824 too small
2014-02-27 21:00:28,733 ERROR [dd] [10931] [core.py:165] -·
Traceback (most recent call last):
...
  File "/usr/local/lib/python2.6/dist-packages/kafka_python-0.9.0_6db14-py2.6.egg/kafka/consumer.py", line 323, in get_messages
    message = self.get_message(block, timeout)
  File "/usr/local/lib/python2.6/dist-packages/kafka_python-0.9.0_6db14-py2.6.egg/kafka/consumer.py", line 343, in get_message
    self._fetch()
  File "/usr/local/lib/python2.6/dist-packages/kafka_python-0.9.0_6db14-py2.6.egg/kafka/consumer.py", line 404, in _fetch
    raise e
ConsumerFetchSizeTooSmall

Hacking the code to just reraise the original exception gives this stack trace:

   File "/usr/local/lib/python2.6/dist-packages/kafka_python-0.9.0_6db14-py2.6.egg/kafka/consumer.py", line 323, in get_messages
     message = self.get_message(block, timeout)
   File "/usr/local/lib/python2.6/dist-packages/kafka_python-0.9.0_6db14-py2.6.egg/kafka/consumer.py", line 343, in get_message
     self._fetch()
   File "/usr/local/lib/python2.6/dist-packages/kafka_python-0.9.0_6db14-py2.6.egg/kafka/consumer.py", line 386, in _fetch
     for message in resp.messages:
   File "/usr/local/lib/python2.6/dist-packages/kafka_python-0.9.0_6db14-py2.6.egg/kafka/protocol.py", line 117, in _decode_message_set_iter
     (msg, cur) = read_int_string(data, cur)
   File "/usr/local/lib/python2.6/dist-packages/kafka_python-0.9.0_6db14-py2.6.egg/kafka/util.py", line 50, in read_int_string
     raise BufferUnderflowError("Not enough data left")
 BufferUnderflowError: Not enough data left

I think it may be a bug in the client logic, since I can read individually from each partition.

Here is a sample response payload:

https://www.dropbox.com/s/cw1i1ct1p1aplp4/raw_fetch_response.txt.gz

[clofresh]

Hmm, maybe I do have more than 1gb of data in the fetch. This is my interpretation of the response payload:

\x00\x00\x00\x02 # correlation id

\x00\x00\x00\x01 # topic count

\x00\x0c # topic length

points.multi # topic

\x00\x00\x00\x10 # partition id

# partition data

\x00\x00 # error code

\x00\x00\x00\x00\x00\x00\x00\x00 # highwater mark

\x06\xb2p9 # message set size is 963686918, but there's only 62002 more bytes left

[clofresh]

This is a better test case:

https://www.dropbox.com/s/m69uy9amozzuwvb/raw_fetch_response2.txt.gz

>> f.read(4)
'\x00\x00\x00\x02'
>>> f.read(4)
'\x00\x00\x00\x01'
>>> f.read(2)
'\x00\x0c'
>>> f.read(12)
'points.multi'
>>> f.read(4)
'\x00\x00\x00\x10'
>>> f.read(2)
'\x00\x00'
>>> f.read(8)
'\x00\x00\x00\x00\x00\x00\x00\x00'
>>> f.read(4)
'\x06\xb2q\x01'
>>> import struct
>>> struct.unpack('i', '\x06\xb2q\x01')
(24228358,)
>>> blah = f.read()
>>> f.tell()
30842
>>>

So the message set size is 24228358, which is under my max fetch size of 1gb.

[clofresh]

Not sure if it's significant, but this is my broker socket settings:

socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=1073741824

[clofresh]

I wrote a script to parse the binary: https://gist.github.com/clofresh/9263197

It does look like some partitions are incomplete, but the code should still be able to parse the rest of the message sets from the partitions.

[hellais]

I am also seeing this error in a consumer after having increased the buffer size to 120 MB. Not sure how to proceed in debugging it.

I am using the current master of kafka-python.