Testing Flink with embedded Kafka

Question

I have a simple Flink application, which sums up the events with the same id and timestamp within the last minute:

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

DataStream 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 {
        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);
        }
    }

I would like to implement the scenario:

Start embedded Kafka
Publish couple of messages to the topic
Consume the messages with Flink
Check the correctness of the output produced by Flink

Does Flink provides utilities to test the job with embedded Kafka? If yes, what is the recommended approach?

Thanks.

user_2000823 · Answer

There's a JUnit rule you can use to bring up an embedded Kafka -- see (see https://github.com/charithe/kafka-junit).

To have tests that terminate cleanly, try something like this:

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

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

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

    return false;
  }
}

Testing Flink with embedded Kafka

1 Answers