使用Kafka Streams实现Kafka事件去重失效，求解决方案

问题背景 #

基于Spring Boot的应用需要消费Kafka Topic中的事件，但该Topic存在外部生成的重复事件（上游逻辑无法修改）。采用Kafka Streams基于Key实现去重，已配置EXACTLY_ONCE_V2语义，但下游消费者仍收到重复事件，需排查问题原因，同时疑惑是否需要改用Spring Cloud Streams。

现有代码 #

配置类代码 #

@Configuration
@EnableKafkaStreams
public class KafkaStreamsConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapAddress;

    @Value("${spring.application.name}")
    private String applicationId;


    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kStreamsConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "deduplication-id");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        return new KafkaStreamsConfiguration(props);
    }


    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        KStream<String, String> stream = streamsBuilder.stream("inputTopic", Consumed.with(Serdes.String(), Serdes.String()))
                .groupByKey()
                .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofSeconds(30), Duration.ofMillis(3000)))
                .reduce((oldValue, newValue) -> newValue)
                .toStream()
                .map(((stringWindowed, value) -> new KeyValue<>(stringWindowed.key(), value)))
                .to("outputTopic", Produced.with(Serdes.String(), Serdes.String()));
        return stream;
    }
}

消费者代码 #

@KafkaListener(topics = {"output-topic"}, groupId = "spring.kafka.consumer.group-id")
    public void consumeEventRelease(ConsumerRecord<String, String> consumerRecord)  {
        log.info("Kafka consumer received message {}", consumerRecord.value());
        //some further implementation
    }

问题排查与解决方案 #

1. Topic名称不匹配（最直接原因） #

注意到Kafka Streams输出的Topic是outputTopic，但消费者监听的是output-topic，名称大小写和连字符不一致，导致消费者实际监听的是另一个未经过去重处理的Topic。需将两者名称统一为完全一致的字符串。

2. 窗口去重的局限性 #

当前使用的滚动窗口（TimeWindows）仅能处理同一窗口内的重复事件：

• 若同一Key的重复事件间隔超过30秒（窗口大小），会被分到不同窗口，最终仍会被输出。
• 如果需要全局去重（无论间隔多久，同一Key只保留最新/首次事件），需改用持久化状态存储实现，而非窗口聚合。

全局去重示例代码：

@Bean
public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
    // 创建持久化状态存储，用于存储已处理过的Key-Value
    StoreBuilder<KeyValueStore<String, String>> dedupStore =
            Stores.keyValueStoreBuilder(
                    Stores.persistentKeyValueStore("dedup-store"),
                    Serdes.String(),
                    Serdes.String());
    streamsBuilder.addStateStore(dedupStore);

    return streamsBuilder.stream("inputTopic")
            .transformValues(() -> new ValueTransformerWithKey<String, String, String>() {
                private KeyValueStore<String, String> store;

                @Override
                public void init(ProcessorContext context) {
                    this.store = context.getStateStore("dedup-store");
                }

                @Override
                public String transform(String key, String value) {
                    String existingValue = store.get(key);
                    // 仅当Key不存在或值不同时，才输出并更新存储
                    if (existingValue == null || !existingValue.equals(value)) {
                        store.put(key, value);
                        return value;
                    }
                    // 重复事件返回null，后续过滤掉
                    return null;
                }

                @Override
                public void close() {}
            }, "dedup-store")
            .filter((key, value) -> value != null) // 过滤重复事件的null值
            .to("outputTopic");
}

3. EXACTLY_ONCE_V2的作用误区 #

EXACTLY_ONCE_V2保证的是Kafka Streams处理过程的精确一次语义（即故障恢复后不会重复输出），但不负责业务层面的去重逻辑。业务去重仍需依赖状态存储或窗口逻辑实现。

4. 是否需要改用Spring Cloud Streams？ #

不需要。Spring Cloud Streams是Kafka Streams等中间件的上层封装，本质仍依赖Kafka Streams的核心能力。当前问题属于代码逻辑和配置错误，改用Spring Cloud Streams无法解决本质问题，反而会增加复杂度。

总结 #

优先检查并修复Topic名称不匹配问题，再根据实际去重需求选择方案：短期窗口内去重可保留现有窗口逻辑，全局去重则使用持久化状态存储实现。无需切换到Spring Cloud Streams。

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