public static <T extends DataflowWorkerHarnessOptions> T createFromSystemProperties(
      Class<T> harnessOptionsClass) throws IOException {
    ObjectMapper objectMapper = new ObjectMapper();
    T options;
    if (System.getProperties().containsKey("sdk_pipeline_options")) {
      // TODO: remove this method of getting pipeline options, once migration is complete.
      String serializedOptions = System.getProperty("sdk_pipeline_options");
      LOG.info("Worker harness starting with: {}", serializedOptions);
      options =
          objectMapper.readValue(serializedOptions, PipelineOptions.class).as(harnessOptionsClass);
    } else if (System.getProperties().containsKey("sdk_pipeline_options_file")) {
      String filePath = System.getProperty("sdk_pipeline_options_file");
      LOG.info("Loading pipeline options from " + filePath);
      String serializedOptions =
          new String(Files.readAllBytes(Paths.get(filePath)), StandardCharsets.UTF_8);
      LOG.info("Worker harness starting with: " + serializedOptions);
      options =
          objectMapper.readValue(serializedOptions, PipelineOptions.class).as(harnessOptionsClass);
    } else {
      LOG.info("Using empty PipelineOptions, as none were provided.");
      options = PipelineOptionsFactory.as(harnessOptionsClass);
    }

    // These values will not be known at job submission time and must be provided.
    if (System.getProperties().containsKey("worker_id")) {
      options.setWorkerId(System.getProperty("worker_id"));
    }
    if (System.getProperties().containsKey("job_id")) {
      options.setJobId(System.getProperty("job_id"));
    }
    if (System.getProperties().containsKey("worker_pool")) {
      options.setWorkerPool(System.getProperty("worker_pool"));
    }

    // Remove impersonate information from workers
    // More details:
    // https://cloud.google.com/dataflow/docs/reference/pipeline-options#security_and_networking
    if (options.getImpersonateServiceAccount() != null) {
      LOG.info(
          "Remove the impersonateServiceAccount pipeline option ({}) when starting the Worker harness.",
          options.getImpersonateServiceAccount());
      options.setImpersonateServiceAccount(null);
    }

    return options;
  }
@Test
  public void testCreationWithPipelineOptionsFile() throws Exception {
    File file = tmpFolder.newFile();
    String jsonOptions = "{\"options\":{\"numWorkers\":1000}}";
    Files.write(Paths.get(file.getPath()), jsonOptions.getBytes(StandardCharsets.UTF_8));
    System.getProperties()
        .putAll(
            ImmutableMap.<String, String>builder()
                .put("worker_id", "test_worker_id_2")
                .put("job_id", "test_job_id_2")
                // Set a non-default value for testing
                .put("sdk_pipeline_options_file", file.getPath())
                .build());

    @SuppressWarnings("deprecation") // testing deprecated functionality
    DataflowWorkerHarnessOptions options =
        WorkerPipelineOptionsFactory.createFromSystemProperties(DataflowWorkerHarnessOptions.class);
    assertEquals("test_worker_id_2", options.getWorkerId());
    assertEquals("test_job_id_2", options.getJobId());
    assertEquals(1000, options.getNumWorkers());
  }