如何测试 WindowStore 保留期？答案

【问题标题】：How to test a WindowStore retention period?如何测试 WindowStore 保留期？
【发布时间】：2020-08-18 23:52:23
【问题描述】：

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

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

我尝试使用 kafka-streams-test-utils 测试重复数据删除。重复数据删除工作得很好，但 windowStore 似乎并没有“忘记”密钥。我尝试使用更短的窗口大小和持续时间（1 秒），但我仍然无法让它忘记超过保留期的键/值。

存储配置：我希望对象在存储中停留约 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");

我对 windowStore 的保留有什么不正确的理解吗？

【问题讨论】：

标签： apache-kafka apache-kafka-streams

【解决方案1】：

WindowedStore 在内部使用所谓的 segments 来使数据过期。即，您的保留时间的时间范围被分成更小的时间范围，并且每个时间范围都有一个段来存储相应的数据（在内部，一个段映射到一个存储，即WindowedStore 是实际上内部有多家商店）。如果段中的所有条记录都已过期，则通过删除相应的存储来删除整个段（这比逐记录过期更有效）。

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

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

【讨论】：