DataflowJobManager implements JobManager { @Override public List<Job> listRunningJobs() { List<com.google.api.services.dataflow.model.Job> jobs; try { jobs = dataflow .projects() .locations() .jobs() .list(projectId, location) .setFilter("ACTIVE") .execute() .getJobs(); } catch (IOException e) { throw new RuntimeException( String.format("Unable to retrieve list of jobs from dataflow: %s", e.getMessage())); } if (jobs == null) { return Collections.emptyList(); } return jobs.stream() .map( dfJob -> { try { return dataflow .projects() .locations() .jobs() .get(projectId, location, dfJob.getId()) .setView("JOB_VIEW_ALL") .execute(); } catch (IOException e) { log.error( "Job's detailed info {} couldn't be loaded from Dataflow: {}", dfJob.getId(), e.getMessage()); return null; } }) .filter( dfJob -> dfJob != null && (dfJob.getLabels() != null || this.jobSelector.isEmpty()) && this.jobSelector.entrySet().stream() .allMatch( entry -> dfJob .getLabels() .getOrDefault(entry.getKey(), "") .equals(entry.getValue()))) .map( dfJob -> { Map<String, Object> options = (Map<String, Object>) dfJob.getEnvironment().getSdkPipelineOptions().get("options"); List<StoreProto.Store> stores = parseStoreJsonList((List<String>) options.get("storesJson")); SourceProto.Source source = parseSourceJson((String) options.get("sourceJson")); Job job = Job.builder() .setId((String) options.get("jobName")) .setSource(source) .setStores( stores.stream() .collect(Collectors.toMap(StoreProto.Store::getName, s -> s))) .setLabels( dfJob.getLabels() == null ? new HashMap<>() : new HashMap<>(dfJob.getLabels())) .build(); job.setExtId(dfJob.getId()); job.setStatus(JobStatus.RUNNING); if (dfJob.getCreateTime() != null) { job.setCreated(DateTime.parse(dfJob.getCreateTime()).toDate()); } return job; }) .collect(Collectors.toList()); } DataflowJobManager(
      DataflowRunnerConfigOptions runnerConfigOptions,
      MetricsProperties metricsProperties,
      IngestionJobProto.SpecsStreamingUpdateConfig specsStreamingUpdateConfig,
      Map<String, String> jobSelector,
      Dataflow dataflow); static DataflowJobManager of(
      DataflowRunnerConfigOptions runnerConfigOptions,
      MetricsProperties metricsProperties,
      IngestionJobProto.SpecsStreamingUpdateConfig specsStreamingUpdateConfig,
      Map<String, String> jobSelector); @Override Runner getRunnerType(); @Override Job startJob(Job job); @Override Job updateJob(Job job); @Override Job abortJob(Job job); @Override Job restartJob(Job job); @Override JobStatus getJobStatus(Job job); @Override List<Job> listRunningJobs(); DataflowPipelineJob runPipeline(ImportOptions pipelineOptions);  }
@Test @SneakyThrows public void shouldHandleNullResponseFromDataflow() { when(dataflow .projects() .locations() .jobs() .list("project", "region") .setFilter("ACTIVE") .execute() .getJobs()) .thenReturn(null); assertThat(dfJobManager.listRunningJobs(), hasSize(0)); }