我尝试使用类MyVeryCustomAggregator在Kafka的SessionWindows上计算聚合，该类将聚合信息存储为属性，并提供方法aggregate和result，分别用于处理新消息和检索最终结果。
我的主要困难在于实现aggregate方法所需的Merger接口，即MyVeryCustomAggregator::mergeWith，我所计算的聚合高度依赖于消息的顺序和时间戳，因此我无法像Confluent文档中那样简单地添加两个聚合器，该文档仅指出：
当窗口基于会话时，您必须另外提供一个“会话合并器”聚合器（例如，mergedAggValue = leftAggValue + rightAggValue）。
使用会话窗口时：无论何时合并两个会话，都调用会话合并。

如能就下列问题提供进一步资料，将不胜感激

• 为什么聚合需要被合并呢？因为我没有使用宽限期，所以我希望每次会话关闭时只计算一次聚合。
• 即使必须进行无序处理，我也希望加载以前的聚合器并用于进一步处理。
• 什么参数被传递给合并器（agg1和agg2）？这些是例如两个连续计算的聚合还是一系列聚合可以由Kafka以任意顺序合并？

示例实现可能更容易在答案中引用。

builder.stream("INPUT_TOPIC", Consumed.with(Serdes.String(), CustomSerdes.Json(MyVeryCustomMessage.class)))
    .groupByKey(Grouped.with(Serdes.String(), CustomSerdes.Json(MyVeryCustomMessage.class)))
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofSeconds(15)))
    .aggregate(
        MyVeryCustomAggregator::new,
        (key, newValue, agg) -> agg.aggregate(newValue),
        (aggKey, agg1, agg2) -> agg1.mergeWith(agg2),
        Named.as("MyVeryCustomAggregator"),
        Materialized.with(Serdes.String(), CustomSerdes.Json(MyVeryCustomAggregator.class))
    )
    .toStream()
    .map((windowedKey, agg) -> KeyValue.pair(windowedKey.key(), agg.result()))
    .to("OUTPUT_TOPIC", Produced.with(Serdes.String(), CustomSerdes.Json(MyVeryCustomOutput.class)));