Flink 2 support prerequisites

* Honor getUseDataStreamForBatch pipeline option for Flink portable runner

* Refactor gradle scripts in preparation for Flink 2 support

Author: Abacn

runners/flink/flink_runner.gradle

@@ -28,7 +28,8 @@ import groovy.json.JsonOutput
 def base_path = ".."
 
 def overrides(versions, type, base_path) {
-  versions.collect { "${base_path}/${it}/src/${type}/java" } + ["./src/${type}/java"]
+  // order is important
+  ["${base_path}/src/${type}/java"] + versions.collect { "${base_path}/${it}/src/${type}/java" } + ["./src/${type}/java"]
 }
 
 def all_versions = flink_versions.split(",")
@@ -49,7 +50,8 @@ applyJavaNature(
     automaticModuleName: 'org.apache.beam.runners.flink',
     archivesBaseName: archivesBaseName,
     // flink runner jars are in same package name. Publish javadoc once.
-    exportJavadoc: project.ext.flink_version.startsWith(all_versions.first())
+    exportJavadoc: project.ext.flink_version.startsWith(all_versions.first()),
+    requireJavaVersion: project.ext.flink_major.startsWith('2') ? JavaVersion.VERSION_11 : null
 )
 
 description = "Apache Beam :: Runners :: Flink $flink_version"
@@ -68,10 +70,16 @@ evaluationDependsOn(":examples:java")
  */
 def sourceOverridesBase = project.layout.buildDirectory.dir('source-overrides/src').get()
 
