[BEAM-4176] Initial implementation for running portable runner tests #5935

angoenka · 2018-07-12T00:36:33Z

Basic idea is to start a job server using a JobServerDriver which can be initialized with provided parameters.
TestPortableRunner will start a new instance of JobServer for every test and then run the test on it.

I have kept the maxParallelForks to 1 for now as i want to first make sure that all the tests pass with no concurrency.

Follow this checklist to help us incorporate your contribution quickly and easily:

Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

It will help us expedite review of your Pull Request if you tag someone (e.g. @username) to look at it.

Post-Commit Tests Status (on master branch)

Lang	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go	---	---	---	---	---	---
Java
Python	---		---	---	---	---

angoenka · 2018-07-12T00:37:42Z

R: @bsidhom @jkff @robertwb @axelmagn @ryan-williams

angoenka · 2018-07-12T00:38:28Z

Note that the validate runner tests are not yet passing for flink and will have to be debugged separately

jkff · 2018-07-12T00:40:45Z

Wow, nice!!! (didn't look at code yet) Can we start with ULR then? CC: @youngoli

angoenka · 2018-07-12T00:42:09Z

Yes, though I have not plugged in the gradle task for ULR but it should just follow the Flink task.

angoenka · 2018-07-12T01:11:17Z

verifyPAssertsSucceeded requires support for metrics in pipelineResult which is causing almost all the tests to fail.

bsidhom · 2018-07-12T01:17:23Z

Ah, yes. I had forgotten about that. That will cause problems for us unless we somehow rewrite PAssert to not depend on metrics if we're running portably. I guess a pipeline option could do it.

tweise · 2018-07-14T20:17:39Z

JIRA?

angoenka · 2018-07-16T21:54:34Z

The tests are not passing yet but that seems to be a separate issue.
We can proceed with this PR review and get it in so that people can easily run the tests and iterate.

axelmagn

Does validatesRunner get run as part of any automated framework? I'm happy to merge this and then iterate on getting tests to pass, but I worry about putting any automated tests into a red state.

axelmagn · 2018-07-18T22:18:03Z

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkJobServerDriver.java

-    @Option(name = "--job-host", required = true, usage = "The job server host string")
+  /** Configuration for the jobServer. */
+  public static class ServerConfiguration {
+    @Option(name = "--job-host", usage = "The job server host string")


why are we no longer requiring this?

Ports can be chosen dynamically if not provided.
This help facilitate launch of multiple jobServices on the same machine by finding an unused port.

Finding and managing unused ports reliably is surprisingly difficult. The major challenge is that once you pick a port without using it, it can be picked and used by some other process without your knowledge.

Please let me know of any other ways/library to get dynamic unused ports in tests case.

Gotcha. that makes sense.

I have no idea about other ways to manage unused ports (short of writing our own manager), so I think you're right that this is the way to do it.

Is this a local port? If so, it's quite easy to do in Java by allocating a new TCP server socket at port 0. You can then query it for the chosen port. This works as long as you are able to pass that socket directly to the service implementation. (Note that you can do this with gRPC and we have a Beam-specific tool that does this for you).

If you're trying to find an unused port for a non-Java server or for a server that will run separately (i.e., for which you do not directly control the socket), you can use the same server allocation trick and note the chosen port number. You can then kill the server socket and pass that port into your new server's configuration. Note that doing this approach introduces a race condition but will work in most cases (especially with retries).

That seems to be possible.
Comparing the approach

Optional port and query the port back

Passing a socket to the flink driver

I feel option 1 seems to be simpler while keeping initialization flexible.

…lidates runner test

angoenka · 2018-07-20T21:05:06Z

Ping for the review!
cc: @bsidhom @lukecwik @axelmagn @ryan-williams

lukecwik · 2018-07-23T18:15:07Z

runners/flink/job-server/build.gradle

+  boolean streaming
+}
+
+def createPortableValidatesRunnerTask(Map m) {


nit: use a lambda and the implicit it parameter as it allows you to call the method without any paramemters or with parameters

SG.
I am not very sure with the semantics so please take a look again.

lukecwik · 2018-07-23T18:16:40Z

runners/flink/job-server/build.gradle

+  }
+}
+
+createPortableValidatesRunnerTask(name: "validatesPortableRunner", jobServerDriver: "org.apache.beam.runners.flink.FlinkJobServerDriver", jobServerConfig: "[]", streaming: false)


Do you expect jobServerDriver to change?

Not for Flink.

But I want to move it to createPortableValidatesRunnerTask reference runner project so that it can be reused. I am not sure how to do it though.

You'll want to move it into the BeamModulePlugin.groovy and expose it like applyJavaNature and the other methods there. This can happen in a follow up PR.

lukecwik · 2018-07-23T18:16:57Z

runners/flink/job-server/build.gradle

@@ -45,3 +59,55 @@ runShadow {
  // Enable remote debugging.
  jvmArgs = ["-Xdebug", "-Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=5005"]
 }
