KafkaConsumer fails with ValueError in 1.2.0

[zoltan-fedor]

Just upgraded to 1.2.0 and the KafkaConsumer stopped working immediately, throwing a ValueError.

Even when using the most simplistic KafkaConsumer the error is thrown at the first record consumed:

kafka_consumer = KafkaConsumer(
    topic,
    group_id=kafka_group_id,
    bootstrap_servers=kafka_servers
)

for message in kafka_consumer:
    print(message)

Error:

e/import/opsc.py", line 7, in <module>\n    for message in kafka_consumer:
venv1/lib64/python3.4/site-packages/kafka/consumer/group.py", line 850, in __next__\n    return next(self._iterator)
venv1/lib64/python3.4/site-packages/kafka/consumer/group.py", line 790, in _message_generator\n    self._client.poll(timeout_ms=poll_ms, sleep=True)
venv1/lib64/python3.4/site-packages/kafka/client_async.py", line 436, in poll\n    responses.extend(self._poll(timeout, sleep=sleep))
venv1/lib64/python3.4/site-packages/kafka/client_async.py", line 479, in _poll\n    response = conn.recv() # Note: conn.recv runs callbacks / errbacks
venv1/lib64/python3.4/site-packages/kafka/conn.py", line 477, in recv\n    response = self._process_response(self._rbuffer) venv1/lib64/python3.4/site-packages/kafka/conn.py", line 510, in _process_response\n    response = ifr.response_type.decode(read_buffer)
venv1/lib64/python3.4/site-packages/kafka/protocol/struct.py", line 39, in decode\n    return cls(*[field.decode(data) for field in cls.SCHEMA.fields])
venv1/lib64/python3.4/site-packages/kafka/protocol/struct.py", line 39, in <listcomp>\n    return cls(*[field.decode(data) for field in cls.SCHEMA.fields])
venv1/lib64/python3.4/site-packages/kafka/protocol/types.py", line 155, in decode\n    return [self.array_of.decode(data) for _ in range(length)]
venv1/lib64/python3.4/site-packages/kafka/protocol/types.py", line 155, in <listcomp>\n    return [self.array_of.decode(data) for _ in range(length)] '  
venv1/lib64/python3.4/site-packages/kafka/protocol/types.py", line 118, in decode\n    return tuple([field.decode(data) for field in self.fields])
venv1/lib64/python3.4/site-packages/kafka/protocol/types.py", line 118, in <listcomp>\n    return tuple([field.decode(data) for field in self.fields])
venv1/lib64/python3.4/site-packages/kafka/protocol/types.py", line 155, in decode\n    return [self.array_of.decode(data) for _ in range(length)]
venv1/lib64/python3.4/site-packages/kafka/protocol/types.py", line 155, in <listcomp>\n    return [self.array_of.decode(data) for _ in range(length)]
venv1/lib64/python3.4/site-packages/kafka/protocol/types.py", line 118, in decode\n    return tuple([field.decode(data) for field in self.fields])
venv1/lib64/python3.4/site-packages/kafka/protocol/types.py", line 118, in <listcomp>\n    return tuple([field.decode(data) for field in self.fields])
venv1/lib64/python3.4/site-packages/kafka/protocol/types.py", line 50, in decode\n    return _unpack(\'>i\', data.read(4))
venv1/lib64/python3.4/site-packages/kafka/protocol/types.py", line 20, in _unpack\n    raise ValueError(error)

It has worked in 1.1.1

[dpkp]

What data is in your topic? What broker version are you using? Can you write a test case that produces a record that crashes the consumer?

[zoltan-fedor]

Downgraded back to 1.1.1 and so it works again, so it is definitely a 1.2.0 issue.

I have snappy compressed data in the topics.
Broker version 0.9.0.1

