[ZEPPELIN-6284] Added Flink 1.20 interpreter.

.github/workflows/core.yml

@@ -212,7 +212,7 @@ jobs:
             ${{ runner.os }}-zeppelin-
       - name: install environment
         run: |
-          ./mvnw install -DskipTests -Pintegration -pl zeppelin-interpreter-integration,zeppelin-web,spark-submit,spark/scala-2.12,spark/scala-2.13,markdown,flink-cmd,flink/flink-scala-2.12,jdbc,shell -am -Pweb-classic -Pflink-117 ${MAVEN_ARGS}
+          ./mvnw install -DskipTests -Pintegration -pl zeppelin-interpreter-integration,zeppelin-web,spark-submit,spark/scala-2.12,spark/scala-2.13,markdown,flink-cmd,flink/flink-scala-2.12,jdbc,shell -am -Pweb-classic -Pflink-120 ${MAVEN_ARGS}
           ./mvnw package -pl zeppelin-plugins -amd -DskipTests ${MAVEN_ARGS}
       - name: Setup conda environment with python 3.9 and R
         uses: conda-incubator/setup-miniconda@v3
@@ -239,7 +239,7 @@ jobs:
       fail-fast: false
       matrix:
         python: [ 3.9 ]
-        flink: [116, 117]
+        flink: [116, 117, 120]
         include:
           # Flink 1.15 supports Python 3.6, 3.7, and 3.8
           # https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/installation/

flink/README.md

@@ -13,6 +13,7 @@ Currently, it has the following modules clustered into two groups:
 * flink1.15-shims
 * flink1.16-shims
 * flink1.17-shims
+* flink1.20-shims
 
 * flink-scala-2.12
 

flink/flink-scala-2.12/pom.xml

@@ -33,7 +33,7 @@
 
   <properties>
     <!--library versions-->
-    <flink.version>${flink1.17.version}</flink.version>
+    <flink.version>${flink1.20.version}</flink.version>
     <flink.scala.version>2.12.7</flink.scala.version>
     <flink.scala.binary.version>2.12</flink.scala.binary.version>
     <flink.scala.compile.version>${flink.scala.version}</flink.scala.compile.version>
@@ -71,6 +71,12 @@
       <version>${project.version}</version>
     </dependency>
 
+    <dependency>
+      <groupId>org.apache.zeppelin</groupId>
+      <artifactId>flink1.20-shims</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
     <dependency>
       <groupId>org.apache.zeppelin</groupId>
       <artifactId>zeppelin-python</artifactId>
@@ -1300,5 +1306,39 @@
         </dependency>
       </dependencies>
     </profile>
+    <profile>
+      <id>flink-120</id>
+      <properties>
+        <flink.version>${flink1.20.version}</flink.version>
+        <flink.scala.version>2.12.7</flink.scala.version>
+        <flink.scala.binary.version>2.12</flink.scala.binary.version>
+      </properties>
+      <dependencies>
+        <dependency>
+          <groupId>org.apache.flink</groupId>
+          <artifactId>flink-runtime</artifactId>
+          <version>${flink.version}</version>
+          <scope>provided</scope>
+        </dependency>
+        <dependency>
+          <groupId>org.apache.flink</groupId>
+          <artifactId>flink-table-planner_${flink.scala.binary.version}</artifactId>
+          <version>${flink.version}</version>
+          <scope>provided</scope>
+        </dependency>
+        <dependency>
+          <groupId>org.apache.flink</groupId>
+          <artifactId>flink-sql-client</artifactId>
+          <version>${flink.version}</version>
+          <scope>provided</scope>
+        </dependency>
+        <dependency>
+          <groupId>org.apache.flink</groupId>
+          <artifactId>flink-python</artifactId>
+          <version>${flink.version}</version>
+          <scope>provided</scope>
+        </dependency>
+      </dependencies>
+    </profile>
   </profiles>
 </project>

flink/flink-scala-2.12/src/main/scala/org/apache/zeppelin/flink/FlinkScalaInterpreter.scala

