带有 max.request.size 参数的 kafka 流状态存储问题答案

【问题标题】：kafka stream state store issue with max.request.size paramater带有 max.request.size 参数的 kafka 流状态存储问题
【发布时间】：2020-09-23 02:59:30
【问题描述】：

我们在项目中使用 Kafka 流状态存储，我们想要存储超过 1MB 的数据，但出现以下异常：

序列化时消息为 1760923 字节，大于您使用 max.request.size 配置的最大请求大小配置。

然后我点击链接Add prefix to StreamsConfig to enable setting default internal topic configs 并添加了以下配置：

topic.max.request.size=50000000

然后应用程序工作正常，当状态存储内部主题已创建但重新启动Kafka并且状态存储主题丢失/删除时，它可以正常工作，然后Kafka流处理器需要创建内部状态存储主题启动应用程序时会自动抛出异常，提示：

"Aorg.apache.kafka.streams.errors.StreamsException: Could not create topic data-msg-seq-state-store-changelog. at org.apache.kafka.streams.processor.internals.InternalTopicManager.makeReady(InternalTopicManager.java:148)....
.....
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:805) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:774) Caused by: org.apache.kafka.common.errors.InvalidConfigurationException: Unknown topic config name: max.request.size".

解决方案是我们可以手动创建内部主题，但这应该不是一个好方法。

你能帮我解决这个问题吗？如果我错过了任何配置？

非常感谢。

2020 年 6 月 17 日更新：仍未解决问题。有人可以帮忙吗？

【问题讨论】：

max.request.size 是生产者配置，而不是主题配置：kafka.apache.org/documentation/#max.request.size
是的。这是否意味着我需要为代理上的生产者配置这个？或者我误解了一些东西。因为这是内部话题。
意思是，需要在producer上配置。 -- 在内部，Kafka Streams 使用 Kafka cosumer/producer/admin 客户端，您可以像直接使用这些客户端一样配置它们：docs.confluent.io/current/streams/developer-guide/…
其实我已经为生产者配置了max.request.size，如果我没有为内部主题（changelog）配置它，那么它会抛出异常。

标签： apache-kafka apache-kafka-streams

【解决方案1】：

您正在寻找的解决方案在于您在启动流之前设置的 Kafka Stream 的配置属性。

props.put(StreamsConfig.PRODUCER_PREFIX + ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "5242880");

我在这里使用的值是 5 MB 字节。您可以更改该值以满足您的需要。

【讨论】：

【解决方案2】：

我没有看到带有max.request.size 的配置。可能是 max.message.bytes (Topic configuration reference)。所以，你可以尝试设置这个。

您可以参考经纪人设置max.message.bytes并增加它。它将它设置在代理级别。

文档说明：

Kafka 允许的最大记录批大小（压缩后如果压缩已启用）。如果增加并且有消费者早于 0.10.2，消费者的获取大小也必须增加，所以他们可以获取这么大的记录批次。在最新消息格式版本，记录总是分组为效率。在以前的消息格式版本中，未压缩的记录不分组，此限制仅适用于单个在这种情况下记录。可以使用主题级别为每个主题设置 max.message.bytes 配置。

默认值：1048588 (~1Mb) (Confluent Kafka)

也可参考以下Stackoverflow answer

【讨论】：

感谢您的回复。我也没有在代码中看到这个配置。但我有一个问题是当我不添加此参数时，为什么它会抛出异常说：“序列化时消息为 1760923 字节，大于您使用 max.request.size配置的最大请求大小> 配置。”。其中提到了“max.request.size”。
感谢您的回复。
你使用的是哪个版本的 Kafka？
版本为：confluentinc/cp-kafka:5.2.1
max.request.size 是生产者配置，而不是主题配置：kafka.apache.org/documentation/#max.request.size