Add support for Flink 1.19

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

@@ -14,7 +14,7 @@
   },
   "JavaTestProperties": {
     "SUPPORTED_VERSIONS": ["8", "11", "17", "21"],
-    "FLINK_VERSIONS": ["1.15", "1.16", "1.17", "1.18"],
+    "FLINK_VERSIONS": ["1.15", "1.16", "1.17", "1.18", "1.19"],
     "SPARK_VERSIONS": ["2", "3"]
   },
   "GoTestProperties": {

.github/trigger_files/beam_PostCommit_Go_VR_Flink.json

@@ -1,4 +1,5 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 1
+  "modification": 1,
+  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
 }

.github/trigger_files/beam_PostCommit_Java_Examples_Flink.json

@@ -0,0 +1,3 @@
+{
+    "https://github.com/apache/beam/pull/32648": "testing flink 1.19 support"
+}

.github/trigger_files/beam_PostCommit_Java_Jpms_Flink_Java11.json

@@ -0,0 +1,3 @@
+{
+    "https://github.com/apache/beam/pull/32648": "testing flink 1.19 support"
+}

.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Flink.json

@@ -1,4 +1,5 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "https://github.com/apache/beam/pull/31156": "noting that PR #31156 should run this test"
+  "https://github.com/apache/beam/pull/31156": "noting that PR #31156 should run this test",
+  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
 }

.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Flink_Java11.json

@@ -1,4 +1,5 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "https://github.com/apache/beam/pull/31156": "noting that PR #31156 should run this test"
+  "https://github.com/apache/beam/pull/31156": "noting that PR #31156 should run this test",
+  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
 }

.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Flink_Java8.json

@@ -0,0 +1,4 @@
+{
+
+  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
+}

.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Flink.json

@@ -0,0 +1,3 @@
+{
+  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
+}

.github/trigger_files/beam_PostCommit_XVR_Flink.json

@@ -0,0 +1,3 @@
+{
+  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
+}

.github/workflows/beam_PostCommit_Java_Tpcds_Flink.yml

@@ -101,5 +101,5 @@ jobs:
         with:
           gradle-command: :sdks:java:testing:tpcds:run
           arguments: |
-            -Ptpcds.runner=:runners:flink:1.18 \
+            -Ptpcds.runner=:runners:flink:1.19 \
             "-Ptpcds.args=${{env.tpcdsBigQueryArgs}} ${{env.tpcdsInfluxDBArgs}} ${{ env.GRADLE_COMMAND_ARGUMENTS }} --queries=${{env.tpcdsQueriesArg}}" \

.github/workflows/beam_PostCommit_Java_ValidatesRunner_Flink.yml

@@ -78,7 +78,7 @@ jobs:
       - name: run validatesRunner script
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
-          gradle-command: :runners:flink:1.18:validatesRunner
+          gradle-command: :runners:flink:1.19:validatesRunner
       - name: Archive JUnit Test Results
         uses: actions/upload-artifact@v4
         if: ${{ !success() }}

CHANGES.md

