WriteFeatureValueMetricsDoFn extends DoFn<KV<String, Iterable<FeatureRow>>, Void> { static Builder newBuilder() { return new AutoValue_WriteFeatureValueMetricsDoFn.Builder(); }  @Setup void setup(); @Teardown void tearDown(); @ProcessElement void processElement(
      ProcessContext context,
      @Element KV<String, Iterable<FeatureRow>> featureSetRefToFeatureRows); static String GAUGE_NAME_FEATURE_VALUE_MIN; static String GAUGE_NAME_FEATURE_VALUE_MAX; static String GAUGE_NAME_FEATURE_VALUE_MEAN; static String GAUGE_NAME_FEATURE_VALUE_PERCENTILE_25; static String GAUGE_NAME_FEATURE_VALUE_PERCENTILE_50; static String GAUGE_NAME_FEATURE_VALUE_PERCENTILE_90; static String GAUGE_NAME_FEATURE_VALUE_PERCENTILE_95; static String GAUGE_NAME_FEATURE_VALUE_PERCENTILE_99; }
@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(); 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(); }