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 testCachingMetricUpdates() throws IOException {
    Job modelJob = new Job();
    modelJob.setCurrentState(State.RUNNING.toString());

    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.DONE);
    when(job.getJobId()).thenReturn(JOB_ID);

    JobMetrics jobMetrics = new JobMetrics();
    jobMetrics.setMetrics(ImmutableList.of());
    DataflowClient dataflowClient = mock(DataflowClient.class);
    when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics);

    DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
    verify(dataflowClient, times(0)).getJobMetrics(JOB_ID);
    dataflowMetrics.allMetrics();
    verify(dataflowClient, times(1)).getJobMetrics(JOB_ID);
    dataflowMetrics.allMetrics();
    verify(dataflowClient, times(1)).getJobMetrics(JOB_ID);
  }