How performance can be tested in Kafka and Flink environment?

Question

How is performance tested for Flink with kafka as input source. Also, recommend if any performance test tools are available for this case.

Answer 1

Flink includes metrics for both throughput (numRecordsInPerSecond and numRecordsOutPerSecond) and latency.

If you want to more carefully measure latency end-to-end, you can add a custom metric in a sink (or other terminal node) that compares timestamps in your events to the current time. That would look something like this:

public class LatencyMeasuringSink<T> extends RichSinkFunction<T> {
  private transient DescriptiveStatisticsHistogram eventTimeLag;
  private static final int EVENT_TIME_LAG_WINDOW_SIZE = 10_000;

  @Override
  public void open(Configuration parameters) throws Exception {
    super.open(parameters);

    eventTimeLag = getRuntimeContext().getMetricGroup().histogram("eventTimeLag",
            new DescriptiveStatisticsHistogram(EVENT_TIME_LAG_WINDOW_SIZE));
  }

  @Override
  public void invoke(T dataPoint, Context context) throws Exception {
    eventTimeLag.update(System.currentTimeMillis() - dataPoint.getTimeStampMs());
  }
}

You might want to configure your Kafka producer to put LogAppendTime timestamps in your events, and use those as the basis for comparison. This assumes, of course, that the clocks in the different machines involved are synchronized well enough for this measurement to be meaningful -- or you might run the tests on a single machine.

FLIP-83: Flink End-to-end Performance Testing Framework may also be of interest. This is work in progress.