[Flink] Drop 1.12–1.16 shims, baseline 1.17+, add 2.x support

.sonarcloud.properties

@@ -0,0 +1,18 @@
+# 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.
+#
+# SonarCloud automatic analysis reads this file (not sonar-project.properties).
+# Multi-version Flink shims intentionally share parallel structure per Flink release.
+sonar.cpd.exclusions=**/streampark-flink-shims/**,**/streampark-flink-shims-base/**,**/streampark-flink-shims-base-v2/**,**/streampark-spark-shims/**

pom.xml

@@ -98,7 +98,7 @@
         <flink.version>1.20.1</flink.version>
         <flink.shaded.version>1.8.1</flink.shaded.version>
         <streampark.shaded.version>1.0.0</streampark.shaded.version>
-        <streampark.flink.shims.version>1.14</streampark.flink.shims.version>
+        <streampark.flink.shims.version>1.17</streampark.flink.shims.version>
         <spark.version>3.1.2</spark.version>
         <scalikejdbc.version>4.0.0</scalikejdbc.version>
         <hive.version>2.3.4</hive.version>
@@ -222,7 +222,7 @@
 
             <dependency>
                 <groupId>org.apache.streampark</groupId>
-                <artifactId>streampark-common_${scala.binary.version}</artifactId>
+                <artifactId>streampark-common</artifactId>
                 <version>${project.version}</version>
             </dependency>
 
@@ -932,6 +932,7 @@
                             <exclude>src/main/resources/*.dict</exclude>
 
                             <exclude>streampark-console-webapp/**</exclude>
+                            <exclude>streampark-console-webapp-v2/**</exclude>
                         </excludes>
                     </configuration>
                     <executions>

streampark-common-scala-bridge/pom.xml

@@ -30,7 +30,7 @@
     <dependencies>
         <dependency>
             <groupId>org.apache.streampark</groupId>
-            <artifactId>streampark-common_${scala.binary.version}</artifactId>
+            <artifactId>streampark-common</artifactId>
         </dependency>
     </dependencies>
 

streampark-common/pom.xml

@@ -24,7 +24,7 @@
         <version>${revision}</version>
     </parent>
 
-    <artifactId>streampark-common_${scala.binary.version}</artifactId>
+    <artifactId>streampark-common</artifactId>
     <name>StreamPark : Common</name>
 
     <dependencies>
@@ -132,8 +132,8 @@
         <plugins>
 
             <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-compiler-plugin</artifactId>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>build-helper-maven-plugin</artifactId>
             </plugin>
 
             <plugin>

streampark-common/src/main/java/org/apache/streampark/common/conf/FlinkVersion.java

@@ -28,6 +28,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.function.Consumer;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -43,8 +44,11 @@ public class FlinkVersion implements Serializable {
     private static final Pattern FLINK_VER_PATTERN = Pattern.compile("^(\\d+\\.\\d+)(?:\\..*)?$");
     private static final Pattern FLINK_VERSION_PATTERN =
         Pattern.compile("^Version: ([^,]*), Commit ID: (.*)$");
-    private static final Pattern FLINK_SCALA_VERSION_PATTERN =
-        Pattern.compile("^flink-dist_(\\d+\\.\\d+)[^/\\\\]*\\.jar$");
+    private static final Pattern FLINK_DIST_UNDERSCORE_PATTERN =
+        Pattern.compile(
+            "^flink-dist_(\\d+\\.\\d+)-(\\d+\\.\\d+(?:\\.\\d+)?(?:-SNAPSHOT)?)\\.jar$");
+    private static final Pattern FLINK_DIST_DASH_PATTERN =
+        Pattern.compile("^flink-dist-(\\d+\\.\\d+(?:\\.\\d+)?(?:-SNAPSHOT)?)\\.jar$");
     private static final Pattern APACHE_FLINK_VERSION_PATTERN =
         Pattern.compile("^(\\d+\\.\\d+\\.\\d+)");
     private static final Pattern OTHER_FLINK_VERSION_PATTERN = Pattern.compile("^(\\d+\\.\\d+)-?$");
@@ -84,10 +88,14 @@ public String version() {
         return getVersion();
     }
 
+    /** Backward-compatible alias for {@link #getFlinkLibs()}. */
+    public List<URL> flinkLibs() throws Exception {
+        return getFlinkLibs();
+    }
+
     public String getScalaVersion() {
         if (scalaVersion == null) {
-            Matcher matcher = FLINK_SCALA_VERSION_PATTERN.matcher(getFlinkDistJar().getName());
-            scalaVersion = matcher.matches() ? matcher.group(1) : "2.12";
+            scalaVersion = parseFromDistJar().map(pair -> pair[1]).orElse("2.12");
         }
         return scalaVersion;
     }
