有没有办法根据大小刷新Kafka Streams的Windows Store？

Question

我正在使用将从kafka主题读取元组的DSL应用编程接口编写Kafka Streams应用程序。在拓扑结构中，我想批处理元组。然后如果（1）30秒已经过去或（2）批处理的大小超过1 GB，我想将浴写入磁盘上的文件。

我写的拓扑使用TimeWindowedKStream对元组进行分组。然后调用聚合并传递窗口存储。

我的问题是，当国有商店试图写入Kafka变更日志时，我得到了一个

组织。阿帕奇。Kafka。常见的错误。RecordTooLargeException

例外。

特别地：

原因：org.apache.kafka.streams.errors.StreamsExctive：任务[1_1]由于用以前的记录捕获错误（key\x00\x00\x00\x06\x00\x00\x01h$\xE7\x88\x00\x00\x00\x00\x00\x00值[B@419761c时间戳1546807396524)主题ibv2-capt-消费者-组-3-记录-存储-变更日志由于org.apache.kafka.common.errors.RecordTooLargeExctive：请求包含的消息大于服务器将接受的最大消息大小。.

我尝试将缓存\u最大字节\u缓冲\u配置设置为1MB，但正如文档中所述，该配置适用于整个拓扑。

这是我的拓扑结构

这是我一直在使用的Scala代码。注意，我在这里使用的是kafka streams scala。

val builder = new StreamsBuilderS()

import com.lightbend.kafka.scala.streams.DefaultSerdes._

implicit val recordSerde = (new RecordSerde).asInstanceOf[Serde[Record]]
implicit val recordSeqSerde = (new RecordSeqSerde).asInstanceOf[Serde[RecordSeq]]

val inputStream: KStreamS[String, Record] = builder.stream[String,Record](topic)

val keyed = inputStream.selectKey[Int]((k,r) => random.nextInt(10)) 

val grouped: TimeWindowedKStreamS[Int, Record] = keyed.groupByKey.windowedBy(TimeWindows.of(TimeUnit.SECONDS.toMillis(30L)))

import org.apache.kafka.common.utils.Bytes

val windowedStore: Materialized[Int, RecordSeq, WindowStore[Bytes, Array[Byte]]] = Materialized
  .as[Int,RecordSeq,WindowStore[Bytes, Array[Byte]]]("record-store")
  .withKeySerde(integerSerde)
  .withValueSerde(recordSeqSerde)
  .withLoggingEnabled(ChangeLogConfig.getChangeLogConfig.asJava)  // increased max.request.size to 10 x default

val records: KTableS[Windowed[Int], RecordSeq] = grouped.aggregate(
  () => RecordSeq(Seq()),
  (randon: Int, record: Record, recordSeq: RecordSeq) => RecordSeq(recordSeq.records :+ record),
  windowedStore
)

val recordSeqStream: KStreamS[String, RecordSeq] = records.toStream((ws, r) => s"${ws.key()}-${ws.window().start()}-${ws.window().end()}")

recordSeqStream.foreach((k: String, rs: RecordSeq) => WrappedRecordFileWriter.write(k, rs))

注意：case类RecordSeq（记录：Seq[记录]）




Answer 1

主题可以有消息中定义的最大大小的记录。最大字节数属性。这是代理可以在主题中接收和附加的最大消息大小。您的记录大小可能超过该限制。因此，您需要更改此属性的配置，以允许更大的记录大小。

它可以在代理级别和主题级别设置。您可以在这里参考更多详细信息：

http://kafka.apache.org/documentation/#brokerconfigs

http://kafka.apache.org/documentation/#topicconfigs