[BEAM-10955] Update Flink minor versions and enable testSavepointRest…

CHANGES.md

@@ -45,6 +45,33 @@
 
 * Fixed X (Java/Python) ([BEAM-X](https://issues.apache.org/jira/browse/BEAM-X)).
 -->
+# [2.33.X] - Unreleased
+
+## Highlights
+
+* New highly anticipated feature X added to Python SDK ([BEAM-X](https://issues.apache.org/jira/browse/BEAM-X)).
+* New highly anticipated feature Y added to Java SDK ([BEAM-Y](https://issues.apache.org/jira/browse/BEAM-Y)).
+
+## I/Os
+
+* Support for X source added (Java/Python) ([BEAM-X](https://issues.apache.org/jira/browse/BEAM-X)).
+
+## New Features / Improvements
+
+* X feature added (Java/Python) ([BEAM-X](https://issues.apache.org/jira/browse/BEAM-X)).
+* Upgrade Flink runner to Flink versions 1.13.2, 1.12.5 and 1.11.4 ([BEAM-10955](https://issues.apache.org/jira/browse/BEAM-10955)).
+
+## Breaking Changes
+
+* X behavior was changed ([BEAM-X](https://issues.apache.org/jira/browse/BEAM-X)).
+
+## Deprecations
+
+* X behavior is deprecated and will be removed in X versions ([BEAM-X](https://issues.apache.org/jira/browse/BEAM-X)).
+
+## Known Issues
+
+* Fixed X (Java/Python) ([BEAM-X](https://issues.apache.org/jira/browse/BEAM-X)).
 
 # [2.32.X] - Unreleased
 

runners/flink/1.11/build.gradle

@@ -20,7 +20,7 @@ def basePath = '..'
 /* All properties required for loading the Flink build script */
 project.ext {
   // Set the version of all Flink-related dependencies here.
-  flink_version = '1.11.3'
+  flink_version = '1.11.4'
   // Version specific code overrides.
   main_source_overrides = ['./src/main/java']
   test_source_overrides = ['./src/test/java']

runners/flink/1.12/build.gradle

@@ -20,7 +20,7 @@ def basePath = '..'
 /* All properties required for loading the Flink build script */
 project.ext {
   // Set the version of all Flink-related dependencies here.
-  flink_version = '1.12.4'
+  flink_version = '1.12.5'
   // Version specific code overrides.
   main_source_overrides = ["${basePath}/1.11/src/main/java", './src/main/java']
   test_source_overrides = ["${basePath}/1.11/src/test/java", './src/test/java']

runners/flink/1.13/build.gradle

@@ -20,7 +20,7 @@ def basePath = '..'
 /* All properties required for loading the Flink build script */
 project.ext {
   // Set the version of all Flink-related dependencies here.
-  flink_version = '1.13.1'
+  flink_version = '1.13.2'
   // Version specific code overrides.
   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']

runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSavepointTest.java

@@ -18,6 +18,7 @@
 package org.apache.beam.runners.flink;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assume.assumeFalse;
 
 import java.io.Serializable;
 import java.net.URI;
@@ -61,14 +62,14 @@
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 import org.apache.flink.runtime.minicluster.MiniCluster;
 import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
+import org.apache.flink.runtime.util.EnvironmentInformation;
 import org.hamcrest.Matchers;
 import org.hamcrest.core.IsIterableContaining;
 import org.joda.time.Instant;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
@@ -142,9 +143,10 @@ public void afterTest() throws Exception {
     ensureNoJobRunning();
   }
 
-  @Ignore("https://issues.apache.org/jira/projects/BEAM/issues/BEAM-10955")
   @Test
   public void testSavepointRestoreLegacy() throws Exception {
+    // Don't run on Flink 1.11. https://issues.apache.org/jira/browse/BEAM-10955
+    assumeFalse(EnvironmentInformation.getVersion().startsWith("1.11"));
     runSavepointAndRestore(false);
   }