如何为Quarkus Kafka Streams物化状态存储设置LZ4主题压缩？

你的场景是：在Quarkus Kafka Streams应用中执行重键与聚合操作后，系统隐式创建了communications-grouped-repartition（重分区主题）和communications-grouped-aggregated-materialized-changelog（状态存储变更日志主题），但全局配置kafka-streams.producer.compression-type=lz4对这两个主题无效，需要将它们的压缩类型改为LZ4。

解决方案：针对操作显式配置生产者压缩参数 #

全局producer配置无法覆盖Kafka Streams内部为隐式主题创建的生产者实例，需要分别在Grouped和Materialized配置中显式指定压缩参数：

1. 处理重分区主题
   重分区主题由groupBy操作生成，需在Grouped配置中添加生产者参数：

   Grouped.with(
           "communications-grouped",
           longKeySerde,
           communicationSerde)
           .withProducerConfig(Collections.singletonMap(
                   ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4"
           ))
   

2. 处理变更日志主题
   变更日志主题由物化存储生成，需在Materialized配置中添加生产者参数：

   Materialized
           .<Long, GroupedCommunications, KeyValueStore<Bytes, byte[]>>as("communications-grouped-aggregated-materialized")
           .withKeySerde(longKeySerde)
           .withValueSerde(groupedCommunicationSerde)
           .withProducerConfig(Collections.singletonMap(
                   ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4"
           ))
   

修改后的完整代码 #

import java.util.Collections;
import org.apache.kafka.clients.producer.ProducerConfig;

final KTable<Long, GroupedCommunications> groupedCommunications = communications //
        .groupBy( //
                (key, value) -> KeyValue.pair(value.personId(), value), //
                Grouped.with( //
                        "communications-grouped", //
                        longKeySerde, //
                        communicationSerde)
                        .withProducerConfig(Collections.singletonMap(
                                ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4"
                        ))) //
        .aggregate( //
                GroupedCommunications::new, //
                (key, value, aggregate) -> aggregate.add(value), //
                (key, value, aggregate) -> aggregate.remove(value), //
                Named.as("communications-grouped-aggregated"), //
                Materialized //
                        .<Long, GroupedCommunications, KeyValueStore<Bytes, byte[]>>as("communications-grouped-aggregated-materialized") //
                        .withKeySerde(longKeySerde) //
                        .withValueSerde(groupedCommunicationSerde)
                        .withProducerConfig(Collections.singletonMap(
                                ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4"
                        )));

原理说明 #

Kafka Streams在创建重分区和变更日志主题时，会使用独立的内部生产者实例，这些实例不会继承全局的kafka-streams.producer.*配置，必须通过对应操作的配置器（Grouped/Materialized）显式传入生产者参数，才能覆盖默认的压缩类型设置。

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