@@ -426,39 +426,65 @@ abstract class FlinkScalaInterpreter(val properties: Properties,
   }
 
   private def createTableEnvs(): Unit = {
-    val originalClassLoader = Thread.currentThread().getContextClassLoader
+    val originalCl = Thread.currentThread().getContextClassLoader
     try {
-      Thread.currentThread().setContextClassLoader(getFlinkScalaShellLoader)
-      val tableConfig = new TableConfig
-      tableConfig.getConfiguration.addAll(configuration)
-
-      this.tblEnvFactory = new TableEnvFactory(this.flinkVersion, this.flinkShims,
-        this.benv, this.senv, tableConfig, this.userJars.map(new URL(_)).asJava)
+      val shellCl = getFlinkScalaShellLoader
+      Thread.currentThread().setContextClassLoader(shellCl)
 
-      // blink planner
-      val btEnvSetting = this.flinkShims.createBlinkPlannerEnvSettingBuilder()
-        .asInstanceOf[EnvironmentSettings.Builder]
+      // Create settings (no Blink planner in 1.20; this is the unified planner)
+      val btSettings = EnvironmentSettings.newInstance()
         .inBatchMode()
         .build()
-      this.btenv = tblEnvFactory.createJavaBlinkBatchTableEnvironment(btEnvSetting, getFlinkScalaShellLoader);
-      flinkILoop.bind("btenv", btenv.getClass().getCanonicalName(), btenv, List("@transient"))
-      this.java_btenv = this.btenv
 
-      val stEnvSetting = this.flinkShims.createBlinkPlannerEnvSettingBuilder()
-        .asInstanceOf[EnvironmentSettings.Builder]
+      val stSettings = EnvironmentSettings.newInstance()
         .inStreamingMode()
         .build()
-      this.stenv = tblEnvFactory.createScalaBlinkStreamTableEnvironment(stEnvSetting, getFlinkScalaShellLoader)
-      flinkILoop.bind("stenv", stenv.getClass().getCanonicalName(), stenv, List("@transient"))
-      this.java_stenv = tblEnvFactory.createJavaBlinkStreamTableEnvironment(stEnvSetting, getFlinkScalaShellLoader)
-
-      if (!flinkVersion.isAfterFlink114()) {
-        // flink planner is not supported after flink 1.14
-        this.btenv_2 = tblEnvFactory.createScalaFlinkBatchTableEnvironment()
-        this.java_btenv_2 = tblEnvFactory.createJavaFlinkBatchTableEnvironment()
-      }
+
+      // --- Batch TableEnvironment (unified, not Blink/Java/Scala-specific) ---
+      // NOTE: TableConfig ctor is deprecated/removed; configure AFTER creation.
+      val btEnv = org.apache.flink.table.api.TableEnvironment.create(btSettings)
+      btEnv.getConfig.getConfiguration.addAll(configuration)
+      this.btenv = btEnv
+      this.java_btenv = btEnv  // single impl for both Java/Scala APIs
+      flinkILoop.bind("btenv",
+        classOf[org.apache.flink.table.api.TableEnvironment].getCanonicalName,
+        btEnv,
+        List("@transient"))
+
+      // --- Streaming TableEnvironment (Scala & Java bridges) ---
+      // Scala bridge
+      val stEnv =
+        org.apache.flink.table.api.bridge.scala.StreamTableEnvironment.create(this.senv, stSettings)
+      stEnv.getConfig.getConfiguration.addAll(configuration)
+      this.stenv = stEnv
+      flinkILoop.bind("stenv",
+        classOf[org.apache.flink.table.api.bridge.scala.StreamTableEnvironment].getCanonicalName,
+        stEnv,
+        List("@transient"))
+
+
+      // Java StreamExecutionEnvironment from the Scala one
+      val jStreamEnv: org.apache.flink.streaming.api.environment.StreamExecutionEnvironment =
+        this.senv.getJavaEnv   // in Scala this accessor is available as `javaEnv`
+
+      // Java bridge
+      this.java_stenv =
+        org.apache.flink.table.api.bridge.java.StreamTableEnvironment.create(jStreamEnv, stSettings)
+
+
+
+      // --- (Optional) If you previously injected user jars via a custom loader,
+      // consider also setting PipelineOptions.JARS so jobs carry them when submitted:
+      // import org.apache.flink.configuration.PipelineOptions
+      // configuration.set(PipelineOptions.JARS, this.userJars.asJava)
+      // If you do this, set it BEFORE creating the envs so it takes effect.
+
+      // No Flink planner (pre-1.14) fallback in 1.20+
+      this.btenv_2 = null
+      this.java_btenv_2 = null
+
     } finally {
-      Thread.currentThread().setContextClassLoader(originalClassLoader)
+      Thread.currentThread().setContextClassLoader(originalCl)
     }
   }
 

