Kafka Streams时间过滤：动态窗口配置及状态同步问题解决方案

需求描述 #

需要实现基于时间过滤Kafka消息：每N秒仅转发一条同Key的消息，同时支持运行时动态调整窗口大小（比如从2秒修改为10秒）。

示例场景 #

原始消息（按时间顺序）：

1st second: 1 -> 23（1为Key，23为Value）
2nd second: 1 -> 445
3rd second: 1 -> 5
4th second: 1 -> 234
5th second: 1 -> 777

当窗口设为2秒时，过滤后需转发到目标主题的消息：

1 -> 23
1 -> 5
1 -> 777

更新问题 #

现有单事件场景下的代码，但批量事件时GlobalStore lastSeen无法同步更新——该存储似乎只有当数据推送到Kafka后才会接收更新。能否在推送时间戳到Kafka的同时，同步更新本地内存中的lastSeen状态？

现有代码：

KStream<String,String> input = streamsBuilder.stream("in", Consumed.with(Serdes.String(), Serdes.String()));
// 存储Key -> 上次处理该Key的时间戳
GlobalKTable<String, Long> lastSeen = streamsBuilder.globalTable("ls", Consumed.with(Serdes.String(), Serdes.Long()));
// 存储Key -> 消息过滤的时间间隔（毫秒）
GlobalKTable<String, Long> retentionPolicy = streamsBuilder.globalTable("rp", Consumed.with(Serdes.String(), Serdes.Long()));

record TempRec(String value, Long lastSeen, Long retentionPolicy){}

KStream<String, TempRec> a = input
        .leftJoin(lastSeen, (k1,v1) -> k1, (v1,v2) -> new TempRec(v1, v2, null))
        .leftJoin(retentionPolicy, (k1,v1) -> k1, (v1,v2) -> new TempRec(v1.value(), v1.lastSeen(), v2))
        .filter((k,v) -> Instant.now().minusMillis(v.retentionPolicy()).isAfter(Instant.ofEpochMilli(v.lastSeen())));

// 如何让这里的更新同步执行？
a.mapValues((k,v) -> Instant.now().toEpochMilli())
        .to("ls", Produced.with(Serdes.String(), Serdes.Long()));

a.mapValues((k,v) -> v.value())
        .to("out");

解决方案 #

核心问题分析 #

当前方案的问题在于GlobalKTable是异步更新的：当你把数据写入ls主题后，GlobalKTable需要消费该主题的消息才能更新本地状态，导致批量处理时后续消息无法立即获取最新的lastSeen值，过滤逻辑失效。

优化方案：使用本地状态存储替代GlobalKTable #

要实现同步更新本地状态，应使用Kafka Streams的本地状态存储（KeyValueStore），而非依赖GlobalKTable的异步消费。具体步骤如下：

1. 定义本地状态存储
   创建内存型KeyValueStore存储每个Key的上次处理时间戳，确保状态更新同步：

// 存储上次处理时间的本地状态
StoreBuilder<KeyValueStore<String, Long>> lastSeenStoreBuilder =
        Stores.keyValueStoreBuilder(
                Stores.inMemoryKeyValueStore("last-seen-store"),
                Serdes.String(),
                Serdes.Long()
        );
streamsBuilder.addStateStore(lastSeenStoreBuilder);

// 存储动态时间间隔的本地状态（关联GlobalKTable的配置主题）
StoreBuilder<KeyValueStore<String, Long>> retentionPolicyStoreBuilder =
        Stores.keyValueStoreBuilder(
                Stores.inMemoryKeyValueStore("retention-policy-store"),
                Serdes.String(),
                Serdes.Long()
        );
streamsBuilder.addStateStore(retentionPolicyStoreBuilder);

// 绑定配置主题到GlobalKTable，同步动态时间间隔
streamsBuilder.globalTable(
        "retention-policy-topic",
        Consumed.with(Serdes.String(), Serdes.Long()),
        Materialized.as("retention-policy-store")
);

2. 用Processor API实现同步过滤与状态更新
   通过transformValues结合本地状态存储，实现同步的过滤逻辑：

KStream<String, String> filteredStream = input.transformValues(() -> new ValueTransformerWithKey<String, String, String>() {
    private KeyValueStore<String, Long> lastSeenStore;
    private KeyValueStore<String, Long> retentionPolicyStore;

    @Override
    public void init(ProcessorContext context) {
        // 初始化本地状态存储
        lastSeenStore = context.getStateStore("last-seen-store");
        retentionPolicyStore = context.getStateStore("retention-policy-store");
    }

    @Override
    public String transform(String key, String value) {
        long currentTimestamp = Instant.now().toEpochMilli();
        // 获取当前Key的时间间隔，默认2秒
        Long retentionMillis = retentionPolicyStore.get(key);
        if (retentionMillis == null) {
            retentionMillis = 2000L;
        }

        Long lastProcessedTime = lastSeenStore.get(key);
        // 首次处理或已超过时间间隔，允许转发并同步更新状态
        if (lastProcessedTime == null || currentTimestamp - lastProcessedTime >= retentionMillis) {
            lastSeenStore.put(key, currentTimestamp);
            return value;
        }
        // 不满足条件，返回null过滤消息
        return null;
    }

    @Override
    public void close() {
        // 资源清理
    }
}, "last-seen-store", "retention-policy-store");

// 转发过滤后的消息到目标主题
filteredStream.to("out");

3. 动态调整窗口的实现
   要在运行时修改时间间隔，只需向retention-policy-topic主题发送Key对应的新时间间隔消息（比如1 -> 10000表示Key=1的窗口改为10秒）。GlobalKTable会自动将配置同步到本地状态存储，下一次处理该Key的消息时就会使用新的窗口值。

关键优势 #

• 状态同步更新：本地状态存储的读写是同步操作，批量处理时后续消息能立即获取最新的lastSeen值
• 动态配置低延迟：GlobalKTable消费配置主题的延迟极低，窗口调整几乎实时生效
• 性能更优：避免了额外的Kafka主题写入/消费开销，减少了端到端延迟

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