python之无法使用 `kafka-python` 轮询二进制消息

我有一个正在接收二进制数据(原始数据包捕获数据)的 Kafka 主题。我可以确认它确实是使用 Kafka CLI 工具登陆数据。我每秒收到多条消息。

kafka-console-consumer.sh --zookeeper svr:2181 --topic test 

但是当我使用 kafka-python 时，我无法检索任何消息。 poll 方法不返回任何结果。

(Pdb) consumer = kafka.KafkaConsumer("test", bootstrap_servers=["svr:9092"]) 
(Pdb) consumer.poll(5000) 
{} 

我已经能够使用 kafka-python 从仅包含文本字符串的单独主题中提取消息。

我很好奇 kafka-python 是否以某种方式在内部丢弃消息，因为它们是二进制的并且未通过某种验证。我如何才能更深入地了解为什么无法检索到消息？

请您参考如下方法:

问题是发送到主题的数据使用的是 snappy 压缩。我所要做的就是安装一个额外的模块来处理 snappy。

pip install python-snappy 

不幸的是，使用我在问题中概述的代码，它只是不返回任何数据，而不是告诉我问题与压缩有关。

为了进行比较，我使用了旧的消费者 API，它确实正确地报告了问题并引导我找到了这个解决方案。

>>> client = kafka.SimpleClient("svr:9092") 
>>> consumer.close() 
>>> consumer = kafka.SimpleConsumer(client, "group", "test") 
>>> for message in consumer: 
...     print(message) 
... 
Traceback (most recent call last): 
  File "<stdin>", line 1, in <module> 
  File "/usr/lib/python2.7/site-packages/kafka/consumer/simple.py", line 353, in __iter__ 
    message = self.get_message(True, timeout) 
  File "/usr/lib/python2.7/site-packages/kafka/consumer/simple.py", line 305, in get_message 
    return self._get_message(block, timeout, get_partition_info) 
  File "/usr/lib/python2.7/site-packages/kafka/consumer/simple.py", line 320, in _get_message 
    self._fetch() 
  File "/usr/lib/python2.7/site-packages/kafka/consumer/simple.py", line 379, in _fetch 
    fail_on_error=False 
  File "/usr/lib/python2.7/site-packages/kafka/client.py", line 665, in send_fetch_request 
    KafkaProtocol.decode_fetch_response) 
  File "/usr/lib/python2.7/site-packages/kafka/client.py", line 295, in _send_broker_aware_request 
    for payload_response in decoder_fn(future.value): 
  File "/usr/lib/python2.7/site-packages/kafka/protocol/legacy.py", line 212, in decode_fetch_response 
    for partition, error, highwater_offset, messages in partitions 
  File "/usr/lib/python2.7/site-packages/kafka/protocol/legacy.py", line 219, in decode_message_set 
    inner_messages = message.decompress() 
  File "/usr/lib/python2.7/site-packages/kafka/protocol/message.py", line 121, in decompress 
    assert has_snappy(), 'Snappy decompression unsupported' 
AssertionError: Snappy decompression unsupported