private JobMetrics getJobMetrics() throws IOException {
    if (cachedMetricResults != null) {
      // Metric results have been cached after the job ran.
      return cachedMetricResults;
    }
    JobMetrics result = dataflowClient.getJobMetrics(dataflowPipelineJob.getJobId());
    if (dataflowPipelineJob.getState().isTerminal()) {
      // Add current query result to the cache.
      cachedMetricResults = result;
    }
    return result;
  }
@Test
  public void testSingleCounterUpdates() throws IOException {
    AppliedPTransform<?, ?, ?> myStep = mock(AppliedPTransform.class);
    when(myStep.getFullName()).thenReturn("myStepName");
    BiMap<AppliedPTransform<?, ?, ?>, String> transformStepNames = HashBiMap.create();
    transformStepNames.put(myStep, "s2");

    JobMetrics jobMetrics = new JobMetrics();
    DataflowPipelineJob job = mock(DataflowPipelineJob.class);
    DataflowPipelineOptions options = mock(DataflowPipelineOptions.class);
    when(options.isStreaming()).thenReturn(false);
    when(job.getDataflowOptions()).thenReturn(options);
    when(job.getState()).thenReturn(State.RUNNING);
    when(job.getJobId()).thenReturn(JOB_ID);
    when(job.getTransformStepNames()).thenReturn(transformStepNames);

    MetricUpdate update = new MetricUpdate();
    long stepValue = 1234L;
    update.setScalar(new BigDecimal(stepValue));

    // The parser relies on the fact that one tentative and one committed metric update exist in
    // the job metrics results.
    MetricUpdate mu1 =
        makeCounterMetricUpdate("counterName", "counterNamespace", "s2", 1234L, false);
    MetricUpdate mu1Tentative =
        makeCounterMetricUpdate("counterName", "counterNamespace", "s2", 1233L, true);
    jobMetrics.setMetrics(ImmutableList.of(mu1, mu1Tentative));
    DataflowClient dataflowClient = mock(DataflowClient.class);
    when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics);

    DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
    MetricQueryResults result = dataflowMetrics.allMetrics();
    assertThat(
        result.getCounters(),
        containsInAnyOrder(
            attemptedMetricsResult("counterNamespace", "counterName", "myStepName", 1234L)));
    assertThat(
        result.getCounters(),
        containsInAnyOrder(
            committedMetricsResult("counterNamespace", "counterName", "myStepName", 1234L)));
  }