DataflowRunner extends PipelineRunner<DataflowPipelineJob> { public static DataflowRunner fromOptions(PipelineOptions options) { DataflowPipelineOptions dataflowOptions = PipelineOptionsValidator.validate(DataflowPipelineOptions.class, options); ArrayList<String> missing = new ArrayList<>(); if (dataflowOptions.getAppName() == null) { missing.add("appName"); } if (Strings.isNullOrEmpty(dataflowOptions.getRegion()) && isServiceEndpoint(dataflowOptions.getDataflowEndpoint())) { missing.add("region"); } if (missing.size() > 0) { throw new IllegalArgumentException( "Missing required pipeline options: " + Joiner.on(',').join(missing)); } validateWorkerSettings(PipelineOptionsValidator.validate(GcpOptions.class, options)); PathValidator validator = dataflowOptions.getPathValidator(); String gcpTempLocation; try { gcpTempLocation = dataflowOptions.getGcpTempLocation(); } catch (Exception e) { throw new IllegalArgumentException( "DataflowRunner requires gcpTempLocation, " + "but failed to retrieve a value from PipelineOptions", e); } validator.validateOutputFilePrefixSupported(gcpTempLocation); String stagingLocation; try { stagingLocation = dataflowOptions.getStagingLocation(); } catch (Exception e) { throw new IllegalArgumentException( "DataflowRunner requires stagingLocation, " + "but failed to retrieve a value from PipelineOptions", e); } validator.validateOutputFilePrefixSupported(stagingLocation); if (!isNullOrEmpty(dataflowOptions.getSaveProfilesToGcs())) { validator.validateOutputFilePrefixSupported(dataflowOptions.getSaveProfilesToGcs()); } if (dataflowOptions.getFilesToStage() != null) { dataflowOptions.getFilesToStage().stream() .forEach( stagedFileSpec -> { File localFile; if (stagedFileSpec.contains("=")) { String[] components = stagedFileSpec.split("=", 2); localFile = new File(components[1]); } else { localFile = new File(stagedFileSpec); } if (!localFile.exists()) { throw new RuntimeException( String.format("Non-existent files specified in filesToStage: %s", localFile)); } }); } else { dataflowOptions.setFilesToStage( detectClassPathResourcesToStage(DataflowRunner.class.getClassLoader(), options)); if (dataflowOptions.getFilesToStage().isEmpty()) { throw new IllegalArgumentException("No files to stage has been found."); } else { LOG.info( "PipelineOptions.filesToStage was not specified. " + "Defaulting to files from the classpath: will stage {} files. " + "Enable logging at DEBUG level to see which files will be staged.", dataflowOptions.getFilesToStage().size()); LOG.debug("Classpath elements: {}", dataflowOptions.getFilesToStage()); } } String jobName = dataflowOptions.getJobName().toLowerCase(); checkArgument( jobName.matches("[a-z]([-a-z0-9]*[a-z0-9])?"), "JobName invalid; the name must consist of only the characters " + "[-a-z0-9], starting with a letter and ending with a letter " + "or number"); if (!jobName.equals(dataflowOptions.getJobName())) { LOG.info( "PipelineOptions.jobName did not match the service requirements. " + "Using {} instead of {}.", jobName, dataflowOptions.getJobName()); } dataflowOptions.setJobName(jobName); String project = dataflowOptions.getProject(); if (project.matches("[0-9]*")) { throw new IllegalArgumentException( "Project ID '" + project + "' invalid. Please make sure you specified the Project ID, not project number."); } else if (!project.matches(PROJECT_ID_REGEXP)) { throw new IllegalArgumentException( "Project ID '" + project + "' invalid. Please make sure you specified the Project ID, not project description."); } DataflowPipelineDebugOptions debugOptions = dataflowOptions.as(DataflowPipelineDebugOptions.class); if (debugOptions.getNumberOfWorkerHarnessThreads() < 0) { throw new IllegalArgumentException( "Number of worker harness threads '" + debugOptions.getNumberOfWorkerHarnessThreads() + "' invalid. Please make sure the value is non-negative."); } if (dataflowOptions.isStreaming() && dataflowOptions.getGcsUploadBufferSizeBytes() == null) { dataflowOptions.setGcsUploadBufferSizeBytes(GCS_UPLOAD_BUFFER_SIZE_BYTES_DEFAULT); } String agentJavaVer = (Environments.getJavaVersion() == Environments.JavaVersion.v8) ? "(JRE 8 environment)" : "(JDK 11 environment)"; DataflowRunnerInfo dataflowRunnerInfo = DataflowRunnerInfo.getDataflowRunnerInfo(); String userAgent = String.format( "%s/%s%s", dataflowRunnerInfo.getName(), dataflowRunnerInfo.getVersion(), agentJavaVer) .replace(" ", "_"); dataflowOptions.setUserAgent(userAgent); return new DataflowRunner(dataflowOptions); } @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 testPathValidatorOverride() { String[] args = new String[] { "--runner=DataflowRunner", "--region=some-region-1", "--tempLocation=/tmp/testing", "--project=test-project", "--credentialFactoryClass=" + NoopCredentialFactory.class.getName(), "--pathValidatorClass=" + NoopPathValidator.class.getName(), }; TestPipeline.fromOptions(PipelineOptionsFactory.fromArgs(args).create()); }
@Test public void testFromOptionsWithUppercaseConvertsToLowercase() throws Exception { String mixedCase = "ThisJobNameHasMixedCase"; DataflowPipelineOptions options = buildPipelineOptions(); options.setJobName(mixedCase); DataflowRunner.fromOptions(options); assertThat(options.getJobName(), equalTo(mixedCase.toLowerCase())); }
@Test public void testFromOptionsUserAgentFromPipelineInfo() throws Exception { DataflowPipelineOptions options = buildPipelineOptions(); DataflowRunner.fromOptions(options); String expectedName = DataflowRunnerInfo.getDataflowRunnerInfo().getName().replace(" ", "_"); assertThat(options.getUserAgent(), containsString(expectedName)); String expectedVersion = DataflowRunnerInfo.getDataflowRunnerInfo().getVersion(); assertThat(options.getUserAgent(), containsString(expectedVersion)); }
@Test public void testRegionRequiredForServiceRunner() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setRegion(null); options.setDataflowEndpoint("https: assertThrows(IllegalArgumentException.class, () -> DataflowRunner.fromOptions(options)); }
@Test public void testRegionOptionalForNonServiceRunner() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setRegion(null); options.setDataflowEndpoint("http: DataflowRunner.fromOptions(options); }
@Test public void runWithDefaultFilesToStage() throws Exception { DataflowPipelineOptions options = buildPipelineOptions(); options.setFilesToStage(null); DataflowRunner.fromOptions(options); assertFalse(options.getFilesToStage().isEmpty()); }
@Test public void testGcsStagingLocationInitialization() throws Exception { DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class); options.setTempLocation(VALID_TEMP_BUCKET); options.setProject(PROJECT_ID); options.setRegion(REGION_ID); options.setGcpCredential(new TestCredential()); options.setGcsUtil(mockGcsUtil); options.setRunner(DataflowRunner.class); DataflowRunner.fromOptions(options); assertNotNull(options.getStagingLocation()); }
@Test public void testInvalidGcpTempLocation() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setGcpTempLocation("file: thrown.expect(IllegalArgumentException.class); thrown.expectMessage(containsString("Expected a valid 'gs: DataflowRunner.fromOptions(options); ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class); Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture()); assertValidJob(jobCaptor.getValue()); }
@Test public void testNonGcsTempLocation() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setTempLocation("file: thrown.expect(IllegalArgumentException.class); thrown.expectMessage( "DataflowRunner requires gcpTempLocation, " + "but failed to retrieve a value from PipelineOptions"); DataflowRunner.fromOptions(options); }
@Test public void testInvalidStagingLocation() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setStagingLocation("file: try { DataflowRunner.fromOptions(options); fail("fromOptions should have failed"); } catch (IllegalArgumentException e) { assertThat(e.getMessage(), containsString("Expected a valid 'gs: } options.setStagingLocation("my/staging/location"); try { DataflowRunner.fromOptions(options); fail("fromOptions should have failed"); } catch (IllegalArgumentException e) { assertThat(e.getMessage(), containsString("Expected a valid 'gs: } }
@Test public void testInvalidProfileLocation() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setSaveProfilesToGcs("file: try { DataflowRunner.fromOptions(options); fail("fromOptions should have failed"); } catch (IllegalArgumentException e) { assertThat(e.getMessage(), containsString("Expected a valid 'gs: } options.setSaveProfilesToGcs("my/staging/location"); try { DataflowRunner.fromOptions(options); fail("fromOptions should have failed"); } catch (IllegalArgumentException e) { assertThat(e.getMessage(), containsString("Expected a valid 'gs: } }
@Test public void testNonExistentTempLocation() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setGcpTempLocation(NON_EXISTENT_BUCKET); thrown.expect(IllegalArgumentException.class); thrown.expectMessage( containsString("Output path does not exist or is not writeable: " + NON_EXISTENT_BUCKET)); DataflowRunner.fromOptions(options); ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class); Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture()); assertValidJob(jobCaptor.getValue()); }
@Test public void testNonExistentStagingLocation() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setStagingLocation(NON_EXISTENT_BUCKET); thrown.expect(IllegalArgumentException.class); thrown.expectMessage( containsString("Output path does not exist or is not writeable: " + NON_EXISTENT_BUCKET)); DataflowRunner.fromOptions(options); ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class); Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture()); assertValidJob(jobCaptor.getValue()); }
@Test public void testNonExistentProfileLocation() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setSaveProfilesToGcs(NON_EXISTENT_BUCKET); thrown.expect(IllegalArgumentException.class); thrown.expectMessage( containsString("Output path does not exist or is not writeable: " + NON_EXISTENT_BUCKET)); DataflowRunner.fromOptions(options); ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class); Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture()); assertValidJob(jobCaptor.getValue()); }
@Test public void testNoProjectFails() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setProject(null); thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Project id"); thrown.expectMessage("when running a Dataflow in the cloud"); DataflowRunner.fromOptions(options); }
@Test public void testProjectId() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setProject("foo-12345"); DataflowRunner.fromOptions(options); }
@Test public void testProjectPrefix() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setProject("google.com:some-project-12345"); DataflowRunner.fromOptions(options); }
@Test public void testProjectNumber() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setProject("12345"); thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Project ID"); thrown.expectMessage("project number"); DataflowRunner.fromOptions(options); }
@Test public void testProjectDescription() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setProject("some project"); thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Project ID"); thrown.expectMessage("project description"); DataflowRunner.fromOptions(options); }
@Test public void testInvalidNumberOfWorkerHarnessThreads() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.as(DataflowPipelineDebugOptions.class).setNumberOfWorkerHarnessThreads(-1); thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Number of worker harness threads"); thrown.expectMessage("Please make sure the value is non-negative."); DataflowRunner.fromOptions(options); }
@Test public void testNoStagingLocationAndNoTempLocationFails() { DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class); options.setRunner(DataflowRunner.class); options.setProject("foo-project"); options.setRegion(REGION_ID); thrown.expect(IllegalArgumentException.class); thrown.expectMessage( "DataflowRunner requires gcpTempLocation, " + "but failed to retrieve a value from PipelineOption"); DataflowRunner.fromOptions(options); }
@Test public void testGcpTempAndNoTempLocationSucceeds() throws Exception { DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class); options.setRunner(DataflowRunner.class); options.setGcpCredential(new TestCredential()); options.setProject("foo-project"); options.setRegion(REGION_ID); options.setGcpTempLocation(VALID_TEMP_BUCKET); options.setGcsUtil(mockGcsUtil); DataflowRunner.fromOptions(options); }
@Test public void testTempLocationAndNoGcpTempLocationSucceeds() throws Exception { DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class); options.setRunner(DataflowRunner.class); options.setGcpCredential(new TestCredential()); options.setProject("foo-project"); options.setRegion(REGION_ID); options.setTempLocation(VALID_TEMP_BUCKET); options.setGcsUtil(mockGcsUtil); DataflowRunner.fromOptions(options); }
@Test public void testValidProfileLocation() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); options.setSaveProfilesToGcs(VALID_PROFILE_BUCKET); DataflowRunner.fromOptions(options); }
@Test public void testInvalidJobName() throws IOException { List<String> invalidNames = Arrays.asList("invalid_name", "0invalid", "invalid-"); List<String> expectedReason = Arrays.asList("JobName invalid", "JobName invalid", "JobName invalid"); for (int i = 0; i < invalidNames.size(); ++i) { DataflowPipelineOptions options = buildPipelineOptions(); options.setJobName(invalidNames.get(i)); try { DataflowRunner.fromOptions(options); fail("Expected IllegalArgumentException for jobName " + options.getJobName()); } catch (IllegalArgumentException e) { assertThat(e.getMessage(), containsString(expectedReason.get(i))); } } }
@Test public void testValidJobName() throws IOException { List<String> names = Arrays.asList("ok", "Ok", "A-Ok", "ok-123", "this-one-is-fairly-long-01234567890123456789"); for (String name : names) { DataflowPipelineOptions options = buildPipelineOptions(); options.setJobName(name); DataflowRunner runner = DataflowRunner.fromOptions(options); assertNotNull(runner); } }
@Test public void testTransformTranslatorMissing() throws IOException { DataflowPipelineOptions options = buildPipelineOptions(); Pipeline p = Pipeline.create(options); p.apply(Create.of(Arrays.asList(1, 2, 3))).apply(new TestTransform()); thrown.expect(IllegalStateException.class); thrown.expectMessage(containsString("no translator registered")); SdkComponents sdkComponents = SdkComponents.create(options); RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p, sdkComponents, true); DataflowPipelineTranslator.fromOptions(options) .translate( p, pipelineProto, sdkComponents, DataflowRunner.fromOptions(options), Collections.emptyList()); ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class); Mockito.verify(mockJobs).create(eq(PROJECT_ID), eq(REGION_ID), jobCaptor.capture()); assertValidJob(jobCaptor.getValue()); }