@@ -25,6 +25,8 @@
 * New highly anticipated feature X added to Python SDK ([#X](https://github.com/apache/beam/issues/X)).
 * New highly anticipated feature Y added to Java SDK ([#Y](https://github.com/apache/beam/issues/Y)).
 
+* Added support for Flink 1.19
+
 ## I/Os
 
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).

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

@@ -637,6 +637,7 @@ class BeamModulePlugin implements Plugin<Project> {
     def sbe_tool_version = "1.25.1"
     def singlestore_jdbc_version = "1.1.4"
     def slf4j_version = "1.7.30"
+    def snakeyaml_engine_version = "2.6"
     def snakeyaml_version = "2.2"
     def solace_version = "10.21.0"
     def spark2_version = "2.4.8"
@@ -870,6 +871,7 @@ class BeamModulePlugin implements Plugin<Project> {
         singlestore_jdbc                            : "com.singlestore:singlestore-jdbc-client:$singlestore_jdbc_version",
         slf4j_api                                   : "org.slf4j:slf4j-api:$slf4j_version",
         snake_yaml                                  : "org.yaml:snakeyaml:$snakeyaml_version",
+        snakeyaml_engine                            : "org.snakeyaml:snakeyaml-engine:$snakeyaml_engine_version",
         slf4j_android                               : "org.slf4j:slf4j-android:$slf4j_version",
         slf4j_ext                                   : "org.slf4j:slf4j-ext:$slf4j_version",
         slf4j_jdk14                                 : "org.slf4j:slf4j-jdk14:$slf4j_version",

contributor-docs/release-guide.md

@@ -887,7 +887,7 @@ write to BigQuery, and create a cluster of machines for running containers (for
   ```
   **Flink Local Runner**
   ```
-  ./gradlew :runners:flink:1.18:runQuickstartJavaFlinkLocal \
+  ./gradlew :runners:flink:1.19:runQuickstartJavaFlinkLocal \
   -Prepourl=https://repository.apache.org/content/repositories/orgapachebeam-${KEY} \
   -Pver=${RELEASE_VERSION}
   ```

gradle.properties

@@ -39,6 +39,6 @@ docker_image_default_repo_root=apache
 docker_image_default_repo_prefix=beam_
 
 # supported flink versions
-flink_versions=1.15,1.16,1.17,1.18
+flink_versions=1.15,1.16,1.17,1.18,1.19
 # supported python versions
 python_versions=3.8,3.9,3.10,3.11,3.12

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.18:runQuickstartJavaFlinkLocal")
+  dependsOn(":runners:flink:1.19:runQuickstartJavaFlinkLocal")
   dependsOn(":runners:direct-java:runMobileGamingJavaDirect")
   dependsOn(":runners:google-cloud-dataflow-java:runMobileGamingJavaDataflow")
   dependsOn(":runners:twister2:runQuickstartJavaTwister2")

runners/flink/1.15/src/test/java/org/apache/beam/runners/flink/streaming/MemoryStateBackendWrapper.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.flink.streaming;
+
+import java.util.Collection;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.query.TaskKvStateRegistry;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.BackendBuildingException;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyedStateHandle;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
+
+class MemoryStateBackendWrapper {
+  static <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
+      Environment env,
+      JobID jobID,
+      String operatorIdentifier,
+      TypeSerializer<K> keySerializer,
+      int numberOfKeyGroups,
+      KeyGroupRange keyGroupRange,
+      TaskKvStateRegistry kvStateRegistry,
+      TtlTimeProvider ttlTimeProvider,
+      MetricGroup metricGroup,
+      Collection<KeyedStateHandle> stateHandles,
+      CloseableRegistry cancelStreamRegistry)
+      throws BackendBuildingException {
+
+    MemoryStateBackend backend = new MemoryStateBackend();
+    return backend.createKeyedStateBackend(
+        env,
+        jobID,
+        operatorIdentifier,
+        keySerializer,
+        numberOfKeyGroups,
+        keyGroupRange,
+        kvStateRegistry,
+        ttlTimeProvider,
+        metricGroup,
+        stateHandles,
+        cancelStreamRegistry);
+  }
+
+  static OperatorStateBackend createOperatorStateBackend(
+      Environment env,
+      String operatorIdentifier,
+      Collection<OperatorStateHandle> stateHandles,
+      CloseableRegistry cancelStreamRegistry)
+      throws Exception {
+    MemoryStateBackend backend = new MemoryStateBackend();
+    return backend.createOperatorStateBackend(
+        env, operatorIdentifier, stateHandles, cancelStreamRegistry);
+  }
+}

runners/flink/1.19/build.gradle

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an AS IS BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+project.ext {
+  flink_major = '1.19'
+  flink_version = '1.19.0'
+}
+
+// Load the main build script which contains all build logic.
+apply from: "../flink_runner.gradle"

runners/flink/1.19/job-server-container/build.gradle

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an AS IS BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+def basePath = '../../job-server-container'
+
+project.ext {
+  resource_path = basePath
+}
+
+// Load the main build script which contains all build logic.
+apply from: "$basePath/flink_job_server_container.gradle"

runners/flink/1.19/job-server/build.gradle

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an AS IS BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+def basePath = '../../job-server'
+
+project.ext {
+  // Look for the source code in the parent module
+  main_source_dirs = ["$basePath/src/main/java"]
+  test_source_dirs = ["$basePath/src/test/java"]
+  main_resources_dirs = ["$basePath/src/main/resources"]
+  test_resources_dirs = ["$basePath/src/test/resources"]
+  archives_base_name = 'beam-runners-flink-1.19-job-server'
+}
+
+// Load the main build script which contains all build logic.
+apply from: "$basePath/flink_job_server.gradle"

runners/flink/1.19/src/test/java/org/apache/beam/runners/flink/streaming/MemoryStateBackendWrapper.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.flink.streaming;
+
+import java.util.Collection;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.query.TaskKvStateRegistry;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.BackendBuildingException;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyedStateBackendParametersImpl;
+import org.apache.flink.runtime.state.KeyedStateHandle;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateBackendParametersImpl;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
+
+class MemoryStateBackendWrapper {
+  static <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
+      Environment env,
+      JobID jobID,
+      String operatorIdentifier,
+      TypeSerializer<K> keySerializer,
+      int numberOfKeyGroups,
+      KeyGroupRange keyGroupRange,
+      TaskKvStateRegistry kvStateRegistry,
+      TtlTimeProvider ttlTimeProvider,
+      MetricGroup metricGroup,
+      Collection<KeyedStateHandle> stateHandles,
+      CloseableRegistry cancelStreamRegistry)
+      throws BackendBuildingException {
+
+    MemoryStateBackend backend = new MemoryStateBackend();
+    return backend.createKeyedStateBackend(
+        new KeyedStateBackendParametersImpl<>(
+            env,
+            jobID,
+            operatorIdentifier,
+            keySerializer,
+            numberOfKeyGroups,
+            keyGroupRange,
+            kvStateRegistry,
+            ttlTimeProvider,
+            metricGroup,
+            stateHandles,
+            cancelStreamRegistry));
+  }
+
+  static OperatorStateBackend createOperatorStateBackend(
+      Environment env,
+      String operatorIdentifier,
+      Collection<OperatorStateHandle> stateHandles,
+      CloseableRegistry cancelStreamRegistry)
+      throws Exception {
+    MemoryStateBackend backend = new MemoryStateBackend();
+    return backend.createOperatorStateBackend(
+        new OperatorStateBackendParametersImpl(
+            env, operatorIdentifier, stateHandles, cancelStreamRegistry));
+  }
+}