kafka本地状态存储/changelog中的保留时间

Question

我正在用Kafka和Kafka溪流作为Spring-Cloud-Stream流的一部分。在我的Kafka Streams应用程序中流动的数据在特定的时间窗口内被聚合和物化：

Materialized<String, ErrorScore, WindowStore<Bytes, byte[]>> oneHour = Materialized.as("one-hour-store");
    oneHour.withLoggingEnabled(topicConfig);
    events
            .map(getStringSensorMeasurementKeyValueKeyValueMapper())
            .groupByKey()
            .windowedBy(TimeWindows.of(oneHourStore.getTimeUnit()))
            .reduce((aggValue, newValue) -> getMaxErrorScore(aggValue, newValue),
                    (oneHour));

按照设计，正在具体化的信息也由changelog主题支持。

 ReadOnlyWindowStore<String, Double> windowStore =  queryableStoreRegistry.getQueryableStoreType("one-hour-store", QueryableStoreTypes.windowStore());
 WindowStoreIterator<ErrorScore> iter = windowStore.fetch(key, from, to);

min.insync.replicas 1
cleanup.policy delete
retention.ms 5259600000
retention.bytes -1

用解决方案更新Kafka Streams 2.0.1版不包含Materialized.WithRetention方法。对于这个特定的版本，我可以使用以下代码设置状态存储的保留时间，这解决了我的问题：

TimeWindows timeWindows = TimeWindows.of(windowSizeMs);
    timeWindows.until(retentionMs);

使我的代码编写为：

...

.groupByKey()
        .windowedBy(timeWindows)
        .reduce((aggValue, newValue) -> getMaxErrorScore(aggValue, newValue),
                (oneHour));
...

Answer 1

对于窗口化的ktables，存在本地保留时间和changlog保留时间。您可以通过Materialized.WithRetentionTime(...)设置本地存储保留时间--默认值为24h。

对于较早的Kafka版本，本地存储保持时间是通过Windows#tool()设置的。

如果创建了一个新的应用程序，changelog主题的保留时间将与本地存储区的保留时间相同。但是，如果手动增加日志保留时间，这不会影响存储区保留时间，但需要相应地更新代码。当changelog主题已经存在时也是如此：如果更改本地存储保持时间，changelog主题配置不会自动更新。

还有一个Jira:https://issues.apache.org/Jira/browse/kafka-7591