如何在运行时重新配置Kafka Streams过滤器（无需重启应用）

核心思路 #

利用GlobalKTable存储全局过滤配置（无需与数据主题键关联），结合KStream全量消费数据主题的能力，在配置更新时重置消费位移触发全量数据重处理，最终将结果输出为压缩KTable，确保输出主题始终反映最新过滤规则下的应用状态。

具体实现步骤 #

1. 定义配置结构与Serde #

先定义过滤配置实体类，包含版本号和过滤规则，再实现对应Serde用于序列化/反序列化：

public class FilterConfig {
    private int version;
    private List<String> allowedIds; // 示例过滤规则：允许的ID列表
    // getter、setter、构造方法
}

// 基于JSON实现的自定义Serde
public class FilterConfigSerde extends Serdes.WrapperSerde<FilterConfig> {
    public FilterConfigSerde() {
        super(new JsonSerializer<>(), new JsonDeserializer<>(FilterConfig.class));
    }
}

2. 加载全局配置表 #

将存储过滤配置的Kafka主题加载为GlobalKTable，确保每个应用实例能实时获取最新配置：

StreamsBuilder builder = new StreamsBuilder();

// 加载全局配置表，配置主题的键固定为"current-config"
GlobalKTable<String, FilterConfig> filterConfigGlobalTable = builder.globalTable(
    "filter-config-topic",
    Consumed.with(Serdes.String(), new FilterConfigSerde())
);

3. 构建数据处理拓扑 #

将数据主题作为KStream消费（方便从头全量消费），关联GlobalKTable后应用过滤规则，最终转换为KTable输出到压缩主题：

// 消费数据主题为KStream
KStream<String, DataRecord> dataStream = builder.stream(
    "input-data-topic",
    Consumed.with(Serdes.String(), new DataRecordSerde())
);

// 关联全局配置表，动态应用过滤规则
KStream<String, DataRecord> filteredStream = dataStream.transformValues(
    () -> new ValueTransformerWithKey<String, DataRecord, DataRecord>() {
        private KeyValueStore<String, FilterConfig> configStore;

        @Override
        public void init(ProcessorContext context) {
            // 获取GlobalKTable对应的状态存储
            configStore = context.getStateStore("filter-config-topic");
        }

        @Override
        public DataRecord transform(String key, DataRecord value) {
            // 获取最新过滤配置
            FilterConfig latestConfig = configStore.get("current-config");
            if (latestConfig == null) {
                // 默认逻辑：允许所有数据
                return value;
            }
            // 应用过滤规则：检查数据ID是否在允许列表中
            return latestConfig.getAllowedIds().contains(value.getId()) ? value : null;
        }

        @Override
        public void close() {}
    },
    "filter-config-topic" // 指定依赖的GlobalKTable状态存储名称
);

// 将过滤后的Stream转为KTable，输出到压缩主题
filteredStream.toTable(
    Materialized.<String, DataRecord, KeyValueStore<Bytes, byte[]>>as("output-state-store")
        .withKeySerde(Serdes.String())
        .withValueSerde(new DataRecordSerde())
        .withLoggingEnabled(Map.of(TopicConfig.COMPRESSION_TYPE_CONFIG, "lz4")) // 开启输出压缩
).toStream().to("output-filtered-topic", Produced.with(Serdes.String(), new DataRecordSerde()));

4. 配置更新时触发全量重处理 #

当过滤配置更新时，通过监听配置主题+重置消费位移的方式，触发应用从头全量消费数据主题：

// 初始化AdminClient用于操作消费位移
AdminClient adminClient = AdminClient.create(Map.of(
    AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker:9092"
));

// 单独线程监听配置主题的版本更新
KafkaConsumer<String, FilterConfig> configConsumer = new KafkaConsumer<>(Map.of(
    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker:9092",
    ConsumerConfig.GROUP_ID_CONFIG, "filter-config-monitor",
    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"
), Serdes.String().deserializer(), new FilterConfigSerde().deserializer());
configConsumer.subscribe(Collections.singletonList("filter-config-topic"));

new Thread(() -> {
    int lastVersion = -1;
    while (true) {
        ConsumerRecords<String, FilterConfig> records = configConsumer.poll(Duration.ofSeconds(1));
        for (ConsumerRecord<String, FilterConfig> record : records) {
            FilterConfig newConfig = record.value();
            if (newConfig.getVersion() > lastVersion) {
                lastVersion = newConfig.getVersion();
                // 重置应用消费者组在数据主题的位移到最早位置
                Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
                // 遍历数据主题所有分区，统一重置位移
                List<TopicPartition> partitions = adminClient.describeTopics(Collections.singletonList("input-data-topic"))
                    .values().get("input-data-topic").get().partitions().stream()
                    .map(p -> new TopicPartition("input-data-topic", p.partition()))
                    .collect(Collectors.toList());
                partitions.forEach(p -> offsetSpecs.put(p, OffsetSpec.earliest()));
                
                adminClient.alterConsumerGroupOffsets(
                    "kafka-streams-filter-app-group",
                    offsetSpecs
                );
                System.out.println("过滤配置更新，触发全量数据重处理");
            }
        }
    }
}).start();

替代方案：Processor API细粒度控制 #

如果需要更灵活的状态管理或批量逻辑，可直接使用Processor API：

// 构建Processor拓扑
builder.addSource("data-source", Serdes.String().deserializer(), new DataRecordSerde().deserializer(), "input-data-topic")
       .addProcessor("filter-processor", FilterProcessor::new, "data-source")
       .addStateStore(Stores.keyValueStoreBuilder(
           Stores.persistentKeyValueStore("filter-config-store"),
           Serdes.String(),
           new FilterConfigSerde()
       ), "filter-processor")
       .addSink("output-sink", "output-filtered-topic", Serdes.String().serializer(), new DataRecordSerde().serializer(), "filter-processor");

// 自定义FilterProcessor
public class FilterProcessor extends AbstractProcessor<String, DataRecord> {
    private KeyValueStore<String, FilterConfig> configStore;
    private int lastConfigVersion = -1;

    @Override
    public void init(ProcessorContext context) {
        super.init(context);
        configStore = context.getStateStore("filter-config-store");
        // 每5秒检查一次配置更新
        context.schedule(Duration.ofSeconds(5), PunctuationType.WALL_CLOCK_TIME, timestamp -> {
            FilterConfig latestConfig = configStore.get("current-config");
            if (latestConfig != null && latestConfig.getVersion() > lastConfigVersion) {
                lastConfigVersion = latestConfig.getVersion();
                // 触发消费位移重置逻辑（同上述AdminClient代码）
            }
        });
    }

    @Override
    public void process(String key, DataRecord value) {
        FilterConfig latestConfig = configStore.get("current-config");
        if (latestConfig == null || latestConfig.getAllowedIds().contains(value.getId())) {
            context().forward(key, value);
        }
    }
}

注意事项 #

• 输出主题必须配置压缩（如LZ4、GZIP），确保新处理的记录能覆盖旧状态，最终仅保留最新过滤后的有效数据。
• 配置主题的消息需保证幂等性，通过版本号判断是否触发更新，避免重复全量重处理。
• 重置消费位移时，建议配合Kafka Streams的重平衡机制，确保实例能平稳重新消费数据。

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