@@ -130,41 +138,10 @@ public List<URL> getFlinkLibs() throws Exception {
 
     public String getVersion() {
         if (version == null) {
-            List<String> cmd =
-                Arrays.asList(
-                    "java -classpath "
-                        + getFlinkDistJar().getName()
-                        + " org.apache.flink.client.cli.CliFrontend --version");
-            StringBuilder buffer = new StringBuilder();
-            final String[] flinkVersion = {null};
-            try {
-                CommandUtils.execute(
-                    getFlinkLib().getAbsolutePath(),
-                    cmd,
-                    out -> {
-                        buffer.append(out).append("\n");
-                        Matcher matcher = FLINK_VERSION_PATTERN.matcher(out);
-                        if (matcher.find()) {
-                            String ver = matcher.group(1);
-                            Matcher m1 = APACHE_FLINK_VERSION_PATTERN.matcher(ver);
-                            if (m1.find()) {
-                                flinkVersion[0] = ver;
-                            } else {
-                                Matcher m2 = OTHER_FLINK_VERSION_PATTERN.matcher(ver);
-                                if (m2.find()) {
-                                    flinkVersion[0] = ver;
-                                }
-                            }
-                        }
-                    });
-            } catch (Exception e) {
-                throw new IllegalStateException("[StreamPark] execute flink version command failed", e);
-            }
-            LOG.info("[StreamPark] {}", buffer);
-            if (flinkVersion[0] == null) {
-                throw new IllegalStateException("[StreamPark] parse flink version failed. " + buffer);
-            }
-            version = flinkVersion[0];
+            version =
+                parseFromDistJar()
+                    .map(pair -> pair[0])
+                    .orElseGet(this::parseFromCliFrontend);
         }
         return version;
     }
@@ -205,10 +182,14 @@ public boolean checkVersion() {
 
     public boolean checkVersion(boolean throwException) {
         String[] parts = getVersion().split("\\.");
-        if (parts.length >= 2 && "1".equals(parts[0].trim())) {
+        if (parts.length >= 2) {
             try {
+                int major = Integer.parseInt(parts[0].trim());
                 int minor = Integer.parseInt(parts[1].trim());
-                if (minor >= 12 && minor <= 20) {
+                if (major == 1 && minor >= 17 && minor <= 20) {
+                    return true;
+                }
+                if (major == 2 && minor >= 0 && minor <= 3) {
                     return true;
                 }
             } catch (NumberFormatException ignored) {
@@ -222,15 +203,88 @@ public boolean checkVersion(boolean throwException) {
 
     public boolean checkVersion(int sinceVersion) {
         String[] parts = getVersion().split("\\.");
-        if (parts.length >= 2 && "1".equals(parts[0].trim())) {
+        if (parts.length >= 2) {
             try {
-                return Integer.parseInt(parts[1].trim()) >= sinceVersion;
+                int major = Integer.parseInt(parts[0].trim());
+                int minor = Integer.parseInt(parts[1].trim());
+                if (major == 1 && minor >= sinceVersion) {
+                    return true;
+                }
+                if (major == 2) {
+                    return true;
+                }
             } catch (NumberFormatException ignored) {
             }
         }
         return false;
     }
 
+    private java.util.Optional<String[]> parseFromDistJar() {
+        String jarName = getFlinkDistJar().getName();
+        Matcher underscoreMatcher = FLINK_DIST_UNDERSCORE_PATTERN.matcher(jarName);
+        if (underscoreMatcher.matches()) {
+            String parsedVersion = underscoreMatcher.group(2);
+            String parsedScala = underscoreMatcher.group(1);
+            LOG.info(
+                "Flink version parsed from dist jar name: {}, scala: {}",
+                parsedVersion,
+                parsedScala);
+            return java.util.Optional.of(new String[]{parsedVersion, parsedScala});
+        }
+        Matcher dashMatcher = FLINK_DIST_DASH_PATTERN.matcher(jarName);
+        if (dashMatcher.matches()) {
+            String parsedVersion = dashMatcher.group(1);
+            LOG.info(
+                "Flink version parsed from dist jar name: {}, scala: {}",
+                parsedVersion,
+                "2.12");
+            return java.util.Optional.of(new String[]{parsedVersion, "2.12"});
+        }
+        return java.util.Optional.empty();
+    }
+
+    private String parseFromCliFrontend() {
+        final String[] flinkVersion = {null};
+        StringBuilder buffer = new StringBuilder();
+        List<String> cmd =
+            Arrays.asList(
+                "java -classpath "
+                    + getFlinkDistJar().getName()
+                    + " org.apache.flink.client.cli.CliFrontend --version");
+        try {
+            CommandUtils.execute(
+                getFlinkLib().getAbsolutePath(),
+                cmd,
+                new Consumer<String>() {
+
+                    @Override
+                    public void accept(String out) {
+                        buffer.append(out).append("\n");
+                        Matcher matcher = FLINK_VERSION_PATTERN.matcher(out);
+                        if (matcher.find()) {
+                            String ver = matcher.group(1);
+                            Matcher m1 = APACHE_FLINK_VERSION_PATTERN.matcher(ver);
+                            if (m1.find()) {
+                                flinkVersion[0] = ver;
+                            } else {
+                                Matcher m2 = OTHER_FLINK_VERSION_PATTERN.matcher(ver);
+                                if (m2.find()) {
+                                    flinkVersion[0] = ver;
+                                }
+                            }
+                        }
+                    }
+                });
+        } catch (Exception e) {
+            throw new IllegalStateException("[StreamPark] execute flink version command failed", e);
+        }
+        LOG.info("[StreamPark] {}", buffer);
+        if (flinkVersion[0] == null) {
+            throw new IllegalStateException("[StreamPark] parse flink version failed. " + buffer);
+        }
+        return flinkVersion[0];
+    }
+
     @Override
     public String toString() {
         return "\n----------------------------------------- flink version -----------------------------------\n"