Example record (consumed by 1.1.1 successfully using the above code but failed with 1.2.0 throwing the above quote error):
ConsumerRecord(topic='opsc', partition=5, offset=800890490, key=None, value=b'\x00\x00\x00\x00y\x02@EF0640C1E38956764BD10A78C2FADA8C\x02H4c4c4544-0052-4b10-805a-c4c04f303832\x02H652836b2-eb57-4d01-ab5b-46457556dbe1\x02\x06ekn\x0222016-05-31T18:25:07+00:00\x02\x12EKNUSERID\x02@955DB79F57AA38489939ACBC70B76ED0\x02\x1aE User ID\x02\nE\x020ops-20160531_18_30_00.gz'

[TimEvens]

I'm also having the same issue with 1.2.0. I can also confirm that it works fine in 1.1.1.

Running broker 0.9.0.1 as well.

Traceback (most recent call last):
  File "/usr/lib/python2.7/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/local/lib/python2.7/dist-packages/openbmp/bmp_consumer.py", line 103, in run
    for m in consumer:
  File "/usr/local/lib/python2.7/dist-packages/six.py", line 558, in next
    return type(self).__next__(self)
  File "/usr/local/lib/python2.7/dist-packages/kafka/consumer/group.py", line 850, in __next__
    return next(self._iterator)
  File "/usr/local/lib/python2.7/dist-packages/kafka/consumer/group.py", line 790, in _message_generator
    self._client.poll(timeout_ms=poll_ms, sleep=True)
  File "/usr/local/lib/python2.7/dist-packages/kafka/client_async.py", line 436, in poll
    responses.extend(self._poll(timeout, sleep=sleep))
  File "/usr/local/lib/python2.7/dist-packages/kafka/client_async.py", line 479, in _poll
    response = conn.recv() # Note: conn.recv runs callbacks / errbacks
  File "/usr/local/lib/python2.7/dist-packages/kafka/conn.py", line 477, in recv
    response = self._process_response(self._rbuffer)
  File "/usr/local/lib/python2.7/dist-packages/kafka/conn.py", line 510, in _process_response
    response = ifr.response_type.decode(read_buffer)
  File "/usr/local/lib/python2.7/dist-packages/kafka/protocol/struct.py", line 39, in decode
    return cls(*[field.decode(data) for field in cls.SCHEMA.fields])
  File "/usr/local/lib/python2.7/dist-packages/kafka/protocol/types.py", line 155, in decode
    return [self.array_of.decode(data) for _ in range(length)]
  File "/usr/local/lib/python2.7/dist-packages/kafka/protocol/types.py", line 118, in decode
    return tuple([field.decode(data) for field in self.fields])
  File "/usr/local/lib/python2.7/dist-packages/kafka/protocol/types.py", line 74, in decode
    length = Int16.decode(data)
  File "/usr/local/lib/python2.7/dist-packages/kafka/protocol/types.py", line 40, in decode
    return _unpack('>h', data.read(2))
  File "/usr/local/lib/python2.7/dist-packages/kafka/protocol/types.py", line 20, in _unpack
    raise ValueError(error)
ValueError: <class 'struct.error'>

[dpkp]

There's definitely something wrong here. But I'm not able to reproduce, and the test-suite is not catching it (it does full end-to-end producer -> consumer testing on all compression types).

Can you post a snippet of debug logs? Any ideas on how to reproduce?

[dpkp]

If you are able, can you try PR #715 and post the response buffer that is causing the decode error?

[zoltan-fedor]

I have tried to PR #75 to capture the response buffer, it is a bit large, 14Mb text file, 4.5Mb zipped
Could I email it to you private?

[dpkp]

sure -- dana.powers gmail

[zoltan-fedor]

Thanks. Sent.

[dpkp]

OK -- I am able to reproduce the ValueError decoding the data you sent me. I'll investigate the issue further and post back. Thanks!!

[zoltan-fedor]

Excellent, thanks!

[dpkp]

I believe I found the bug. Thanks for the test data, it was extremely valuable. I believe PR #716 fixes. I've also added tests to avoid another regression.

[zoltan-fedor]

Thanks!
I have tested 1.2.1 and it seems it works well, can't reproduce the issue anymore, so it seems the fix is working!
Closing.