[BEAM-10435] Add ValidatesRunner task for local_job_service and Java SDK harness #11792
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,125 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package org.apache.beam.runners.portability.testing; | ||
|
|
||
| import static org.hamcrest.MatcherAssert.assertThat; | ||
|
|
||
| import com.google.auto.service.AutoService; | ||
| import java.io.IOException; | ||
| import java.nio.file.Files; | ||
| import java.nio.file.Paths; | ||
| import org.apache.beam.runners.portability.PortableRunner; | ||
| import org.apache.beam.sdk.Pipeline; | ||
| import org.apache.beam.sdk.PipelineResult; | ||
| import org.apache.beam.sdk.PipelineRunner; | ||
| import org.apache.beam.sdk.options.Description; | ||
| import org.apache.beam.sdk.options.PipelineOptions; | ||
| import org.apache.beam.sdk.options.PipelineOptionsRegistrar; | ||
| import org.apache.beam.sdk.options.PortablePipelineOptions; | ||
| import org.apache.beam.sdk.runners.PipelineRunnerRegistrar; | ||
| import org.apache.beam.sdk.testing.TestPipelineOptions; | ||
| import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Charsets; | ||
| import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList; | ||
| import org.hamcrest.Matchers; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| /** A {@link PipelineRunner} a {@link Pipeline} against a {@code JobService}. */ | ||
| public class TestUniversalRunner extends PipelineRunner<PipelineResult> { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(TestUniversalRunner.class); | ||
|
|
||
| private final PipelineOptions options; | ||
|
|
||
| private TestUniversalRunner(PipelineOptions options) { | ||
| this.options = options; | ||
| } | ||
|
|
||
| /** | ||
| * Constructs a runner from the provided options. | ||
| * | ||
| * @param options Properties which configure the runner. | ||
| * @return The newly created runner. | ||
| */ | ||
| public static TestUniversalRunner fromOptions(PipelineOptions options) { | ||
|
There was a problem hiding this comment. Had to add this, because |
||
| return new TestUniversalRunner(options); | ||
| } | ||
|
|
||
| @Override | ||
| public PipelineResult run(Pipeline pipeline) { | ||
| Options testOptions = options.as(Options.class); | ||
| if (testOptions.getLocalJobServicePortFile() != null) { | ||
| String localServicePortFilePath = testOptions.getLocalJobServicePortFile(); | ||
| try { | ||
| testOptions.setJobEndpoint( | ||
| "localhost:" | ||
| + new String( | ||
| Files.readAllBytes(Paths.get(localServicePortFilePath)), Charsets.UTF_8) | ||
| .trim()); | ||
| } catch (IOException e) { | ||
| throw new RuntimeException( | ||
| String.format("Error reading local job service port file %s", localServicePortFilePath), | ||
| e); | ||
| } | ||
| } | ||
|
|
||
| PortablePipelineOptions portableOptions = options.as(PortablePipelineOptions.class); | ||
| portableOptions.setRunner(PortableRunner.class); | ||
| PortableRunner runner = PortableRunner.fromOptions(portableOptions); | ||
| PipelineResult result = runner.run(pipeline); | ||
| assertThat( | ||
| "Pipeline did not succeed.", | ||
| result.waitUntilFinish(), | ||
| Matchers.is(PipelineResult.State.DONE)); | ||
| return result; | ||
| } | ||
|
|
||
| public interface Options extends TestPipelineOptions, PortablePipelineOptions { | ||
| /** | ||
| * A file containing the job service port, since Gradle needs to know this filename statically | ||
| * to provide it in Beam testing options. | ||
| * | ||
| * <p>This option conflicts with {@link #getJobEndpoint()}. This option will override the job | ||
| * endpoint to be localhost at the port specified in the file. | ||
| */ | ||
| @Description("File containing local job service port.") | ||
|
There was a problem hiding this comment. Logically, it would make sense to let this be optional (e.g. one could instead provide jobEndpoint directly). A point could be made that the testing infrastructure should be the one reading the file and setting jobEndpoint, rather than passing the file path as an option (but I don't know how much messier that'd make things). There was a problem hiding this comment. Yea, this sort of thing is what took 90% of the time for this PR actually. Scraping around Gradle's docs and the internet for ways to insert that little bit of logic, because realizing it was sort of against the grain. Pipeline options are passed as a Java system property, and those are set up in the Gradle graph construction phase. More generally, there's not a Gradle graph execution-time slot for free-form code that also re-uses the Test task type. Perhaps they expect you to use inheritance and make a new Task type. Which I would rather not do ;_; It would be fine to have two pipeline options, so that simple use could be simple. There was a problem hiding this comment. We already have the plain |
||
| String getLocalJobServicePortFile(); | ||
|
|
||
| void setLocalJobServicePortFile(String endpoint); | ||
| } | ||
|
|
||
| /** Register {@link Options}. */ | ||
| @AutoService(PipelineOptionsRegistrar.class) | ||
| public static class OptionsRegistrar implements PipelineOptionsRegistrar { | ||
|
|
||
| @Override | ||
| public Iterable<Class<? extends PipelineOptions>> getPipelineOptions() { | ||
| return ImmutableList.of(Options.class); | ||
| } | ||
| } | ||
|
|
||
| /** Registrar for the portable runner. */ | ||
| @AutoService(PipelineRunnerRegistrar.class) | ||
| public static class RunnerRegistrar implements PipelineRunnerRegistrar { | ||
|
|
||
| @Override | ||
| public Iterable<Class<? extends PipelineRunner<?>>> getPipelineRunners() { | ||
| return ImmutableList.of(TestUniversalRunner.class); | ||
| } | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@robertwb @lukecwik first thing I hit putting this together was exceeding message size limit