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 testMultipleCounterUpdates() throws IOException {
    AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class);
    when(myStep2.getFullName()).thenReturn("myStepName");
    BiMap<AppliedPTransform<?, ?, ?>, String> transformStepNames = HashBiMap.create();
    transformStepNames.put(myStep2, "s2");
    AppliedPTransform<?, ?, ?> myStep3 = mock(AppliedPTransform.class);
    when(myStep3.getFullName()).thenReturn("myStepName3");
    transformStepNames.put(myStep3, "s3");
    AppliedPTransform<?, ?, ?> myStep4 = mock(AppliedPTransform.class);
    when(myStep4.getFullName()).thenReturn("myStepName4");
    transformStepNames.put(myStep4, "s4");

    JobMetrics jobMetrics = new JobMetrics();
    DataflowClient dataflowClient = mock(DataflowClient.class);
    when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(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);

    // The parser relies on the fact that one tentative and one committed metric update exist in
    // the job metrics results.
    jobMetrics.setMetrics(
        ImmutableList.of(
            makeCounterMetricUpdate("counterName", "counterNamespace", "s2", 1233L, false),
            makeCounterMetricUpdate("counterName", "counterNamespace", "s2", 1234L, true),
            makeCounterMetricUpdate("otherCounter", "otherNamespace", "s3", 12L, false),
            makeCounterMetricUpdate("otherCounter", "otherNamespace", "s3", 12L, true),
            makeCounterMetricUpdate("counterName", "otherNamespace", "s4", 1200L, false),
            makeCounterMetricUpdate("counterName", "otherNamespace", "s4", 1233L, true),
            // The following counter can not have its name translated thus it won't appear.
            makeCounterMetricUpdate("lostName", "otherNamespace", "s5", 1200L, false),
            makeCounterMetricUpdate("lostName", "otherNamespace", "s5", 1200L, true)));

    DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
    MetricQueryResults result = dataflowMetrics.allMetrics();
    assertThat(
        result.getCounters(),
        containsInAnyOrder(
            attemptedMetricsResult("counterNamespace", "counterName", "myStepName", 1233L),
            attemptedMetricsResult("otherNamespace", "otherCounter", "myStepName3", 12L),
            attemptedMetricsResult("otherNamespace", "counterName", "myStepName4", 1200L)));
    assertThat(
        result.getCounters(),
        containsInAnyOrder(
            committedMetricsResult("counterNamespace", "counterName", "myStepName", 1233L),
            committedMetricsResult("otherNamespace", "otherCounter", "myStepName3", 12L),
            committedMetricsResult("otherNamespace", "counterName", "myStepName4", 1200L)));
  }