-def copySourceOverrides = tasks.register('copySourceOverrides', Copy) {
-  it.from main_source_overrides
-  it.into "${sourceOverridesBase}/main/java"
-  it.duplicatesStrategy DuplicatesStrategy.INCLUDE
+def copySourceOverrides = tasks.register('copySourceOverrides', Copy) { copyTask ->
+  copyTask.from main_source_overrides
+  copyTask.into "${sourceOverridesBase}/main/java"
+  copyTask.duplicatesStrategy DuplicatesStrategy.INCLUDE
+
+  if (project.ext.has('excluded_files') && project.ext.excluded_files.containsKey('main')) {
+    project.ext.excluded_files.main.each { file ->
+      copyTask.exclude "**/${file}"
+    }
+  }
 }
 
 def copyResourcesOverrides = tasks.register('copyResourcesOverrides', Copy) {
@@ -119,15 +127,15 @@ def sourceBase = "${project.projectDir}/../src"
 sourceSets {
   main {
     java {
-      srcDirs = ["${sourceBase}/main/java", "${sourceOverridesBase}/main/java"]
+      srcDirs = ["${sourceOverridesBase}/main/java"]
     }
     resources {
       srcDirs = ["${sourceBase}/main/resources", "${sourceOverridesBase}/main/resources"]
     }
   }
   test {
     java {
-      srcDirs = ["${sourceBase}/test/java", "${sourceOverridesBase}/test/java"]
+      srcDirs = ["${sourceOverridesBase}/test/java"]
     }
     resources {
       srcDirs = ["${sourceBase}/test/resources", "${sourceOverridesBase}/test/resources"]
@@ -196,7 +204,10 @@ dependencies {
 
   implementation "org.apache.flink:flink-core:$flink_version"
   implementation "org.apache.flink:flink-metrics-core:$flink_version"
-  implementation "org.apache.flink:flink-java:$flink_version"
+  if (project.ext.flink_major.startsWith('1')) {
+    // FLINK-36336: dataset API removed in Flink 2
+    implementation "org.apache.flink:flink-java:$flink_version"
+  }
 
   implementation "org.apache.flink:flink-runtime:$flink_version"
   implementation "org.apache.flink:flink-metrics-core:$flink_version"

runners/flink/job-server-container/flink_job_server_container.gradle

@@ -53,15 +53,19 @@ task copyDockerfileDependencies(type: Copy) {
 }
 
 def pushContainers = project.rootProject.hasProperty(["isRelease"]) || project.rootProject.hasProperty("push-containers")
+def containerName = project.parent.name.startsWith("2") ? "flink_job_server" : "flink${project.parent.name}_job_server"
+def containerTag = project.rootProject.hasProperty(["docker-tag"]) ? project.rootProject["docker-tag"] : project.sdk_version
+if (project.parent.name.startsWith("2")) {
+  containerTag += "-flink" + ${project.parent.name}
+}
 
 docker {
   name containerImageName(
-          name: project.docker_image_default_repo_prefix + "flink${project.parent.name}_job_server",
+          name: project.docker_image_default_repo_prefix + containerName,
           root: project.rootProject.hasProperty(["docker-repository-root"]) ?
                   project.rootProject["docker-repository-root"] :
                   project.docker_image_default_repo_root,
-          tag: project.rootProject.hasProperty(["docker-tag"]) ?
-                  project.rootProject["docker-tag"] : project.sdk_version)
+          tag: containerTag)
   // tags used by dockerTag task
   tags containerImageTags()
   files "./build/"

runners/flink/job-server/flink_job_server.gradle

@@ -29,6 +29,11 @@ apply plugin: 'application'
 // we need to set mainClassName before applying shadow plugin
 mainClassName = "org.apache.beam.runners.flink.FlinkJobServerDriver"
 
+// Resolve the Flink project name (and version) the job-server is based on
+def flinkRunnerProject = "${project.path.replace(":job-server", "")}"
+evaluationDependsOn(flinkRunnerProject)
+boolean isFlink2 = project(flinkRunnerProject).ext.flink_major.startsWith('2')
+
 applyJavaNature(
   automaticModuleName: 'org.apache.beam.runners.flink.jobserver',
   archivesBaseName: project.hasProperty('archives_base_name') ? archives_base_name : archivesBaseName,
@@ -37,11 +42,9 @@ applyJavaNature(
   shadowClosure: {
     append "reference.conf"
   },
+  requireJavaVersion: isFlink2 ? JavaVersion.VERSION_11 : null
 )
 
-// Resolve the Flink project name (and version) the job-server is based on
-def flinkRunnerProject = "${project.path.replace(":job-server", "")}"
-
 description = project(flinkRunnerProject).description + " :: Job Server"
 
 /*
@@ -126,18 +129,22 @@ runShadow {
     jvmArgs += ["-Dorg.slf4j.simpleLogger.defaultLogLevel=${project.property('logLevel')}"]
 }
 
-def portableValidatesRunnerTask(String name, boolean streaming, boolean checkpointing, boolean docker) {
+def portableValidatesRunnerTask(String name, String mode, boolean checkpointing, boolean docker) {
   def pipelineOptions = [
       // Limit resource consumption via parallelism
       "--parallelism=2",
   ]
+  boolean streaming = (mode == "streaming")
   if (streaming) {
     pipelineOptions += "--streaming"
     if (checkpointing) {
       pipelineOptions += "--checkpointingInterval=3000"
       pipelineOptions += "--shutdownSourcesAfterIdleMs=60000"
     }
   }
+  if (mode == "batch-datastream") {
+    pipelineOptions += "--useDataStreamForBatch=true"
+  }
   createPortableValidatesRunnerTask(
     name: "validatesPortableRunner${name}",
     jobServerDriver: "org.apache.beam.runners.flink.FlinkJobServerDriver",
@@ -214,14 +221,18 @@ def portableValidatesRunnerTask(String name, boolean streaming, boolean checkpoi
   )
 }
 
-project.ext.validatesPortableRunnerDocker = portableValidatesRunnerTask("Docker", false, false, true)
-project.ext.validatesPortableRunnerBatch = portableValidatesRunnerTask("Batch", false, false, false)
-project.ext.validatesPortableRunnerStreaming = portableValidatesRunnerTask("Streaming", true, false, false)
-project.ext.validatesPortableRunnerStreamingCheckpoint = portableValidatesRunnerTask("StreamingCheckpointing", true, true, false)
+project.ext.validatesPortableRunnerDocker = portableValidatesRunnerTask("Docker", "batch", false, true)
+project.ext.validatesPortableRunnerBatch = portableValidatesRunnerTask("Batch", "batch", false, false)
+project.ext.validatesPortableRunnerBatchDataStream = portableValidatesRunnerTask("Batch", "batch-datastream", false, false)
+project.ext.validatesPortableRunnerStreaming = portableValidatesRunnerTask("Streaming", "streaming", false, false)
+project.ext.validatesPortableRunnerStreamingCheckpoint = portableValidatesRunnerTask("StreamingCheckpointing", "streaming", true, false)
 
 tasks.register("validatesPortableRunner") {
   dependsOn validatesPortableRunnerDocker
-  dependsOn validatesPortableRunnerBatch
+  if (!isFlink2) {
+    dependsOn validatesPortableRunnerBatch
+  }
+  dependsOn validatesPortableRunnerBatchDataStream
   dependsOn validatesPortableRunnerStreaming
   dependsOn validatesPortableRunnerStreamingCheckpoint
 }

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

@@ -84,7 +84,9 @@ public PortablePipelineResult run(final Pipeline pipeline, JobInfo jobInfo) thro
     SdkHarnessOptions.getConfiguredLoggerFromOptions(pipelineOptions.as(SdkHarnessOptions.class));
 
     FlinkPortablePipelineTranslator<?> translator;
-    if (!pipelineOptions.isStreaming() && !hasUnboundedPCollections(pipeline)) {
+    if (!pipelineOptions.getUseDataStreamForBatch()
+        && !pipelineOptions.isStreaming()
+        && !hasUnboundedPCollections(pipeline)) {
       // TODO: Do we need to inspect for unbounded sources before fusing?
       translator = FlinkBatchPortablePipelineTranslator.createTranslator();
     } else {