private void run() {
    pushMetrics();
    if (pipelineResult != null) {
      PipelineResult.State pipelineState = pipelineResult.getState();
      if (pipelineState.isTerminal()) {
        tearDown();
      }
    }
  }
@Category({ValidatesRunner.class, UsesAttemptedMetrics.class, UsesCounterMetrics.class})
  @Test
  public void pushesUserMetrics() throws Exception {
    TestMetricsSink.clear();
    pipeline
        .apply(
            // Use maxReadTime to force unbounded mode.
            GenerateSequence.from(0).to(NUM_ELEMENTS).withMaxReadTime(Duration.standardDays(1)))
        .apply(ParDo.of(new CountingDoFn()));
    pipeline.run();
    // give metrics pusher time to push
    Thread.sleep(
        (pipeline.getOptions().as(MetricsOptions.class).getMetricsPushPeriod() + 1L) * 1000);
    assertThat(TestMetricsSink.getCounterValue(COUNTER_NAME), is(NUM_ELEMENTS));
  }