如何对使用会话窗口的 kafka 流应用程序进行单元测试答案

【问题标题】：How to unit test a kafka stream application that uses session window如何对使用会话窗口的 kafka 流应用程序进行单元测试
【发布时间】：2019-08-13 15:24:38
【问题描述】：

我正在使用 Kafka Stream 2.1

我正在尝试为聚合的流应用程序编写一些测试使用具有 300 毫秒不活动间隙的会话窗口，通过键（即关联 ID）对某些事件进行处理。

这里是一个方法表示的聚合实现：

    private static final int INACTIVITY_GAP = 300;

    public KStream<String, AggregatedCustomObject> aggregate(KStream<String, CustomObject> source) {

        return source
                // group by key (i.e by correlation ID)
                .groupByKey(Grouped.with(Serdes.String(), new CustomSerde()))
                // Define a session window with an inactivity gap of 300 ms
                .windowedBy(SessionWindows.with(Duration.ofMillis(INACTIVITY_GAP)).grace(Duration.ofMillis(INACTIVITY_GAP)))
                .aggregate(
                        // initializer
                        () -> new AggregatedCustomObject(),
                        // aggregates records in same session
                        (s, customObject, aggCustomObject) -> {
                            // ...
                            return aggCustomObject;
                        },
                        // merge sessions
                        (s, aggCustomObject1, aggCustomObject2) -> {
                            // ...
                            return aggCustomObject2;
                        },
                        Materialized.with(Serdes.String(), new AggCustomObjectSerde())
                )
                .suppress(Suppressed.untilWindowCloses(unbounded()))
                .toStream()
                .selectKey((stringWindowed, aggCustomObject) -> "someKey");
    ;
    }

此流式处理按预期工作。但对于单元测试，情况就不同了。

我的测试流配置如下所示：

        // ...

        props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "test");
        props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
        props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, myCustomObjectSerde.getClass());
        // disable cache
        props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
        // commit ASAP
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 0);


        StreamsBuilder builder = new StreamsBuilder();
        aggregate(builder.stream(INPUT_TOPIC), OUTPUT_TOPIC, new AggCustomObjectSerde())
.to(OUTPUT_TOPIC);

        Topology topology = builder.build();
        TopologyTestDriver testDriver = new TopologyTestDriver(topology, props);
        ConsumerRecordFactory<String, MyCustomObject> factory = new ConsumerRecordFactory<>(INPUT_TOPIC, new StringSerializer(), myCustomSerializer)

        // ...

测试如下所示：

List<ConsumerRecord<byte[], byte[]>> records = myCustomMessages.stream()
                .map(myCustomMessage -> factory.create(INPUT_TOPIC, myCustomMessage.correlationId, myCustomMessage))
                .collect(Collectors.toList());
testDriver.pipeInput(records);

ProducerRecord<String, AggregatedCustomMessage> record = testDriver.readOutput(OUTPUT_TOPIC, new StringDeserializer(), myAggregatedCustomObjectSerde);

问题是，record 始终为空。我尝试了很多东西：

循环读取超时
更改配置中的提交间隔，以便尽快提交结果
紧接着发送一条带有不同键的附加记录（以触发窗口关闭，因为在 KafkaStream 中，事件时间基于记录时间戳）
调用测试驱动的advanceWallClockTime方法

好吧，没有什么帮助。有人可以告诉我我缺少什么，我应该如何测试基于会话窗口的流应用程序？

非常感谢

【问题讨论】：

suppress() 只会在事件时间超过窗口结束时间加上宽限期时发出。向提前时间发送和附加记录似乎是正确的 - 您为附加记录分配了什么时间戳？
@MatthiasJ.Sax 谢谢！这实际上是我的问题。我为附加记录设置的时间戳太低，所以 kafkaStream 没有注意到窗口已关闭
嗨@gnostrenoff 我和你有同样的问题。我正在测试 1 分钟的时间窗口。 (.windowedBy(TimeWindows.of(Duration.ofMinutes(1)).grace(Duration.ZERO)))，我将一条记录通过管道传输到拓扑，休眠 2 分钟并发送另一条记录。但我仍然得到null 结果。你能解释一下你是如何修复你的测试的吗？非常感谢！
@thinktwice 是关于 event-time 而不是 wall-clock time -- 您需要在生成数据时设置相应的记录时间戳。
感谢@MatthiasJ.Sax！一旦我设置了正确的活动时间，它就对我有用！我注意到suppress() 的另一个问题是，我认为它应该只填充每个窗口的最后一条记录？但是从单元测试来看，它会发送所有更新。是预期的吗？

标签： java unit-testing apache-kafka-streams windowing

【解决方案1】：

SessionWindows 使用 event-time 而不是 wall-clock 。尝试正确设置记录的事件时间以模拟不活动间隙。比如：

testDriver.pipeInput(factory.create(INPUT_TOPIC, key1, record1, eventTimeMs));
testDriver.pipeInput(factory.create(INPUT_TOPIC, key2, record2, eventTimeMs + inactivityGapMs));

但首先，您需要一个自定义的TimestampExtractor，例如：

 public static class RecordTimestampExtractor implements TimestampExtractor {

    @Override
    public long extract(ConsumerRecord<Object, Object> record, long previousTimestamp) {
      return record.timestamp();
    }
  }

必须像这样注册：

 streamProperties.setProperty(
        StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
        RecordTimestampExtractor.class.getName()
    );

【讨论】：