[BEAM-12281] Drop support for Flink 1.10

CHANGES.md

@@ -75,6 +75,7 @@
 * Kafka Beam SQL tables now ascribe meaning to the LOCATION field; previously
   it was ignored if provided.
 * `TopCombineFn` disallow `compare` as its argument (Python) ([BEAM-7372](https://issues.apache.org/jira/browse/BEAM-7372)).
+* Drop support for Flink 1.10 ([BEAM-12281](https://issues.apache.org/jira/browse/BEAM-12281)).
 
 ## Deprecations
 

gradle.properties

@@ -32,5 +32,5 @@ javaVersion=1.8
 docker_image_default_repo_root=apache
 docker_image_default_repo_prefix=beam_
 
-flink_versions=1.10,1.11,1.12,1.13
+flink_versions=1.11,1.12,1.13
 

release/build.gradle.kts

@@ -39,7 +39,7 @@ task("runJavaExamplesValidationTask") {
   dependsOn(":runners:direct-java:runQuickstartJavaDirect")
   dependsOn(":runners:google-cloud-dataflow-java:runQuickstartJavaDataflow")
   dependsOn(":runners:spark:2:runQuickstartJavaSpark")
-  dependsOn(":runners:flink:1.10:runQuickstartJavaFlinkLocal")
+  dependsOn(":runners:flink:1.13:runQuickstartJavaFlinkLocal")
   dependsOn(":runners:direct-java:runMobileGamingJavaDirect")
   dependsOn(":runners:google-cloud-dataflow-java:runMobileGamingJavaDataflow")
   dependsOn(":runners:twister2:runQuickstartJavaTwister2")

release/src/main/scripts/run_rc_validation.sh

@@ -199,7 +199,7 @@ if [[ "$java_quickstart_flink_local" = true ]]; then
   echo "*************************************************************"
   echo "* Running Java Quickstart with Flink local runner"
   echo "*************************************************************"
-  ./gradlew :runners:flink:1.10:runQuickstartJavaFlinkLocal \
+  ./gradlew :runners:flink:1.13:runQuickstartJavaFlinkLocal \
   -Prepourl=${REPO_URL} \
   -Pver=${RELEASE_VER}
 else

runners/flink/1.11/build.gradle

@@ -22,8 +22,8 @@ project.ext {
   // Set the version of all Flink-related dependencies here.
   flink_version = '1.11.3'
   // Version specific code overrides.
-  main_source_overrides = ["${basePath}/1.10/src/main/java", './src/main/java']
-  test_source_overrides = ["${basePath}/1.10/src/test/java", './src/test/java']
+  main_source_overrides = ['./src/main/java']
+  test_source_overrides = ['./src/test/java']
   main_resources_overrides = []
   test_resources_overrides = []
   archives_base_name = 'beam-runners-flink-1.11'

runners/flink/1.12/build.gradle

@@ -22,8 +22,8 @@ project.ext {
   // Set the version of all Flink-related dependencies here.
   flink_version = '1.12.3'
   // Version specific code overrides.
-  main_source_overrides = ["${basePath}/1.10/src/main/java", "${basePath}/1.11/src/main/java", './src/main/java']
-  test_source_overrides = ["${basePath}/1.10/src/test/java", "${basePath}/1.11/src/test/java", './src/test/java']
+  main_source_overrides = ["${basePath}/1.11/src/main/java", './src/main/java']
+  test_source_overrides = ["${basePath}/1.11/src/test/java", './src/test/java']
   main_resources_overrides = []
   test_resources_overrides = []
   archives_base_name = 'beam-runners-flink-1.12'

runners/flink/1.13/build.gradle

@@ -22,8 +22,8 @@ project.ext {
   // Set the version of all Flink-related dependencies here.
   flink_version = '1.13.0'
   // Version specific code overrides.
-  main_source_overrides = ["${basePath}/1.10/src/main/java", "${basePath}/1.11/src/main/java", "${basePath}/1.12/src/main/java", './src/main/java']
-  test_source_overrides = ["${basePath}/1.10/src/test/java", "${basePath}/1.11/src/test/java", "${basePath}/1.12/src/test/java", './src/test/java']
+  main_source_overrides = ["${basePath}/1.11/src/main/java", "${basePath}/1.12/src/main/java", './src/main/java']
+  test_source_overrides = ["${basePath}/1.11/src/test/java", "${basePath}/1.12/src/test/java", './src/test/java']
   main_resources_overrides = []
   test_resources_overrides = []
   archives_base_name = 'beam-runners-flink-1.13'

runners/flink/flink_runner.gradle

@@ -266,7 +266,7 @@ tasks.register('validatesRunner') {
   dependsOn validatesRunnerStreamingCheckpointing
 }
 
-// Generates :runners:flink:1.10:runQuickstartJavaFlinkLocal
+// Generates :runners:flink:1.13:runQuickstartJavaFlinkLocal
 createJavaExamplesArchetypeValidationTask(type: 'Quickstart', runner: 'FlinkLocal')
 
 /**

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

@@ -122,13 +122,8 @@ PortablePipelineResult runPipelineWithTranslator(
 
   private PortablePipelineResult createPortablePipelineResult(
       JobExecutionResult result, PipelineOptions options) {
-    // The package of DetachedJobExecutionResult has been changed in 1.10.
-    // Refer to https://github.com/apache/flink/commit/c36b35e6876ecdc717dade653e8554f9d8b543c9 for
-    // details.
     String resultClassName = result.getClass().getCanonicalName();
-    if (resultClassName.equals(
-            "org.apache.flink.client.program.DetachedEnvironment.DetachedJobExecutionResult")
-        || resultClassName.equals("org.apache.flink.core.execution.DetachedJobExecutionResult")) {
+    if (resultClassName.equals("org.apache.flink.core.execution.DetachedJobExecutionResult")) {
       LOG.info("Pipeline submitted in Detached mode");
       // no metricsPusher because metrics are not supported in detached mode
       return new FlinkPortableRunnerResult.Detached();

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

@@ -109,13 +109,8 @@ public PipelineResult run(Pipeline pipeline) {
   }
 
   static PipelineResult createPipelineResult(JobExecutionResult result, PipelineOptions options) {
-    // The package of DetachedJobExecutionResult has been changed in 1.10.
-    // Refer to https://github.com/apache/flink/commit/c36b35e6876ecdc717dade653e8554f9d8b543c9 for
-    // more details.
     String resultClassName = result.getClass().getCanonicalName();
-    if (resultClassName.equals(
-            "org.apache.flink.client.program.DetachedEnvironment.DetachedJobExecutionResult")
-        || resultClassName.equals("org.apache.flink.core.execution.DetachedJobExecutionResult")) {
+    if (resultClassName.equals("org.apache.flink.core.execution.DetachedJobExecutionResult")) {
       LOG.info("Pipeline submitted in Detached mode");
       // no metricsPusher because metrics are not supported in detached mode
       return new FlinkDetachedRunnerResult();