MetricsPusher implements Serializable { private void run() { pushMetrics(); if (pipelineResult != null) { PipelineResult.State pipelineState = pipelineResult.getState(); if (pipelineState.isTerminal()) { tearDown(); } } } MetricsPusher(
      MetricsContainerStepMap metricsContainerStepMap,
      MetricsOptions pipelineOptions,
      PipelineResult pipelineResult); void start();  }
@Category({ValidatesRunner.class, UsesAttemptedMetrics.class, UsesCounterMetrics.class}) @Test public void pushesUserMetrics() throws Exception { TestMetricsSink.clear(); pipeline .apply( GenerateSequence.from(0).to(NUM_ELEMENTS).withMaxReadTime(Duration.standardDays(1))) .apply(ParDo.of(new CountingDoFn())); pipeline.run(); Thread.sleep( (pipeline.getOptions().as(MetricsOptions.class).getMetricsPushPeriod() + 1L) * 1000); assertThat(TestMetricsSink.getCounterValue(COUNTER_NAME), is(NUM_ELEMENTS)); }
@Category({ ValidatesRunner.class, UsesAttemptedMetrics.class, UsesCounterMetrics.class, UsesSystemMetrics.class }) @Test public void pushesSystemMetrics() throws InterruptedException { TestMetricsSink.clear(); pipeline .apply( GenerateSequence.from(0).to(NUM_ELEMENTS).withMaxReadTime(Duration.standardDays(1))) .apply(ParDo.of(new CountingDoFn())); pipeline.run(); Thread.sleep( (pipeline.getOptions().as(MetricsOptions.class).getMetricsPushPeriod() + 1L) * 1000); assertThat(TestMetricsSink.getSystemCounters().isEmpty(), is(false)); }