TopologyTestDriver 在 KTable 聚合上发送错误消息

Question

我有一个在 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之前发送的消息处于错误的阶段。

其他人可以确认这是一个错误吗？我已经针对 Kafka 版本 2.0.1 和 2.1.0 对此进行了测试。

编辑：
我在 github 中创建了一个测试用例来说明这个问题：https://github.com/mulho/topology-testcase

Answer 1

有两条输出记录（一条“减号”记录和一条“加号”记录）是预期行为。理解它的工作原理有点棘手，所以让我试着解释一下。

假设你有以下输入表：

 key |  value
-----+---------
  A  |  <10,2>
  B  |  <10,3>
  C  |  <11,4>

在KTable#groupBy() 上，您将值的第一部分提取为新键（即10 或11），然后对第二部分求和（即2、3、4）在聚合中。因为A 和B 记录都将10 作为新键，所以您可以对2+3 求和，也可以对4 求和作为新键11。结果表将是：

 key |  value
-----+---------
  10 |  5
  11 |  4

现在假设一条更新记录<B,<11,5>>将原来的输入KTable改为：

 key |  value
-----+---------
  A  |  <10,2>
  B  |  <11,5>
  C  |  <11,4>

因此，新结果表应将5+4 与11 和2 与10 相加：

 key |  value
-----+---------
  10 |  2
  11 |  9

如果您将第一个结果表与第二个结果表进行比较，您可能会注意到 两行 行都得到了更新。从10|5 中减去旧的B|<10,3> 记录得到10|2，并将新的B|<11,5> 记录添加到11|4 得到11|9。

这正是您看到的两条输出记录。第一条输出记录（在执行减法之后）更新第一行（它减去不再属于聚合结果的旧值），而第二条记录将新值添加到聚合结果中。在我们的例子中，减记录是<10,<null,<10,3>>>，加记录是<11,<<11,5>,null>>（这些记录的格式是<key, <plus,minus>>（注意减记录只设置minus部分，而加记录只设置plus 部分）。

最后说明：加号和减号记录不能放在一起，因为加号和减号记录的键可以不同（在我们的示例中为11和10），因此可能进入不同的分区.这意味着加号和减号操作可能由不同的机器执行，因此不可能只发出一条同时包含加号和减号的记录。