@Override
  public @Nullable State waitUntilFinish() {
    return waitUntilFinish(Duration.millis(-1));
  }
@Test
  public void testWaitToFinishMessagesFail() throws Exception {
    Dataflow.Projects.Locations.Jobs.Get statusRequest =
        mock(Dataflow.Projects.Locations.Jobs.Get.class);

    Job statusResponse = new Job();
    statusResponse.setCurrentState("JOB_STATE_" + State.DONE.name());
    when(mockJobs.get(eq(PROJECT_ID), eq(REGION_ID), eq(JOB_ID))).thenReturn(statusRequest);
    when(statusRequest.execute()).thenReturn(statusResponse);

    MonitoringUtil.JobMessagesHandler jobHandler = mock(MonitoringUtil.JobMessagesHandler.class);
    Dataflow.Projects.Locations.Jobs.Messages mockMessages =
        mock(Dataflow.Projects.Locations.Jobs.Messages.class);
    Messages.List listRequest = mock(Dataflow.Projects.Locations.Jobs.Messages.List.class);
    when(mockJobs.messages()).thenReturn(mockMessages);
    when(mockMessages.list(eq(PROJECT_ID), eq(REGION_ID), eq(JOB_ID))).thenReturn(listRequest);
    when(listRequest.setPageToken(eq((String) null))).thenReturn(listRequest);
    when(listRequest.execute()).thenThrow(SocketTimeoutException.class);

    DataflowPipelineJob job =
        new DataflowPipelineJob(DataflowClient.create(options), JOB_ID, options, ImmutableMap.of());

    State state =
        job.waitUntilFinish(
            Duration.standardMinutes(5), jobHandler, fastClock::sleep, fastClock::nanoTime);
    assertEquals(null, state);
  }