class 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);

}