从基于日窗口的状态存储中获取非压缩键/值答案

【问题标题】：Getting non compacted key/value from day window based statestore从基于日窗口的状态存储中获取非压缩键/值
【发布时间】：2020-08-05 12:43:46
【问题描述】：

拓扑定义：

 KStream<String, JsonNode> transactions = builder.stream(inputTopic, Consumed.with(Serdes.String(), jsonSerde));
    
     KTable<Windowed<String>, JsonNode> aggregation =
                    transactions
                            .groupByKey()
                            .windowedBy(
                                    TimeWindows.of(Duration.ofSeconds(windowDuration)).grace(Duration.ofSeconds(windowGraceDuration)))
                            .aggregate(() -> new Service().buildInitialStats(),
                                    (key, transaction, previous) -> new Service().build(key, transaction, previous),
                                    Materialized.<String, JsonNode, WindowStore<Bytes, byte[]>>as(statStoreName).withRetention(Duration.ofSeconds((windowDuration + windowGraceDuration + windowRetentionDuration)))
                                            .withKeySerde(Serdes.String())
                                            .withValueSerde(jsonSerde)
.withCacheDisabled())
                            .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()));
            aggregation.toStream()
                      .to(outputTopic, Produced.with(windowedSerde, jsonSerde));

State Store API：通过查找所有时间窗口来获取密钥。

 Instant timeFrom = Instant.ofEpochMilli(0);
    Instant timeTo = Instant.now();
    WindowStoreIterator<ObjectNode> value = store.fetch(key,timeFrom,timeTo);
    while(value.hasNext()){
                System.out.println(value.next());
            }

作为测试的一部分，执行 2 个事务并生成密钥 1，我的要求是在查找 statestore 时获取 key1 两次（当前和上一个）而不进行压缩。结果总是返回带有键和最终聚合值的最终结果。

Txn1 --> Key - Key1 |值 - {Count=1,attribute='test'}

Txn2 --> Key - Key1 |值 - {Count=2,attribute='test1'}

statestore 查找后的当前行为：始终使用 value = {Count=2,attribute='test1'} 获得压缩的 key1

相反，我想在该窗口持续时间内获取所有 key1。

作为解决方案的一部分，我做了以下更改，但不幸的是它没有奏效。

在拓扑级别禁用缓存
cache.max.bytes.buffering 到 0
从内部变更日志主题中手动删除精简策略

怀疑变更日志主题被压缩，因此在调用 statestore api 时获得压缩键。

需要进行哪些更改才能通过 statestore API 获取非压缩密钥？

【问题讨论】：

@MatthiasJ.Sax - 需要您的专家意见。如果可能的话，请看看我的问题。
顺便说一句：标记在 SO 上不起作用——我没有收到任何通知，只是在浏览时遇到了这个问题......

标签： apache-kafka-streams

【解决方案1】：

如果你想得到所有的中间结果，你不应该使用suppress() 操作符。 suppress() 旨在为每个窗口发出一条结果记录，也就是说，它的作用与您想要的完全相反。

【讨论】：