DataflowRunner extends PipelineRunner<DataflowPipelineJob> { @VisibleForTesting static void validateWorkerSettings(GcpOptions gcpOptions) { Preconditions.checkArgument( gcpOptions.getZone() == null || gcpOptions.getWorkerRegion() == null, "Cannot use option zone with workerRegion. Prefer either workerZone or workerRegion."); Preconditions.checkArgument( gcpOptions.getZone() == null || gcpOptions.getWorkerZone() == null, "Cannot use option zone with workerZone. Prefer workerZone."); Preconditions.checkArgument( gcpOptions.getWorkerRegion() == null || gcpOptions.getWorkerZone() == null, "workerRegion and workerZone options are mutually exclusive."); DataflowPipelineOptions dataflowOptions = gcpOptions.as(DataflowPipelineOptions.class); boolean hasExperimentWorkerRegion = false; if (dataflowOptions.getExperiments() != null) { for (String experiment : dataflowOptions.getExperiments()) { if (experiment.startsWith("worker_region")) { hasExperimentWorkerRegion = true; break; } } } Preconditions.checkArgument( !hasExperimentWorkerRegion || gcpOptions.getWorkerRegion() == null, "Experiment worker_region and option workerRegion are mutually exclusive."); Preconditions.checkArgument( !hasExperimentWorkerRegion || gcpOptions.getWorkerZone() == null, "Experiment worker_region and option workerZone are mutually exclusive."); if (gcpOptions.getZone() != null) { LOG.warn("Option --zone is deprecated. Please use --workerZone instead."); gcpOptions.setWorkerZone(gcpOptions.getZone()); gcpOptions.setZone(null); } } @VisibleForTesting protected  DataflowRunner(DataflowPipelineOptions options); static DataflowRunner fromOptions(PipelineOptions options); @Override DataflowPipelineJob run(Pipeline pipeline); static boolean hasExperiment(DataflowPipelineDebugOptions options, String experiment); DataflowPipelineTranslator getTranslator(); @Experimental void setHooks(DataflowRunnerHooks hooks); @Override String toString(); static final String PROJECT_ID_REGEXP; }
@Test public void testZoneAndWorkerRegionMutuallyExclusive() { GcpOptions options = PipelineOptionsFactory.as(GcpOptions.class); options.setZone("us-east1-b"); options.setWorkerRegion("us-east1"); assertThrows( IllegalArgumentException.class, () -> DataflowRunner.validateWorkerSettings(options)); }
@Test public void testZoneAndWorkerZoneMutuallyExclusive() { GcpOptions options = PipelineOptionsFactory.as(GcpOptions.class); options.setZone("us-east1-b"); options.setWorkerZone("us-east1-c"); assertThrows( IllegalArgumentException.class, () -> DataflowRunner.validateWorkerSettings(options)); }
@Test public void testExperimentRegionAndWorkerRegionMutuallyExclusive() { GcpOptions options = PipelineOptionsFactory.as(GcpOptions.class); DataflowPipelineOptions dataflowOptions = options.as(DataflowPipelineOptions.class); ExperimentalOptions.addExperiment(dataflowOptions, "worker_region=us-west1"); options.setWorkerRegion("us-east1"); assertThrows( IllegalArgumentException.class, () -> DataflowRunner.validateWorkerSettings(options)); }
@Test public void testExperimentRegionAndWorkerZoneMutuallyExclusive() { GcpOptions options = PipelineOptionsFactory.as(GcpOptions.class); DataflowPipelineOptions dataflowOptions = options.as(DataflowPipelineOptions.class); ExperimentalOptions.addExperiment(dataflowOptions, "worker_region=us-west1"); options.setWorkerZone("us-east1-b"); assertThrows( IllegalArgumentException.class, () -> DataflowRunner.validateWorkerSettings(options)); }
@Test public void testWorkerRegionAndWorkerZoneMutuallyExclusive() { GcpOptions options = PipelineOptionsFactory.as(GcpOptions.class); options.setWorkerRegion("us-east1"); options.setWorkerZone("us-east1-b"); assertThrows( IllegalArgumentException.class, () -> DataflowRunner.validateWorkerSettings(options)); }
@Test public void testZoneAliasWorkerZone() { GcpOptions options = PipelineOptionsFactory.as(GcpOptions.class); options.setZone("us-east1-b"); DataflowRunner.validateWorkerSettings(options); assertNull(options.getZone()); assertEquals("us-east1-b", options.getWorkerZone()); }