使用嵌入式 Kafka 测试 Flink

Question

我有一个简单的 Flink 应用，它总结了最后一分钟内具有相同 id 和时间戳的事件：

DataStream<String> input = env
                .addSource(consumerProps)
                .uid("app");

DataStream<Event> events = input.map(record -> mapper.readValue(record, Event.class));

pixels
        .assignTimestampsAndWatermarks(new TimestampsAndWatermarks())
        .keyBy("id")
        .timeWindow(Time.minutes(1))
        .sum("constant")
        .addSink(simpleNotificationServiceSink);


env.execute(jobName);


private static class TimestampsAndWatermarks extends BoundedOutOfOrdernessTimestampExtractor<Pixel> {
        public TimestampsAndWatermarks() {
            super(Time.seconds(90));
        }

        // timestampReadable is timestamp rounded on minutes, in format yyyyMMddhhmm
        @Override
        public long extractTimestamp(Pixel pixel) {
            return Long.parseLong(pixel.timestampReadable);
        }
    }

我想实现这个场景：

1. 启动嵌入式 Kafka

2. 向主题发布几条消息

3. 使用 Flink 消费消息

4. 检查 Flink 产生的输出的正确性

Flink 是否提供实用程序来测试嵌入式 Kafka 的作业？如果是，推荐的方法是什么？

谢谢。

Answer 1

您可以使用一条 JUnit 规则来调出嵌入式 Kafka——请参阅（请参阅https://github.com/charithe/kafka-junit）。

要让测试完全终止，请尝试以下操作：

public class TestDeserializer extends YourKafkaDeserializer<T> {
  public final static String END_APP_MARKER = "END_APP_MARKER"; // tests send as last record

  @Override
  public boolean isEndOfStream(ParseResult<T> nextElement) {
    if (nextElement.getParseError() == null)
      return false;

    if (END_APP_MARKER.equals(nextElement.getParseError().getRawData()))
      return true;

    return false;
  }
}