我有一个在ktable上聚合的拓扑。这是我创建的一个通用方法，用于在不同的主题上构建此拓扑。

public static <A, B, C> KTable<C, Set<B>> groupTable(KTable<A, B> table, Function<B, C> getKeyFunction,
        Serde<C> keySerde, Serde<B> valueSerde, Serde<Set<B>> aggregatedSerde) {
    return table
            .groupBy((key, value) -> KeyValue.pair(getKeyFunction.apply(value), value),
                    Serialized.with(keySerde, valueSerde))
            .aggregate(() -> new HashSet<>(), (key, newValue, agg) -> {
                agg.remove(newValue);
                agg.add(newValue);
                return agg;
            }, (key, oldValue, agg) -> {
                agg.remove(oldValue);
                return agg;
            }, Materialized.with(keySerde, aggregatedSerde));
}

这在使用kafka时非常有效，但在通过“topologytestdriver”进行测试时就不行了。
在这两种情况下，当我得到一个更新时 subtractor 先调用，然后调用 adder 被称为。问题是当使用 TopologyTestDriver ，将发送两条消息进行更新：一条在 subtractor 打电话，然后再打一个 adder 打电话。更不用说在 subrtractor 在那之前 adder 处于不正确的阶段。
其他人能确认这是个窃听器吗？我在Kafka2.0.1版和2.1.0版上都测试过这个。
编辑：
我在github中创建了一个测试用例来说明这个问题：https://github.com/mulho/topology-testcase