如果有单个消息大于 fetch size，则 kafka fetch 请求无法获取任何消息答案

【问题标题】：kafka fetch request cannot fetch any messages if there is a single message that is bigger than fetch size如果有单个消息大于 fetch size，则 kafka fetch 请求无法获取任何消息
【发布时间】：2015-06-18 06:47:25
【问题描述】：

在 kafka 手册网站中，有这个从 kafka 获取消息的代码块。但如果有一条消息大于 fetch size，则它不起作用。

但是，我们事先不知道进入队列的最大大小是多少。有没有办法让它总是获取至少一条消息？

import kafka.api.FetchRequest;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.javaapi.message.ByteBufferMessageSet;
import kafka.message.Message;
import kafka.message.MessageSet;
import kafka.utils.Utils;

...

// create a consumer to connect to the kafka server running on localhost, port 9092, socket timeout of 10 secs, socket receive buffer of ~1MB
SimpleConsumer consumer = new SimpleConsumer("127.0.0.1", 9092, 10000, 1024000);

long offset = 0;
while (true) {
  // create a fetch request for topic “test”, partition 0, current offset, and fetch size of 1MB
  FetchRequest fetchRequest = new FetchRequest("test", 0, offset, 1000000);

  // get the message set from the consumer and print them out
  ByteBufferMessageSet messages = consumer.fetch(fetchRequest);
  for(MessageAndOffset msg : messages) {
    System.out.println("consumed: " + Utils.toString(msg.message.payload(), "UTF-8"));
    // advance the offset after consuming each message
    offset = msg.offset;
  }
}

【问题讨论】：

标签： apache-kafka

【解决方案1】：

这里有两种方法：将您的服务器配置为拒绝大于消费者可以获取的最大消息大小的消息。确保您的代理的 max.message.bytes 设置和您的消费者的 fetch.message.max.bytes 在生产/消费主题之前相同（请参阅https://kafka.apache.org/08/configuration.html）。这会将问题推给生产者，以弄清楚如何保持消息大小不至于过大。

另一种方法是增加消费者的 fetch-max-bytes 并重试。因此，从 consumer.fetch(fetchRequest) 捕获 MessageSizeTooLarge 错误并使用更大的最大消息大小重试此错误（示例代码中的最后一个参数为 1000000）： FetchRequest fetchRequest = new FetchRequest("test", 0, offset, 1000000) 但这很危险——这是最大消息大小配置的重点。

我建议将消息大小限制推送给您的生产者。

【讨论】：