@Override
  public @Nullable State waitUntilFinish() {
    return waitUntilFinish(Duration.millis(-1));
  }
@Test
  public void testWaitUntilFinishNoRepeatedLogs() throws Exception {
    DataflowPipelineJob job = new DataflowPipelineJob(mockDataflowClient, JOB_ID, options, null);
    Sleeper sleeper = new ZeroSleeper();
    NanoClock nanoClock = mock(NanoClock.class);

    Instant separatingTimestamp = new Instant(42L);
    JobMessage theMessage = infoMessage(separatingTimestamp, "nothing");

    MonitoringUtil mockMonitor = mock(MonitoringUtil.class);
    when(mockMonitor.getJobMessages(anyString(), anyLong()))
        .thenReturn(ImmutableList.of(theMessage));

    // The Job just always reports "running" across all calls
    Job fakeJob = new Job();
    fakeJob.setCurrentState("JOB_STATE_RUNNING");
    when(mockDataflowClient.getJob(anyString())).thenReturn(fakeJob);

    // After waitUntilFinish the DataflowPipelineJob should record the latest message timestamp
    when(nanoClock.nanoTime()).thenReturn(0L).thenReturn(2000000000L);
    job.waitUntilFinish(Duration.standardSeconds(1), mockHandler, sleeper, nanoClock, mockMonitor);
    verify(mockHandler).process(ImmutableList.of(theMessage));

    // Second waitUntilFinish should request jobs with `separatingTimestamp` so the monitor
    // will only return new messages
    when(nanoClock.nanoTime()).thenReturn(3000000000L).thenReturn(6000000000L);
    job.waitUntilFinish(Duration.standardSeconds(1), mockHandler, sleeper, nanoClock, mockMonitor);
    verify(mockMonitor).getJobMessages(anyString(), eq(separatingTimestamp.getMillis()));
  }