Merge pull request #29924: Merge runners-core-construction into sdks-…

…java-core

CHANGES.md

@@ -67,6 +67,8 @@
 
 * [Enrichment Transform](https://s.apache.org/enrichment-transform) along with GCP BigTable handler added to Python SDK ([#30001](https://github.com/apache/beam/pull/30001)).
 * Allow writing clustered and not time partitioned BigQuery tables (Java) ([#30094](https://github.com/apache/beam/pull/30094)).
+* Merged sdks/java/fn-execution and runners/core-construction-java into the main SDK. These artifacts were never meant for users, but noting
+  that they no longer exist. These are steps to bring portability into the core SDK alongside all other core functionality.
 
 ## Breaking Changes
 

build.gradle.kts

@@ -247,7 +247,6 @@ tasks.register("javaPreCommit") {
   dependsOn(":model:fn-execution:build")
   dependsOn(":model:job-management:build")
   dependsOn(":model:pipeline:build")
-  dependsOn(":runners:core-construction-java:build")
   dependsOn(":runners:core-java:build")
   dependsOn(":runners:direct-java:build")
   dependsOn(":runners:direct-java:needsRunnerTests")

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -2591,7 +2591,6 @@ class BeamModulePlugin implements Plugin<Project> {
 
       project.evaluationDependsOn(":sdks:python")
       project.evaluationDependsOn(config.expansionProjectPath)
-      project.evaluationDependsOn(":runners:core-construction-java")
       project.evaluationDependsOn(":sdks:java:extensions:python")
 
       // Setting up args to launch the expansion service
@@ -2685,7 +2684,6 @@ class BeamModulePlugin implements Plugin<Project> {
 
       project.evaluationDependsOn(":sdks:python")
       project.evaluationDependsOn(":sdks:java:testing:expansion-service")
-      project.evaluationDependsOn(":runners:core-construction-java")
       project.evaluationDependsOn(":sdks:java:extensions:python")
       project.evaluationDependsOn(":sdks:go:test")
 
@@ -2750,11 +2748,9 @@ class BeamModulePlugin implements Plugin<Project> {
           systemProperty "expansionPort", port
           systemProperty "semiPersistDir", config.semiPersistDir
           classpath = config.classpath + project.files(
-              project.project(":runners:core-construction-java").sourceSets.test.runtimeClasspath,
               project.project(":sdks:java:extensions:python").sourceSets.test.runtimeClasspath
               )
           testClassesDirs = project.files(
-              project.project(":runners:core-construction-java").sourceSets.test.output.classesDirs,
               project.project(":sdks:java:extensions:python").sourceSets.test.output.classesDirs
               )
           maxParallelForks config.numParallelTests
@@ -2861,7 +2857,6 @@ class BeamModulePlugin implements Plugin<Project> {
       def config = it ? it as TransformServiceConfiguration : new TransformServiceConfiguration()
 
       project.evaluationDependsOn(":sdks:python")
-      project.evaluationDependsOn(":runners:core-construction-java")
       project.evaluationDependsOn(":sdks:java:extensions:python")
       project.evaluationDependsOn(":sdks:java:transform-service:app")
 

playground/backend/internal/preparers/java_preparers.go

@@ -38,7 +38,7 @@ const (
 	newLinePattern                    = "\n"
 	javaPublicClassNamePattern        = "public class (.*?) [{|implements(.*)]"
 	pipelineNamePattern               = `Pipeline\s([A-z|0-9_]*)\s=\sPipeline\.create`
-	graphSavePattern                  = "String dotString = org.apache.beam.runners.core.construction.renderer.PipelineDotRenderer.toDotString(%s);\n" +
+	graphSavePattern                  = "String dotString = org.apache.beam.sdk.util.construction.renderer.PipelineDotRenderer.toDotString(%s);\n" +
 		"    try (java.io.PrintWriter out = new java.io.PrintWriter(\"graph.dot\")) {\n      " +
 		"		out.println(dotString);\n    " +
 		"	} catch (java.io.FileNotFoundException e) {\n" +

playground/backend/internal/preparers/java_preparers_test.go

@@ -130,9 +130,9 @@ func TestGetJavaPreparers(t *testing.T) {
 }
 
 func Test_findPipelineObjectName(t *testing.T) {
-	code := "package org.apache.beam.examples;\n\n/*\n * Licensed to the Apache Software Foundation (ASF) under one\n * or more contributor license agreements.  See the NOTICE file\n * distributed with this work for additional information\n * regarding copyright ownership.  The ASF licenses this file\n * to you under the Apache License, Version 2.0 (the\n * \"License\"); you may not use this file except in compliance\n * with the License.  You may obtain a copy of the License at\n *\n *     http://www.apache.org/licenses/LICENSE-2.0\n *\n * Unless required by applicable law or agreed to in writing, software\n * distributed under the License is distributed on an \"AS IS\" BASIS,\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n * See the License for the specific language governing permissions and\n * limitations under the License.\n */\n\n// beam-playground:\n//   name: Branching\n//   description: Task from katas to branch out the numbers to two different transforms, one transform\n//     is multiplying each number by 5 and the other transform is multiplying each number by 10.\n//   multifile: false\n//   categories:\n//     - Branching\n//     - Core Transforms\n\nimport static org.apache.beam.sdk.values.TypeDescriptors.integers;\n\nimport org.apache.beam.sdk.Pipeline;\nimport org.apache.beam.sdk.options.PipelineOptions;\nimport org.apache.beam.sdk.options.PipelineOptionsFactory;\nimport org.apache.beam.sdk.transforms.Create;\nimport org.apache.beam.sdk.transforms.MapElements;\nimport org.apache.beam.sdk.values.PCollection;\nimport org.apache.beam.runners.core.construction.renderer.PipelineDotRenderer;\n\npublic class Task {\n\n\n\n  public static void main(String[] args) {\n    PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();\n    Pipeline pipeline = Pipeline.create(options);\n\n    PCollection<Integer> numbers =\n        pipeline.apply(Create.of(1, 2, 3, 4, 5));\n\n    PCollection<Integer> mult5Results = applyMultiply5Transform(numbers);\n    PCollection<Integer> mult10Results = applyMultiply10Transform(numbers);\n\n    mult5Results.apply(\"Log multiply 5\", Log.ofElements(\"Multiplied by 5: \"));\n    mult10Results.apply(\"Log multiply 10\", Log.ofElements(\"Multiplied by 10: \"));\n\n    String dotString = PipelineDotRenderer.toDotString(pipeline);\n    System.out.println(dotString);\n    pipeline.run();\n\n  }\n\n  static PCollection<Integer> applyMultiply5Transform(PCollection<Integer> input) {\n    return input.apply(\"Multiply by 5\", MapElements.into(integers()).via(num -> num * 5));\n  }\n\n  static PCollection<Integer> applyMultiply10Transform(PCollection<Integer> input) {\n    return input.apply(\"Multiply by 10\", MapElements.into(integers()).via(num -> num * 10));\n  }\n\n}\n"
+	code := "package org.apache.beam.examples;\n\n/*\n * Licensed to the Apache Software Foundation (ASF) under one\n * or more contributor license agreements.  See the NOTICE file\n * distributed with this work for additional information\n * regarding copyright ownership.  The ASF licenses this file\n * to you under the Apache License, Version 2.0 (the\n * \"License\"); you may not use this file except in compliance\n * with the License.  You may obtain a copy of the License at\n *\n *     http://www.apache.org/licenses/LICENSE-2.0\n *\n * Unless required by applicable law or agreed to in writing, software\n * distributed under the License is distributed on an \"AS IS\" BASIS,\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n * See the License for the specific language governing permissions and\n * limitations under the License.\n */\n\n// beam-playground:\n//   name: Branching\n//   description: Task from katas to branch out the numbers to two different transforms, one transform\n//     is multiplying each number by 5 and the other transform is multiplying each number by 10.\n//   multifile: false\n//   categories:\n//     - Branching\n//     - Core Transforms\n\nimport static org.apache.beam.sdk.values.TypeDescriptors.integers;\n\nimport org.apache.beam.sdk.Pipeline;\nimport org.apache.beam.sdk.options.PipelineOptions;\nimport org.apache.beam.sdk.options.PipelineOptionsFactory;\nimport org.apache.beam.sdk.transforms.Create;\nimport org.apache.beam.sdk.transforms.MapElements;\nimport org.apache.beam.sdk.values.PCollection;\nimport org.apache.beam.sdk.util.construction.renderer.construction.PipelineDotRenderer;\n\npublic class Task {\n\n\n\n  public static void main(String[] args) {\n    PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();\n    Pipeline pipeline = Pipeline.create(options);\n\n    PCollection<Integer> numbers =\n        pipeline.apply(Create.of(1, 2, 3, 4, 5));\n\n    PCollection<Integer> mult5Results = applyMultiply5Transform(numbers);\n    PCollection<Integer> mult10Results = applyMultiply10Transform(numbers);\n\n    mult5Results.apply(\"Log multiply 5\", Log.ofElements(\"Multiplied by 5: \"));\n    mult10Results.apply(\"Log multiply 10\", Log.ofElements(\"Multiplied by 10: \"));\n\n    String dotString = PipelineDotRenderer.toDotString(pipeline);\n    System.out.println(dotString);\n    pipeline.run();\n\n  }\n\n  static PCollection<Integer> applyMultiply5Transform(PCollection<Integer> input) {\n    return input.apply(\"Multiply by 5\", MapElements.into(integers()).via(num -> num * 5));\n  }\n\n  static PCollection<Integer> applyMultiply10Transform(PCollection<Integer> input) {\n    return input.apply(\"Multiply by 10\", MapElements.into(integers()).via(num -> num * 10));\n  }\n\n}\n"
 	lc := createTempFileWithCode(code)
-	codeWithoutPipeline := "package org.apache.beam.examples;\n\n/*\n * Licensed to the Apache Software Foundation (ASF) under one\n * or more contributor license agreements.  See the NOTICE file\n * distributed with this work for additional information\n * regarding copyright ownership.  The ASF licenses this file\n * to you under the Apache License, Version 2.0 (the\n * \"License\"); you may not use this file except in compliance\n * with the License.  You may obtain a copy of the License at\n *\n *     http://www.apache.org/licenses/LICENSE-2.0\n *\n * Unless required by applicable law or agreed to in writing, software\n * distributed under the License is distributed on an \"AS IS\" BASIS,\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n * See the License for the specific language governing permissions and\n * limitations under the License.\n */\n\n// beam-playground:\n//   name: Branching\n//   description: Task from katas to branch out the numbers to two different transforms, one transform\n//     is multiplying each number by 5 and the other transform is multiplying each number by 10.\n//   multifile: false\n//   categories:\n//     - Branching\n//     - Core Transforms\n\nimport static org.apache.beam.sdk.values.TypeDescriptors.integers;\n\nimport org.apache.beam.sdk.Pipeline;\nimport org.apache.beam.sdk.options.PipelineOptions;\nimport org.apache.beam.sdk.options.PipelineOptionsFactory;\nimport org.apache.beam.sdk.transforms.Create;\nimport org.apache.beam.sdk.transforms.MapElements;\nimport org.apache.beam.sdk.values.PCollection;\nimport org.apache.beam.runners.core.construction.renderer.PipelineDotRenderer;\n\npublic class Task {\n\n\n\n  public static void main(String[] args) {\n    PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();\n     PCollection<Integer> numbers =\n        pipeline.apply(Create.of(1, 2, 3, 4, 5));\n\n    PCollection<Integer> mult5Results = applyMultiply5Transform(numbers);\n    PCollection<Integer> mult10Results = applyMultiply10Transform(numbers);\n\n    mult5Results.apply(\"Log multiply 5\", Log.ofElements(\"Multiplied by 5: \"));\n    mult10Results.apply(\"Log multiply 10\", Log.ofElements(\"Multiplied by 10: \"));\n\n    String dotString = PipelineDotRenderer.toDotString(pipeline);\n    System.out.println(dotString);\n    pipeline.run();\n\n  }\n\n  static PCollection<Integer> applyMultiply5Transform(PCollection<Integer> input) {\n    return input.apply(\"Multiply by 5\", MapElements.into(integers()).via(num -> num * 5));\n  }\n\n  static PCollection<Integer> applyMultiply10Transform(PCollection<Integer> input) {\n    return input.apply(\"Multiply by 10\", MapElements.into(integers()).via(num -> num * 10));\n  }\n\n}\n"
+	codeWithoutPipeline := "package org.apache.beam.examples;\n\n/*\n * Licensed to the Apache Software Foundation (ASF) under one\n * or more contributor license agreements.  See the NOTICE file\n * distributed with this work for additional information\n * regarding copyright ownership.  The ASF licenses this file\n * to you under the Apache License, Version 2.0 (the\n * \"License\"); you may not use this file except in compliance\n * with the License.  You may obtain a copy of the License at\n *\n *     http://www.apache.org/licenses/LICENSE-2.0\n *\n * Unless required by applicable law or agreed to in writing, software\n * distributed under the License is distributed on an \"AS IS\" BASIS,\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n * See the License for the specific language governing permissions and\n * limitations under the License.\n */\n\n// beam-playground:\n//   name: Branching\n//   description: Task from katas to branch out the numbers to two different transforms, one transform\n//     is multiplying each number by 5 and the other transform is multiplying each number by 10.\n//   multifile: false\n//   categories:\n//     - Branching\n//     - Core Transforms\n\nimport static org.apache.beam.sdk.values.TypeDescriptors.integers;\n\nimport org.apache.beam.sdk.Pipeline;\nimport org.apache.beam.sdk.options.PipelineOptions;\nimport org.apache.beam.sdk.options.PipelineOptionsFactory;\nimport org.apache.beam.sdk.transforms.Create;\nimport org.apache.beam.sdk.transforms.MapElements;\nimport org.apache.beam.sdk.values.PCollection;\nimport org.apache.beam.sdk.util.construction.renderer.construction.PipelineDotRenderer;\n\npublic class Task {\n\n\n\n  public static void main(String[] args) {\n    PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();\n     PCollection<Integer> numbers =\n        pipeline.apply(Create.of(1, 2, 3, 4, 5));\n\n    PCollection<Integer> mult5Results = applyMultiply5Transform(numbers);\n    PCollection<Integer> mult10Results = applyMultiply10Transform(numbers);\n\n    mult5Results.apply(\"Log multiply 5\", Log.ofElements(\"Multiplied by 5: \"));\n    mult10Results.apply(\"Log multiply 10\", Log.ofElements(\"Multiplied by 10: \"));\n\n    String dotString = PipelineDotRenderer.toDotString(pipeline);\n    System.out.println(dotString);\n    pipeline.run();\n\n  }\n\n  static PCollection<Integer> applyMultiply5Transform(PCollection<Integer> input) {\n    return input.apply(\"Multiply by 5\", MapElements.into(integers()).via(num -> num * 5));\n  }\n\n  static PCollection<Integer> applyMultiply10Transform(PCollection<Integer> input) {\n    return input.apply(\"Multiply by 10\", MapElements.into(integers()).via(num -> num * 10));\n  }\n\n}\n"
 	lcWithoutPipeline := createTempFileWithCode(codeWithoutPipeline)
 	path, _ := os.Getwd()
 	defer os.RemoveAll(filepath.Join(path, "temp"))

runners/core-java/build.gradle

@@ -42,7 +42,6 @@ dependencies {
   implementation project(path: ":model:pipeline", configuration: "shadow")
   implementation project(path: ":sdks:java:core", configuration: "shadow")
   implementation project(path: ":model:job-management", configuration: "shadow")
-  implementation project(":runners:core-construction-java")
   implementation library.java.vendored_guava_32_1_2_jre
   implementation library.java.joda_time
   implementation library.java.vendored_grpc_1_60_1

runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetNewDoFn.java

@@ -18,14 +18,14 @@
 package org.apache.beam.runners.core;
 
 import java.util.Collection;
-import org.apache.beam.runners.core.construction.TriggerTranslation;
 import org.apache.beam.runners.core.triggers.ExecutableTriggerStateMachine;
 import org.apache.beam.runners.core.triggers.TriggerStateMachines;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.transforms.windowing.PaneInfo;
 import org.apache.beam.sdk.util.SystemDoFnInternal;
 import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.util.construction.TriggerTranslation;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.TupleTag;
 import org.apache.beam.sdk.values.WindowingStrategy;