@Override
  public Job drainJob(String project, String region, String jobId) {
    LOG.info("Draining {} under {}", jobId, project);
    Job job = new Job().setRequestedState(JobState.DRAINED.toString());
    LOG.info("Sending job to update {}:\n{}", jobId, formatForLogging(job));
    return Failsafe.with(clientRetryPolicy())
        .get(
            () ->
                client.projects().locations().jobs().update(project, region, jobId, job).execute());
  }
@Test
  public void testDrainJobThrowsException() throws IOException {
    when(getLocationJobs(client).update(any(), any(), any(), any())).thenThrow(new IOException());
    assertThrows(
        FailsafeException.class,
        () -> new FakePipelineLauncher(client).drainJob(PROJECT, REGION, JOB_ID));
  }