flink/flink-shims/src/main/java/org/apache/zeppelin/flink/FlinkShims.java

@@ -69,7 +69,10 @@ private static FlinkShims loadShims(FlinkVersion flinkVersion,
     } else if (flinkVersion.getMajorVersion() == 1 && flinkVersion.getMinorVersion() == 17) {
       LOGGER.info("Initializing shims for Flink 1.17");
       flinkShimsClass = Class.forName("org.apache.zeppelin.flink.Flink117Shims");
-    } else {
+    } else if (flinkVersion.getMajorVersion() == 1 && flinkVersion.getMinorVersion() == 20) {
+      LOGGER.info("Initializing shims for Flink 1.20");
+      flinkShimsClass = Class.forName("org.apache.zeppelin.flink.Flink120Shims");
+    }else {
       throw new Exception("Flink version: '" + flinkVersion + "' is not supported yet");
     }
 

flink/flink1.20-shims/pom.xml

@@ -0,0 +1,207 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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 xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>flink-parent</artifactId>
+        <groupId>org.apache.zeppelin</groupId>
+        <version>0.13.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+    <groupId>org.apache.zeppelin</groupId>
+    <artifactId>flink1.20-shims</artifactId>
+    <version>0.13.0-SNAPSHOT</version>
+    <packaging>jar</packaging>
+    <name>Zeppelin: Flink1.20 Shims</name>
+
+    <properties>
+        <flink.version>${flink1.20.version}</flink.version>
+        <flink.scala.binary.version>2.12</flink.scala.binary.version>
+    </properties>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.apache.zeppelin</groupId>
+            <artifactId>flink-shims</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-clients</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-runtime</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-api-scala_${flink.scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-api-scala-bridge_${flink.scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-api-java-bridge</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-scala_${flink.scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-scala_${flink.scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-java</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-planner_${flink.scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-python</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-sql-client</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+
+            <plugin>
+                <groupId>net.alchim31.maven</groupId>
+                <artifactId>scala-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>eclipse-add-source</id>
+                        <goals>
+                            <goal>add-source</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>scala-compile-first</id>
+                        <phase>process-resources</phase>
+                        <goals>
+                            <goal>compile</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>scala-test-compile-first</id>
+                        <phase>process-test-resources</phase>
+                        <goals>
+                            <goal>testCompile</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <scalaVersion>${flink.scala.version}</scalaVersion>
+                    <args>
+                        <arg>-unchecked</arg>
+                        <arg>-deprecation</arg>
+                        <arg>-feature</arg>
+                        <arg>-nobootcp</arg>
+                    </args>
+                    <jvmArgs>
+                        <jvmArg>-Xms1024m</jvmArg>
+                        <jvmArg>-Xmx1024m</jvmArg>
+                        <jvmArg>-XX:MaxMetaspaceSize=${MaxMetaspace}</jvmArg>
+                    </jvmArgs>
+                    <javacArgs>
+                        <javacArg>-source</javacArg>
+                        <javacArg>${java.version}</javacArg>
+                        <javacArg>-target</javacArg>
+                        <javacArg>${java.version}</javacArg>
+                        <javacArg>-Xlint:all,-serial,-path,-options</javacArg>
+                    </javacArgs>
+                </configuration>
+            </plugin>
+
+            <plugin>
+                <artifactId>maven-resources-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>copy-interpreter-setting</id>
+                        <phase>none</phase>
+                        <configuration>
+                            <skip>true</skip>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>