+
+class PortableValidatesRunnerConfig {
+  String name


Please add comments for what each of these configuration properties do

lukecwik · 2018-07-23T18:25:23Z

runners/flink/job-server/build.gradle

+ * the following projects are evaluated before we evaluate this project. This is because
+ * we are attempting to reference the "sourceSets.test.output" directly.
+ */
+evaluationDependsOn(":beam-runners-flink_2.11")


You reference :beam-sdks-java-core and :beam-runners-core-java sourceSets.test.output directly and not :beam-runners-flink_2.11

lukecwik · 2018-07-23T18:25:55Z

runners/flink/job-server/build.gradle

 dependencies {
  compile project(path: ":beam-runners-flink_2.11", configuration: "shadow")
+  validatesRunner project(path: ":beam-runners-flink_2.11", configuration: "shadowTest")


You need to add :beam-sdks-java-core / shadowTest and :beam-runners-core-java / shadowTest as a validatesRunner dependency

lukecwik · 2018-07-23T18:31:38Z

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkJobServerDriver.java

+            "ArtifactStagingServer stopped on {}", jobServer.getApiServiceDescriptor().getUrl());
+        artifactStagingServer = null;
+      } catch (Exception e) {
+        LOG.error("Error while closing the artifactStagingServer.");


Log the error

lukecwik · 2018-07-23T18:33:02Z

...ava/src/main/java/org/apache/beam/runners/reference/testing/TestPortablePipelineOptions.java

+  @Required
+  @Description(
+      "Fully qualified class name of TestJobServiceDriver capable of managing the JobService.")
+  String getJobServerDriver();


This can be of type Class, PipelineOptionsFactory will convert string -> Class for you automatically during parsing.

lukecwik · 2018-07-23T18:33:59Z

...ava/src/main/java/org/apache/beam/runners/reference/testing/TestPortablePipelineOptions.java

+  void setJobServerDriver(String jobServerDriver);
+
+  @Description("String containing comma separated arguments for the JobServer.")
+  String getJobServerConfig();


PipelineOptionsFactory supports parsing List or String[] automatically. There is no need to do this yourself.

lukecwik · 2018-07-23T18:35:36Z

...ference/java/src/main/java/org/apache/beam/runners/reference/testing/TestPortableRunner.java

+ * {@link TestPortableRunner} is a pipeline runner that wraps a {@link PortableRunner} when running
+ * tests against the {@link TestPipeline}.
+ *
+ * <p>This runner requires a JobServerDriver with following methods.


Make the FlinkJobServerDriver implement a new interface TestPortableRunner.JobServerDriver which exposes these methods.

Will Take it up in subsequent diff as there is some class dependency issue.

I believe the issue is that TestPortableRunner shouldn't be part of the reference runner subproject but in a place where every runner could refer to it. SGTM as a change in a subsequent PR.

lukecwik · 2018-07-23T18:37:50Z

...ference/java/src/main/java/org/apache/beam/runners/reference/testing/TestPortableRunner.java

+      try {
+        jobServerDriver =
+            jobServerDriverClass
+                .getMethod("fromParams", String[].class)


Reuse InstanceBuilder:
https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/util/InstanceBuilder.java

ryan-williams · 2018-07-23T18:45:19Z

I don't have comments on this change yet, but wanted to just mention this prototype of an ArtifactStagingService that:

detects artifacts that have already been uploaded (by MD5)
closes the connection to prevent redundant artifacts from being uploaded
leaves the manifest with all the entries for each job (just with URIs that may point to where a previous job staged the artifact)

It doesn't help the main pain point here (re-staging ~200MB of artifacts for each test case) because it doesn't work across artifact-server instances, but that's the next extension I'm planning to look into.

angoenka

Thanks Luke.
Applied the review comments.

angoenka · 2018-07-24T17:51:27Z

runners/flink/job-server/build.gradle

+ * the following projects are evaluated before we evaluate this project. This is because
+ * we are attempting to reference the "sourceSets.test.output" directly.
+ */
+evaluationDependsOn(":beam-runners-flink_2.11")


angoenka · 2018-07-24T17:52:45Z

runners/flink/job-server/build.gradle

 dependencies {
  compile project(path: ":beam-runners-flink_2.11", configuration: "shadow")
+  validatesRunner project(path: ":beam-runners-flink_2.11", configuration: "shadowTest")


angoenka · 2018-07-24T17:55:57Z

runners/flink/job-server/build.gradle

@@ -45,3 +59,55 @@ runShadow {
  // Enable remote debugging.
  jvmArgs = ["-Xdebug", "-Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=5005"]
 }
+
+class PortableValidatesRunnerConfig {
+  String name


angoenka · 2018-07-24T18:04:34Z

runners/flink/job-server/build.gradle

+  boolean streaming
+}
+
+def createPortableValidatesRunnerTask(Map m) {


SG.
I am not very sure with the semantics so please take a look again.

angoenka · 2018-07-24T18:06:44Z

runners/flink/job-server/build.gradle

+  }
+}
+
+createPortableValidatesRunnerTask(name: "validatesPortableRunner", jobServerDriver: "org.apache.beam.runners.flink.FlinkJobServerDriver", jobServerConfig: "[]", streaming: false)


Not for Flink.

But I want to move it to createPortableValidatesRunnerTask reference runner project so that it can be reused. I am not sure how to do it though.

angoenka · 2018-07-24T18:18:03Z

...ava/src/main/java/org/apache/beam/runners/reference/testing/TestPortablePipelineOptions.java

+  @Required
+  @Description(
+      "Fully qualified class name of TestJobServiceDriver capable of managing the JobService.")
+  String getJobServerDriver();


angoenka · 2018-07-24T18:19:26Z

...ava/src/main/java/org/apache/beam/runners/reference/testing/TestPortablePipelineOptions.java

+  void setJobServerDriver(String jobServerDriver);
+
+  @Description("String containing comma separated arguments for the JobServer.")
+  String getJobServerConfig();


angoenka · 2018-07-24T18:20:26Z

...ference/java/src/main/java/org/apache/beam/runners/reference/testing/TestPortableRunner.java

+ * <ul>
+ *   <li>public static Object fromParams(String... params)
+ *   <li>public String start() // Start JobServer and returns the JobServer host and port.
+ *   <li>public void start() // Stop the JobServer and free all resources.


angoenka · 2018-07-24T18:23:51Z

...ference/java/src/main/java/org/apache/beam/runners/reference/testing/TestPortableRunner.java

+      try {
+        jobServerDriver =
+            jobServerDriverClass
+                .getMethod("fromParams", String[].class)


angoenka · 2018-07-24T19:01:04Z

...ference/java/src/main/java/org/apache/beam/runners/reference/testing/TestPortableRunner.java

+ * {@link TestPortableRunner} is a pipeline runner that wraps a {@link PortableRunner} when running
+ * tests against the {@link TestPipeline}.
+ *
+ * <p>This runner requires a JobServerDriver with following methods.


Will Take it up in subsequent diff as there is some class dependency issue.

tweise · 2018-07-25T01:49:54Z

Run Java PreCommit

angoenka · 2018-07-25T21:43:44Z

Followup https://issues.apache.org/jira/browse/BEAM-5022

lukecwik

This looks good, made some minor changes directly. Will wait for tests to be green to confirm my edits were ok and then will merge.

angoenka · 2018-07-25T21:47:11Z

Thanks Luke.
I have created a jira to track the 2 changes we discussed.

angoenka force-pushed the portable_test_runner branch 2 times, most recently from 4d2d949 to 36a6595 Compare July 12, 2018 18:32

ryan-williams mentioned this pull request Jul 15, 2018

[BEAM-4800] send PutArtifactResponse in BeamFileSystemArtifactStagingService #5954

Merged

angoenka changed the title ~~Initial implementation for running portable runner tests~~ [BEAM-4176] Initial implementation for running portable runner tests Jul 16, 2018

axelmagn reviewed Jul 18, 2018

View reviewed changes

angoenka added 5 commits July 19, 2018 12:31

TestPortableRunner for validate runner tests

9c4ed17

Dynamically assign jobServer host and port.

89ed41e

Documentation formatting

0e19644

Disable Metrics for Flink Portable Runner

e79c526

Change to gradle files and check for pipeline completion for flink va…

9e1ff24

…lidates runner test

angoenka force-pushed the portable_test_runner branch from 36a6595 to 9e1ff24 Compare July 20, 2018 03:39

angoenka requested a review from lukecwik as a code owner July 20, 2018 03:39

lukecwik reviewed Jul 23, 2018

View reviewed changes

Review Comments

208ea38

angoenka commented Jul 24, 2018

View reviewed changes

Adding javadoc for default option factory

15aca9c

lukecwik added 3 commits July 25, 2018 14:35

Update lambda to handle 0 method args.

e5d1d6f

Fix simple typo

e98efc7

Use varargs in array based method declaration.

9b79adb

lukecwik approved these changes Jul 25, 2018

View reviewed changes

Fixing Log message

ffd0843

lukecwik merged commit 75e3986 into apache:master Jul 25, 2018

kennknowles mentioned this pull request Jun 3, 2022

Python PortableTestRunner that runs VR tests against a given portable runner #19018

Open

[BEAM-4176] Initial implementation for running portable runner tests #5935

[BEAM-4176] Initial implementation for running portable runner tests #5935

Conversation

angoenka commented Jul 12, 2018 • edited

Post-Commit Tests Status (on master branch)

angoenka commented Jul 12, 2018

angoenka commented Jul 12, 2018

jkff commented Jul 12, 2018

angoenka commented Jul 12, 2018

angoenka commented Jul 12, 2018

bsidhom commented Jul 12, 2018

tweise commented Jul 14, 2018

angoenka commented Jul 16, 2018

axelmagn left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

angoenka Jul 18, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

angoenka commented Jul 20, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryan-williams commented Jul 23, 2018

angoenka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tweise commented Jul 25, 2018

angoenka commented Jul 25, 2018

lukecwik left a comment • edited

Choose a reason for hiding this comment

angoenka commented Jul 25, 2018 • edited

angoenka commented Jul 12, 2018 •

edited

angoenka Jul 18, 2018 •

edited

lukecwik left a comment •

edited

angoenka commented Jul 25, 2018 •

edited