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 testMultipleCounterUpdatesStreaming() 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(true);
    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)));

    DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
    MetricQueryResults result = dataflowMetrics.allMetrics();
    try {
      result.getCounters().iterator().next().getCommitted();
      fail("Expected UnsupportedOperationException");
    } catch (UnsupportedOperationException expected) {
      assertThat(
          expected.getMessage(),
          containsString(
              "This runner does not currently support committed"
                  + " metrics results. Please use 'attempted' instead."));
    }
    assertThat(
        result.getCounters(),
        containsInAnyOrder(
            attemptedMetricsResult("counterNamespace", "counterName", "myStepName", 1233L),
            attemptedMetricsResult("otherNamespace", "otherCounter", "myStepName3", 12L),
            attemptedMetricsResult("otherNamespace", "counterName", "myStepName4", 1200L)));
  }