Kafka 流聚合 RecordTooLargeException

Question

尝试从单个数据流聚合数据，滚动 windows 5 分钟。最初它工作正常并且能够打印汇总记录。使用 Java 8 进行开发。

但后来开始收到错误-

"org.apache.kafka.common.errors.RecordTooLargeException：消息序列化后为5292482字节，大于1048576，即max.request.size配置的值"

现在每次在 EKS 集群中启动我的应用程序后，都会在一分钟内崩溃，并出现相同的错误。

尝试设置以下流配置，但也无济于事： StreamsConfig.RECEIVE_BUFFER_CONFIG (50MB) StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG（50MB）

以下任何人都可以解决此问题：

1. 使用 inMemoryKeyValueStore。是否有任何特定的 属性 为 inMemoryKeyValueStore 分配一些内存？
2. 或者应该切换到 persistentKeyValueStore？
3. 使用 AWS MSK，因此在创建集群时定义代理和主题级别设置 - message.max.bytes 具有适当的值。

提前致谢。

Answer 1

您需要增加生产者配置 max.request.size（如错误消息中所述）以解决该问题。

请注意，您可能需要增加您另外提到的 broker/topic 配置 message.max.bytes。