[cherry-pick] add support for flink 1.18 for Beam Flink Runner

.github/actions/setup-default-test-properties/test-properties.json

@@ -13,8 +13,8 @@
     "TOX_ENV": ["Cloud", "Cython"]
   },
   "JavaTestProperties": {
-    "SUPPORTED_VERSIONS": ["8", "11", "17"],
-    "FLINK_VERSIONS": ["1.13", "1.14", "1.15"],
+    "SUPPORTED_VERSIONS": ["8", "11", "17", "21"],
+    "FLINK_VERSIONS": ["1.15", "1.16", "1.17", "1.18"],
     "SPARK_VERSIONS": ["2", "3"]
   },
   "GoTestProperties": {

.github/workflows/li_runners_flink_tests.yml

@@ -30,4 +30,4 @@ jobs:
         settings-path: ${{ github.workspace }} # location for the settings.xml file
 
     - name: Validate Flink runner # Point to the latest
-      run: ./gradlew :runners:flink:build :runners:flink:1.16:build :runners:flink:1.16:validatesRunner
+      run: ./gradlew :runners:flink:build :runners:flink:1.18:build :runners:flink:1.18:validatesRunner

.test-infra/jenkins/job_PreCommit_Java_InfluxDb_IO_Direct.groovy

@@ -16,28 +16,48 @@
  * limitations under the License.
  */
 
-import PrecommitJobBuilder
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.InternalTimeServiceManager;
+import org.apache.flink.streaming.api.operators.InternalTimeServiceManagerImpl;
+import org.apache.flink.streaming.api.operators.sorted.state.BatchExecutionInternalTimeServiceManager;
 
-PrecommitJobBuilder builder = new PrecommitJobBuilder(
-    scope: this,
-    nameBase: 'Java_InfluxDb_IO_Direct',
-    gradleTasks: [
-      ':sdks:java:io:influxdb:build',
-    ],
-    gradleSwitches: [
-      '-PdisableSpotlessCheck=true',
-      '-PdisableCheckStyle=true'
-    ], // spotless checked in separate pre-commit
-    triggerPathPatterns: [
-      '^sdks/java/core/src/main/.*$',
-      '^sdks/java/io/common/.*$',
-      '^sdks/java/io/influxdb/.*$',
-    ],
-    timeoutMins: 60,
-    )
-builder.build {
-  publishers {
-    archiveJunit('**/build/test-results/**/*.xml')
+/** Compatibility layer for {@link AbstractStreamOperator} breaking changes. */
+public abstract class AbstractStreamOperatorCompat<OutputT>
+    extends AbstractStreamOperator<OutputT> {
+
+  /**
+   * Getter for timeServiceManager, which has been made private in Flink 1.11.
+   *
+   * @return Time service manager.
+   */
+  protected InternalTimeServiceManager<?> getTimeServiceManagerCompat() {
+    return getTimeServiceManager()
+        .orElseThrow(() -> new IllegalStateException("Time service manager is not set."));
+  }
+
+  /**
+   * This call has been removed from {@link AbstractStreamOperator} in Flink 1.12.
+   *
+   * <p>{@link InternalTimeServiceManagerImpl#numProcessingTimeTimers()}
+   */
+  protected int numProcessingTimeTimers() {
+    return getTimeServiceManager()
+        .map(
+            manager -> {
+              InternalTimeServiceManager<?> tsm = getTimeServiceManagerCompat();
+              if (tsm instanceof InternalTimeServiceManagerImpl) {
+                final InternalTimeServiceManagerImpl<?> cast =
+                    (InternalTimeServiceManagerImpl<?>) getTimeServiceManagerCompat();
+                return cast.numProcessingTimeTimers();
+              } else if (tsm instanceof BatchExecutionInternalTimeServiceManager) {
+                return 0;
+              } else {
+                throw new IllegalStateException(
+                    String.format(
+                        "Unknown implementation of InternalTimerServiceManager. %s", tsm));
+              }
+            })
+        .orElse(0);
   }
 
   /** Release all of the operator's resources. */

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

@@ -398,7 +398,7 @@ class BeamModulePlugin implements Plugin<Project> {
 
     // Automatically use the official release version if we are performing a release
     // otherwise append '-SNAPSHOT'
-    project.version = '2.45.29'
+    project.version = '2.45.30'
     if (isLinkedin(project)) {
       project.ext.mavenGroupId = 'com.linkedin.beam'
     }

gradle.properties

@@ -30,13 +30,15 @@ signing.gnupg.useLegacyGpg=true
 # buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy.
 # To build a custom Beam version make sure you change it in both places, see
 # https://github.com/apache/beam/issues/21302.
-version=2.45.29
-sdk_version=2.45.29
+version=2.45.30
+sdk_version=2.45.30
 
 javaVersion=1.8
 
 docker_image_default_repo_root=apache
 docker_image_default_repo_prefix=beam_
 
-flink_versions=1.12,1.13,1.14,1.15,1.16
-
+# supported flink versions
+flink_versions=1.15,1.16,1.17,1.18
+# supported python versions
+python_versions=3.8,3.9,3.10,3.11

learning/tour-of-beam/learning-content/java/introduction/introduction-concepts/runner-concepts/description.md

@@ -111,7 +111,7 @@ Additionally, you can read [here](https://beam.apache.org/documentation/runners/
 #### Run example
 
 ##### Portable
-1. Starting with Beam 2.18.0, pre-built Flink Job Service Docker images are available at Docker Hub: `Flink 1.10`, `Flink 1.11`, `Flink 1.12`, `Flink 1.13`, `Flink 1.14`.
+1. Starting with Beam 2.18.0, pre-built Flink Job Service Docker images are available at Docker Hub: `Flink 1.15`, `Flink 1.16`, `Flink 1.17`, `Flink 1.18`.
 2. Start the JobService endpoint: `docker run --net=host apache/beam_flink1.10_job_server:latest`
 3. Submit the pipeline to the above endpoint by using the PortableRunner, job_endpoint set to localhost:8099 (this is the default address of the JobService). Optionally set environment_type set to LOOPBACK. For example:
 

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:3:runQuickstartJavaSpark")
-  dependsOn(":runners:flink:1.13:runQuickstartJavaFlinkLocal")
+  dependsOn(":runners:flink:1.18:runQuickstartJavaFlinkLocal")
   dependsOn(":runners:direct-java:runMobileGamingJavaDirect")
   dependsOn(":runners:google-cloud-dataflow-java:runMobileGamingJavaDataflow")
   dependsOn(":runners:twister2:runQuickstartJavaTwister2")