WorkerPipelineOptionsFactory { public static DataflowWorkerHarnessOptions createFromSystemProperties() throws IOException { ObjectMapper objectMapper = new ObjectMapper(); DataflowWorkerHarnessOptions options; if (System.getProperties().containsKey("sdk_pipeline_options")) { String serializedOptions = System.getProperty("sdk_pipeline_options"); LOG.info("Worker harness starting with: {}", serializedOptions); options = objectMapper .readValue(serializedOptions, PipelineOptions.class) .as(DataflowWorkerHarnessOptions.class); } 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(DataflowWorkerHarnessOptions.class); } else { LOG.info("Using empty PipelineOptions, as none were provided."); options = PipelineOptionsFactory.as(DataflowWorkerHarnessOptions.class); } if (System.getProperties().containsKey("worker_id")) { options.setWorkerId(System.getProperty("worker_id")); } if (System.getProperties().containsKey("job_id")) { options.setJobId(System.getProperty("job_id")); } return options; }  static DataflowWorkerHarnessOptions createFromSystemProperties();  }
@Test public void testCreationFromSystemProperties() throws Exception { System.getProperties() .putAll( ImmutableMap.<String, String>builder() .put("worker_id", "test_worker_id") .put("job_id", "test_job_id") .put("sdk_pipeline_options", "{\"options\":{\"numWorkers\":999}}") .build()); @SuppressWarnings("deprecation") DataflowWorkerHarnessOptions options = WorkerPipelineOptionsFactory.createFromSystemProperties(); assertEquals("test_worker_id", options.getWorkerId()); assertEquals("test_job_id", options.getJobId()); assertEquals(999, options.getNumWorkers()); }
@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") .put("sdk_pipeline_options_file", file.getPath()) .build()); @SuppressWarnings("deprecation") DataflowWorkerHarnessOptions options = WorkerPipelineOptionsFactory.createFromSystemProperties(); assertEquals("test_worker_id_2", options.getWorkerId()); assertEquals("test_job_id_2", options.getJobId()); assertEquals(1000, options.getNumWorkers()); }