如何测试 WindowStore 的保留期？

Question

我正在尝试对传入的 kafka 消息进行重复数据删除（我正在轮询一个数据源，该数据源使给定日期的所有数据点在第二天可用，但时间不一致，所以我每 x 分钟轮询一次，并且我想对数据点进行重复数据删除以获得一个仅包含新点的干净的下游主题）。

为此，我构建了一个自定义转换器，它依赖于商店来跟踪已经处理了哪个“点”。由于数据点的日期时间是重复数据删除键的一部分，我有一组无限的键，所以我不能依赖简单的 KeyValueStore。据我了解，WindowStore 只允许我将密钥保留一段特定的保留期（在我的情况下为 2 天），所以这就是我正在使用的。

我尝试使用 kafka-streams-test-utils 测试重复数据删除。重复数据删除工作得很好，但 windowStore 似乎没有“忘记”密钥。我尝试使用较短的 window 大小和持续时间 (1s)，但我仍然无法让它忘记超过保留期的 keys/values。

商店的配置：我希望物品在商店中停留约 2 秒

config.put(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG,"1");
...
final StoreBuilder<WindowStore<String, AvroBicycleCount>> deduplicationStoreBuilder = Stores.windowStoreBuilder(
            Stores.persistentWindowStore(deduplicationStore, Duration.ofSeconds(1), Duration.ofSeconds(1), false),
            Serdes.String(),
            StreamUtils.AvroSerde()
);

我的变压器逻辑

@Override
public DataPoint transform(final String dataId, final DataPoint incoming) {
    String key = dataId+"_"+incoming.getDateTime();
    DataPoint previous = windowStore.fetch(key, incoming.getDateTime());
    if(previous != null)
        return null;
    
    windowStore.put(key, incoming, incoming.getDateTime());
    return incoming;
}

第三次测试失败

inputTopic.pipeInput("a", newDataPoint);
assertEquals(1, outputTopic.readRecordsToList().size(), "When a new data is emitted, it should go through");
    
inputTopic.pipeInput("a", newDataPoint);
assertEquals(0, outputTopic.readRecordsToList().size(), "When the same data is re-emitted, it should not go through");
    
TimeUnit.SECONDS.sleep(10);

inputTopic.pipeInput("a", newDataPoint);
assertEquals(1, outputTopic.readRecordsToList().size(), "When the same data is re-emitted well past the retention period, it should go through");

对于 window商店的留存率，我有什么不正确的地方吗？

Answer 1

A WindowedStore 在内部使用 so-called 段使数据过期。也就是说，你的 retention-time 的 time-range 被拆分成更小的 time-range 并且每个 time-range 都有一个段来存储相应的数据（在内部，一个段映射到一个store，即一个WindowedStore实际上是内部的多个store）。如果段中的所有条记录都过期，则通过删除相应的存储来删除整个段（这比 record-by-record 过期更有效）。

此外，最小 (hard-coded) 段大小为 60 秒，段数为 2（硬编码），以避免段太小（且效率低下）。因此，对于 2 天保留时间的情况，您将获得两个时间范围为 1 天的片段。因此，数据（在段的开头）最多可以保留 3 天，直到旧段被删除。

因此，数据被有效删除有一些延迟。不能配置段数

如何测试 WindowStore 的保留期？

How to test a WindowStore retention period?

apache-kafka

apache-kafka-streams