如何测试WindowStore的保留期

Question

我正在尝试消除传入kafka消息的重复数据(我正在轮询一个数据源，该数据源使给定一天的所有数据点在第二天都可用，但时间不一致，所以我每x分钟轮询一次，我想消除数据点的重复数据，以便有一个只包含新点的干净的下游主题(。

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

我尝试使用kafka流测试实用程序测试重复数据消除。重复数据消除工作得很好，但windowStore似乎没有"恢复"；忘记"；钥匙。我尝试使用较短的窗口大小和持续时间(1s(，但我仍然无法让它忘记超过保留期的键/值。

商店的配置：我希望对象在商店中停留约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的保留，是否有我理解不正确的地方？

Answer 1

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

此外，存在60秒的最小(硬编码(段大小，并且段的数量为2(硬编码的(，以避免太小(和低效(的段。因此，对于保留时间为2天的情况，您会得到两个时间范围为1天的片段。因此，数据(在一个片段的开头(可以长达3天，直到旧片段被丢弃。

因此，数据在一定延迟的情况下被有效地删除。您无法配置段的数量