java - Kafka 流连接中的 RecordTooLargeException

Question

我有一个 KStream x KStream 连接，它因以下异常而崩溃。

Exception in thread “my-clicks-and-recs-join-streams-4c903fb1-5938-4919-9c56-2c8043b86986-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_15, processor=KSTREAM-SOURCE-0000000001, topic=my_outgoing_recs_prod, partition=15, offset=9248896
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:203)
    at org.apache.kafka.streams.processor.internals.StreamThread.processAndPunctuate(StreamThread.java:679)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:557)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:527)
Caused by: org.apache.kafka.streams.errors.StreamsException: task [0_15] exception caught when producing
    at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.checkForException(RecordCollectorImpl.java:136)
    at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:87)
    at org.apache.kafka.streams.state.internals.StoreChangeLogger.logChange(StoreChangeLogger.java:59)
    at org.apache.kafka.streams.state.internals.ChangeLoggingSegmentedBytesStore.put(ChangeLoggingSegmentedBytesStore.java:59)
    at org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.put(MeteredSegmentedBytesStore.java:105)
    at org.apache.kafka.streams.state.internals.RocksDBWindowStore.put(RocksDBWindowStore.java:107)
    at org.apache.kafka.streams.state.internals.RocksDBWindowStore.put(RocksDBWindowStore.java:100)
    at org.apache.kafka.streams.kstream.internals.KStreamJoinWindow$KStreamJoinWindowProcessor.process(KStreamJoinWindow.java:64)
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:47)
    at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:187)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:82)
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:80)
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:189)
    ... 3 more
Caused by: org.apache.kafka.common.errors.RecordTooLargeException: The request included a message larger than the max message size the server will accept.

我正在加入一个带有Recommendation 主题的Click 主题。 Click 对象非常小(不到 1 KB)。另一方面，Recommendation 可能很大，有时会超过 1 MB。

我用 Google 搜索了异常，发现 ( here ) 我需要在生产者配置中设置 max.request.size。

我不明白的是，生产者在流连接中的什么位置出现？上述异常中的主题 topic=my_outgoing_recs_prod 是推荐主题，而不是最终加入的主题。流媒体应用程序不应该只是“消费”它吗？

不过，我尝试将属性设置为 config.put("max.request.size", "31457280");，即 30MB。我不希望推荐记录超过该限制。尽管如此，代码还是崩溃了。

我无法更改 Kafka 集群中的配置，但如果需要，我可以更改 Kafka 中相关主题的属性。

有人可以建议我还能尝试什么吗？

如果没有任何效果，我愿意忽略此​​类超大消息。但是，我不知道如何处理此 RecordTooLargeException。

我执行连接的代码如下。

Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, JOINER_ID + "-" + System.getenv("HOSTNAME"));
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, booststrapList);
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArray().getClass().getName());
config.put("max.request.size", "314572800");
config.put("message.max.bytes", "314572800");
config.put("max.message.bytes", "314572800");


KStreamBuilder builder = new KStreamBuilder();

KStream<String, byte[]> clicksStream = builder.stream(TopologyBuilder.AutoOffsetReset.LATEST, Serdes.String(), Serdes.ByteArray(), clicksTopic);
KStream<String, byte[]> recsStream = builder.stream(TopologyBuilder.AutoOffsetReset.LATEST, Serdes.String(), Serdes.ByteArray(), recsTopic);

KStream<String, ClickRec> join = clicksStream.join(
        recsStream,
        (click, recs) -> new ClickRec(click, recs),
        JoinWindows.of(windowMillis).until(3*windowMillis));

join.to(Serdes.String(), JoinSerdes.CLICK_SERDE, jointTopic);

KafkaStreams streams = new KafkaStreams(builder, config);
streams.cleanUp();
streams.start();

ClickRec 是连接对象(它远小于 Recommendation 对象，我不希望它大于几 KB)。

我应该在上面的代码中的什么地方放置一个 try...catch 来从这种偶尔超大的对象中恢复？

Answer 1

在不同级别有多个配置:

1. 您有一个代理设置 message.max.bytes(默认为 1000012)(参见 http://kafka.apache.org/documentation/#brokerconfigs)
2. 有一个主题级别的配置max.message.bytes(默认为1000012)(cf http://kafka.apache.org/documentation/#topicconfigs)
3. 生产者有 max.request.size(默认为 1048576)(参见 http://kafka.apache.org/documentation/#producerconfigs)

堆栈跟踪表明，您需要更改代理或主题级别的设置:

Caused by: org.apache.kafka.common.errors.RecordTooLargeException: The request included a message larger than the max message size the server will accept.

也许您还需要增加生产者设置。

为什么你首先需要这个:

当您执行 KStream-KStream 连接时，连接运算符会建立状态(它必须缓冲来自两个流的记录才能计算连接)。默认情况下，状态由 Kafka 主题支持——本地状态基本上是一个缓存，而 Kafka 主题是事实的来源。因此，您的所有记录都将写入 Kafka Streams 自动创建的这个“变更日志主题”。