Remove Pipeline as a field in pipeline classes (#1119)

In tests we use a TestPipelineExtension which does some static
initialization that should not be repeated the same JVM. In our
XXXPipeline classes we save the pipeline as a field and usually write lambdas
that are pass to the pipeline. Because lambdas are effectively anonymous inner
classes they are bound to their enclosing instances. When they get serialized
during pipeline execution, their enclosing classes also do. This might result
in undefined behavior when multiple lambdas in the same XXXPipeline are used
on the same JVM (such as in tests) where the static initialization may be done
multiple times if different class loaders are used. This is very
unlikely to happen but as a best practice we still remove them as
fields.

core/src/test/java/google/registry/beam/initsql/InitSqlPipelineGraphTest.java

@@ -52,7 +52,7 @@ class InitSqlPipelineGraphTest {
 
   @Test
   void createPipeline_compareGraph() throws IOException {
-    new InitSqlPipeline(options, testPipeline).setupPipeline();
+    new InitSqlPipeline(options).setupPipeline(testPipeline);
     String dotString = PipelineDotRenderer.toDotString(testPipeline);
     URL goldenDotUrl = Resources.getResource(InitSqlPipelineGraphTest.class, GOLDEN_DOT_FILE);
     File outputFile = new File(new File(goldenDotUrl.getFile()).getParent(), "pipeline_curr.dot");

core/src/test/java/google/registry/beam/initsql/InitSqlPipelineTest.java

@@ -317,8 +317,8 @@ class InitSqlPipelineTest {
                 "--commitLogDir=" + commitLogDir.getAbsolutePath())
             .withValidation()
             .as(InitSqlPipelineOptions.class);
-    InitSqlPipeline initSqlPipeline = new InitSqlPipeline(options, testPipeline);
-    initSqlPipeline.run().waitUntilFinish();
+    InitSqlPipeline initSqlPipeline = new InitSqlPipeline(options);
+    initSqlPipeline.run(testPipeline).waitUntilFinish();
     try (AppEngineEnvironment env = new AppEngineEnvironment("test")) {
       assertHostResourceEquals(
           jpaTm().transact(() -> jpaTm().loadByKey(hostResource.createVKey())), hostResource);