@Test
public void shouldSendCorrectStatsDMetrics() throws IOException, InterruptedException {

  PipelineOptions pipelineOptions = PipelineOptionsFactory.create();
  pipelineOptions.setJobName("job-12345678");
  Map<String, Iterable<FeatureRow>> input =
      readTestInput("feast/ingestion/transform/WriteFeatureValueMetricsDoFnTest.input");
  List<String> expectedLines =
      readTestOutput("feast/ingestion/transform/WriteFeatureValueMetricsDoFnTest.output");
  pipeline
      .apply(Create.of(input))
      .apply(
          ParDo.of(
              WriteFeatureValueMetricsDoFn.newBuilder()
                  .setStatsdHost("localhost")
                  .setStatsdPort(STATSD_SERVER_PORT)
                  .setStoreName("store")
                  .setMetricsNamespace("test")
                  .build()));
  pipeline.run(pipelineOptions).waitUntilFinish();
  // Wait until StatsD has finished processed all messages, 3 sec is a reasonable duration
  // based on empirical testing.
  Thread.sleep(3000);
  List<String> actualLines = statsDServer.messagesReceived();
  for (String expected : expectedLines) {
    boolean matched = false;
    for (String actual : actualLines) {
      if (actual.equals(expected)) {
        matched = true;
        break;
      }
    }
    if (!matched) {
      System.out.println("Print actual metrics output for debugging:");
      for (String line : actualLines) {
        System.out.println(line);
      }
      fail(String.format("Expected StatsD metric not found:\n%s", expected));
    }
  }
  statsDServer.stop();
}