Kafka Streams错误日志为何存字节数组而非JSON？变更日志发送失败求助

这个问题我太熟悉了——当你用ArrayList持续累加ID做聚合时，状态数据会随着时间无限膨胀，最终触发Kafka的消息大小限制，导致变更日志发送失败。咱们从业务优化、配置调整、存储结构三个维度来解决：

一、从业务逻辑入手，减小状态体积 #

这是最根本的解决办法，毕竟如果能避免状态无限增长，后续的麻烦就都没了：

• 对ID去重：如果你的场景不需要重复存储同一个ID，把ArrayList换成HashSet（或者LinkedHashSet保留插入顺序），这样每个ID只会存一次，能大幅减少状态大小。示例代码：

  // 聚合时用HashSet替代ArrayList
  KStream<String, Event> inputStream = ...;
  inputStream.groupByKey()
             .aggregate(
                 HashSet::new,
                 (key, value, aggregate) -> {
                     aggregate.add(value.getId());
                     return aggregate;
                 },
                 Materialized.as("id-aggregate-store")
             );
  

• 限制列表长度：如果只需要保留最近N个ID，在聚合时检查列表大小，超过阈值就移除最早的元素：

  private static final int MAX_LIST_SIZE = 1000;
  // 聚合逻辑
  (key, value, aggregate) -> {
      if (aggregate.size() >= MAX_LIST_SIZE) {
          aggregate.remove(0); // 移除第一个元素
      }
      aggregate.add(value.getId());
      return aggregate;
  };
  

• 使用窗口聚合：如果你的ID有时间有效期（比如只需要最近24小时的ID），用窗口聚合自动清理过期状态：

  inputStream.groupByKey()
             .windowedBy(TimeWindows.of(Duration.ofHours(24)))
             .aggregate(
                 ArrayList::new,
                 (key, value, aggregate) -> {
                     aggregate.add(value.getId());
                     return aggregate;
                 },
                 Materialized.as("id-window-store")
             );
  

二、调整Kafka消息大小限制（治标方案） #

如果业务上必须保留所有ID，那只能调整Kafka的相关配置，允许更大的消息：

• 生产者配置：在Kafka Streams应用的配置中设置：

  producer.max.request.size=10485760  # 10MB，根据实际需求调整
  

• Broker配置：在Kafka集群的server.properties中修改：

  message.max.bytes=10485760  # 必须大于等于生产者的max.request.size
  replica.fetch.max.bytes=10485760  # 副本同步用，也要对应调整
  

• 消费者配置：如果你的应用还有下游消费者读取变更日志，也要调整：

  consumer.max.partition.fetch.bytes=10485760
  

注意：调大消息大小会增加Broker的内存和磁盘压力，也会影响消息的传输延迟，谨慎使用。

三、优化状态存储与序列化 #

• 启用变更日志压缩：给状态存储的变更日志主题设置cleanup.policy=compact，让Kafka自动清理旧的状态记录（只保留最新的状态值），减少日志体积。可以在Materialized时指定：

  Materialized.<String, HashSet<String>, KeyValueStore<Bytes, byte[]>>as("id-aggregate-store")
              .withLoggingEnabled(Map.of("cleanup.policy", "compact"));
  

• 使用更紧凑的序列化器：默认的JSON序列化比较占空间，换成Avro、Protobuf这类二进制序列化格式，能大幅压缩消息大小。比如用Avro序列化HashSet：

  SpecificAvroSerde<HashSet<String>> avroSerde = new SpecificAvroSerde<>();
  avroSerde.configure(Map.of(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://your-schema-registry:8081"), false);
  
  Materialized.<String, HashSet<String>, KeyValueStore<Bytes, byte[]>>as("id-aggregate-store")
              .withValueSerde(avroSerde);
  

四、调试建议 #

先确认是哪个状态存储导致的问题：

• 用Kafka Streams的状态查询API查看当前状态的大小：

  ReadOnlyKeyValueStore<String, HashSet<String>> store = streams.store("id-aggregate-store", QueryableStoreTypes.keyValueStore());
  store.all().forEachRemaining(entry -> System.out.println("Key: " + entry.key() + ", Size: " + entry.value().size()));
  

• 用kafka-console-consumer.sh直接读取变更日志主题的消息，查看具体的消息大小：

  kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic id-aggregate-store-changelog --formatter kafka.tools.DefaultMessageFormatter --property print.key=true --property print.value=true --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer --property value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
  

内容的提问来源于stack exchange，提问作者mishless