[ZEPPELIN-6086] Remove Spark Shims and unofficial support for Spark 4.0

### What is this PR for?

Add support for Spark 4 by removing the spark shims. There is only one version of shims now, and they are compatible with the current Spark 4.0 preview releases, so we can just remove the shim concept.

### What type of PR is it?
Improvement

### Todos
* [x] - Disable 4.0 tests in CI and re-enable supported version check

### What is the Jira issue?

ZEPPELIN-6086

### How should this be tested?
A CI stage is added for Spark 4.0 tests, but these will need to be disabled until 4.0 is officially supported.

### Screenshots (if appropriate)

### Questions:
* Does the license files need to update? No
* Is there breaking changes for older versions? No
* Does this needs documentation? Maybe


Closes #4825 from Kimahriman/spark-4.0.

Signed-off-by: Cheng Pan <[email protected]>

Author: Kimahriman

.gitignore

@@ -142,3 +142,6 @@ tramp
 
 # pyenv file
 .python-version
+
+# python venv
+.venv

spark/README.md

@@ -16,9 +16,7 @@ Spark interpreter is the first and most important interpreter of Zeppelin. It su
   - Scala module for Scala 2.12
 * scala-2.13
   - Scala module for Scala 2.13
-* spark-shims 
-  - Parent module for each Spark module
-* spark3-shims
-  - Shims module for Spark3
+* spark-common
+  - Common utils for all Scala versions
 
 

spark/interpreter/pom.xml

@@ -59,12 +59,14 @@
     <!-- settings -->
     <pyspark.test.exclude>**/PySparkInterpreterMatplotlibTest.java</pyspark.test.exclude>
     <pyspark.test.include>**/*Test.*</pyspark.test.include>
+
+    <extraJavaTestArgs></extraJavaTestArgs>
   </properties>
 
   <dependencies>
     <dependency>
       <groupId>org.apache.zeppelin</groupId>
-      <artifactId>spark3-shims</artifactId>
+      <artifactId>spark-common</artifactId>
       <version>${project.version}</version>
     </dependency>
 
@@ -162,6 +164,13 @@
       <artifactId>spark-core_${spark.scala.binary.version}</artifactId>
       <version>${spark.version}</version>
       <scope>provided</scope>
+      <exclusions>
+        <!-- Leads to conflicting Jackson versions in tests -->
+        <exclusion>
+          <groupId>org.apache.avro</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
 
     <dependency>
@@ -313,7 +322,7 @@
               <target>
                 <delete dir="../../interpreter/spark/pyspark" />
                 <copy file="${project.build.directory}/${spark.archive}/python/lib/py4j-${py4j.version}-src.zip" todir="${project.build.directory}/../../../interpreter/spark/pyspark" />
-                <zip basedir="${project.build.directory}/${spark.archive}/python" destfile="${project.build.directory}/../../../interpreter/spark/pyspark/pyspark.zip" includes="pyspark/*.py,pyspark/**/*.py" />
+                <zip basedir="${project.build.directory}/${spark.archive}/python" destfile="${project.build.directory}/../../../interpreter/spark/pyspark/pyspark.zip" includes="pyspark/" />
               </target>
             </configuration>
           </execution>
@@ -359,7 +368,7 @@
         <configuration>
           <forkCount>1</forkCount>
           <reuseForks>false</reuseForks>
-          <argLine>-Xmx3072m -XX:MaxMetaspaceSize=256m</argLine>
+          <argLine>-Xmx3072m -XX:MaxMetaspaceSize=256m ${extraJavaTestArgs}</argLine>
           <excludes>
             <exclude>${pyspark.test.exclude}</exclude>
             <exclude>${tests.to.exclude}</exclude>
@@ -513,6 +522,35 @@
 
   <profiles>
 
+    <profile>
+      <id>java-17</id>
+      <activation>
+        <jdk>[17,)</jdk>
+      </activation>
+      <properties>
+        <extraJavaTestArgs>
+          -XX:+IgnoreUnrecognizedVMOptions
+          --add-modules=jdk.incubator.vector
+          --add-opens=java.base/java.lang=ALL-UNNAMED
+          --add-opens=java.base/java.lang.invoke=ALL-UNNAMED
+          --add-opens=java.base/java.lang.reflect=ALL-UNNAMED
+          --add-opens=java.base/java.io=ALL-UNNAMED
+          --add-opens=java.base/java.net=ALL-UNNAMED
+          --add-opens=java.base/java.nio=ALL-UNNAMED
+          --add-opens=java.base/java.util=ALL-UNNAMED
+          --add-opens=java.base/java.util.concurrent=ALL-UNNAMED
+          --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED
+          --add-opens=java.base/jdk.internal.ref=ALL-UNNAMED
+          --add-opens=java.base/sun.nio.ch=ALL-UNNAMED
+          --add-opens=java.base/sun.nio.cs=ALL-UNNAMED
+          --add-opens=java.base/sun.security.action=ALL-UNNAMED
+          --add-opens=java.base/sun.util.calendar=ALL-UNNAMED
+          -Djdk.reflect.useDirectMethodHandle=false
+          -Dio.netty.tryReflectionSetAccessible=true
+        </extraJavaTestArgs>
+      </properties>
+    </profile>
+
     <!-- profile spark-scala-x only affect the unit test in spark/interpreter module -->
 
     <profile>
@@ -534,6 +572,16 @@
       </properties>
     </profile>
 
+    <profile>
+      <id>spark-4.0</id>
+      <properties>
+        <spark.version>4.0.0-preview2</spark.version>
+        <protobuf.version>3.21.12</protobuf.version>
+        <py4j.version>0.10.9.7</py4j.version>
+        <libthrift.version>0.16.0</libthrift.version>
+      </properties>
+    </profile>
+
     <!-- profile spark-x only affect spark version used in test -->
     <profile>
       <id>spark-3.5</id>

spark/spark-shims/src/main/java/org/apache/zeppelin/spark/SparkVersion.java renamed to spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkVersion.java

@@ -19,10 +19,10 @@
 
 from py4j.java_gateway import java_import
 from pyspark.conf import SparkConf
-from pyspark.context import SparkContext
+from pyspark import SparkContext
 
 # for back compatibility
-from pyspark.sql import SQLContext, Row
+from pyspark.sql import SQLContext
 
 intp = gateway.entry_point
 

spark/interpreter/src/main/resources/python/zeppelin_pyspark.py

@@ -592,7 +592,6 @@ public void tearDown() throws InterpreterException {
     if (this.interpreter != null) {
       this.interpreter.close();
     }
-    SparkShims.reset();
   }
 
   private InterpreterContext getInterpreterContext() {

spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkInterpreterTest.java

@@ -37,7 +37,7 @@
 import org.junit.jupiter.params.provider.CsvSource;
 import org.mockito.ArgumentCaptor;
 
-class SparkShimsTest {
+class SparkUtilsTest {
 
   @ParameterizedTest
   @CsvSource({"2.6.0, false",
@@ -64,29 +64,12 @@ class SparkShimsTest {
   "3.0.0-alpha4, true", // The latest fixed version
   "3.0.1, true"}) // Future versions
   void checkYarnVersionTest(String version, boolean expected) {
-      SparkShims sparkShims =
-          new SparkShims(new Properties()) {
-            @Override
-            public void setupSparkListener(String master,
-                                           String sparkWebUrl,
-                                           InterpreterContext context) {}
-
-            @Override
-            public String showDataFrame(Object obj, int maxResult, InterpreterContext context) {
-              return null;
-            }
-
-            @Override
-            public Object getAsDataFrame(String value) {
-              return null;
-            }
-          };
-      assertEquals(expected, sparkShims.supportYarn6615(version));
+      assertEquals(expected, SparkUtils.supportYarn6615(version));
   }
 
   @Nested
   class SingleTests {
-    SparkShims sparkShims;
+    SparkUtils sparkUtils;
     InterpreterContext mockContext;
     RemoteInterpreterEventClient mockIntpEventClient;
 
@@ -96,18 +79,14 @@ public void setUp() {
       mockIntpEventClient = mock(RemoteInterpreterEventClient.class);
       when(mockContext.getIntpEventClient()).thenReturn(mockIntpEventClient);
 
-      try {
-        sparkShims = SparkShims.getInstance(SparkVersion.SPARK_3_3_0.toString(), new Properties(), null);
-      } catch (Throwable e1) {
-        throw new RuntimeException("All SparkShims are tried, but no one can be created.");
-      }
+      sparkUtils = new SparkUtils(new Properties(), null);
     }
 
     @Test
     void runUnderLocalTest() {
       Properties properties = new Properties();
       properties.setProperty("spark.jobGroup.id", "zeppelin|user1|noteId|paragraphId");
-      sparkShims.buildSparkJobUrl("local", "http://sparkurl", 0, properties, mockContext);
+      sparkUtils.buildSparkJobUrl("local", "http://sparkurl", 0, properties, mockContext);
       @SuppressWarnings("unchecked")
       ArgumentCaptor<Map<String, String>> argument = ArgumentCaptor.forClass(HashMap.class);
       verify(mockIntpEventClient).onParaInfosReceived(argument.capture());
@@ -120,14 +99,14 @@ void runUnderLocalTest() {
     void runUnderYarnTest() {
       Properties properties = new Properties();
       properties.setProperty("spark.jobGroup.id", "zeppelin|user1|noteId|paragraphId");
-      sparkShims.buildSparkJobUrl("yarn", "http://sparkurl", 0, properties, mockContext);
+      sparkUtils.buildSparkJobUrl("yarn", "http://sparkurl", 0, properties, mockContext);
       @SuppressWarnings("unchecked")
       ArgumentCaptor<Map<String, String>> argument = ArgumentCaptor.forClass(HashMap.class);
       verify(mockIntpEventClient).onParaInfosReceived(argument.capture());
       Map<String, String> mapValue = argument.getValue();
       assertTrue(mapValue.keySet().contains("jobUrl"));
 
-      if (sparkShims.supportYarn6615(VersionInfo.getVersion())) {
+      if (sparkUtils.supportYarn6615(VersionInfo.getVersion())) {
         assertTrue(mapValue.get("jobUrl").contains("/jobs/job?id="));
       } else {
         assertFalse(mapValue.get("jobUrl").contains("/jobs/job?id="));

spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkShimsTest.java renamed to spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkUtilsTest.java

@@ -55,8 +55,7 @@
         <module>spark-scala-parent</module>
         <module>scala-2.12</module>
         <module>scala-2.13</module>
-        <module>spark-shims</module>
-        <module>spark3-shims</module>
+        <module>spark-common</module>
     </modules>
 
     <build>

spark/pom.xml

@@ -198,9 +198,9 @@ class SparkScala212Interpreter(conf: SparkConf,
   }
 
   override def createZeppelinContext(): Unit = {
-    val sparkShims = SparkShims.getInstance(sc.version, properties, sparkSession)
-    sparkShims.setupSparkListener(sc.master, sparkUrl, InterpreterContext.get)
-    z = new SparkZeppelinContext(sc, sparkShims,
+    val sparkUtils = new SparkUtils(properties, sparkSession)
+    sparkUtils.setupSparkListener(sc.master, sparkUrl, InterpreterContext.get)
+    z = new SparkZeppelinContext(sc, sparkUtils,
       interpreterGroup.getInterpreterHookRegistry,
       properties.getProperty("zeppelin.spark.maxResult", "1000").toInt)
     bind("z", z.getClass.getCanonicalName, z, List("""@transient"""))

spark/scala-2.12/src/main/scala/org/apache/zeppelin/spark/SparkScala212Interpreter.scala

@@ -33,7 +33,7 @@ import scala.collection.JavaConverters._
   * ZeppelinContext for Spark
   */
 class SparkZeppelinContext(val sc: SparkContext,
-                           val sparkShims: SparkShims,
+                           val sparkUtils: SparkUtils,
                            val hooks2: InterpreterHookRegistry,
                            val maxResult2: Int) extends ZeppelinContext(hooks2, maxResult2) {
 
@@ -65,7 +65,7 @@ class SparkZeppelinContext(val sc: SparkContext,
 
   override def getInterpreterClassMap: util.Map[String, String] = interpreterClassMap.asJava
 
-  override def showData(obj: Any, maxResult: Int): String = sparkShims.showDataFrame(obj, maxResult, interpreterContext)
+  override def showData(obj: Any, maxResult: Int): String = sparkUtils.showDataFrame(obj, maxResult, interpreterContext)
 
   /**
    * create paragraph level of dynamic form of Select with no item selected.
@@ -237,6 +237,6 @@ class SparkZeppelinContext(val sc: SparkContext,
   }
 
   def getAsDataFrame(name: String): Object = {
-    sparkShims.getAsDataFrame(get(name).toString)
+    sparkUtils.getAsDataFrame(get(name).toString)
   }
 }

spark/scala-2.12/src/main/scala/org/apache/zeppelin/spark/SparkZeppelinContext.scala

@@ -200,9 +200,9 @@ class SparkScala213Interpreter(conf: SparkConf,
   }
 
   override def createZeppelinContext(): Unit = {
-    val sparkShims = SparkShims.getInstance(sc.version, properties, sparkSession)
-    sparkShims.setupSparkListener(sc.master, sparkUrl, InterpreterContext.get)
-    z = new SparkZeppelinContext(sc, sparkShims,
+    val sparkUtils = new SparkUtils(properties, sparkSession)
+    sparkUtils.setupSparkListener(sc.master, sparkUrl, InterpreterContext.get)
+    z = new SparkZeppelinContext(sc, sparkUtils,
       interpreterGroup.getInterpreterHookRegistry,
       properties.getProperty("zeppelin.spark.maxResult", "1000").toInt)
     bind("z", z.getClass.getCanonicalName, z, List("""@transient"""))

spark/scala-2.13/src/main/scala/org/apache/zeppelin/spark/SparkScala213Interpreter.scala

@@ -33,7 +33,7 @@ import scala.collection.JavaConverters._
   * ZeppelinContext for Spark
   */
 class SparkZeppelinContext(val sc: SparkContext,
-                           val sparkShims: SparkShims,
+                           val sparkUtils: SparkUtils,
                            val hooks2: InterpreterHookRegistry,
                            val maxResult2: Int) extends ZeppelinContext(hooks2, maxResult2) {
 
@@ -65,7 +65,7 @@ class SparkZeppelinContext(val sc: SparkContext,
 
   override def getInterpreterClassMap: util.Map[String, String] = interpreterClassMap.asJava
 
-  override def showData(obj: Any, maxResult: Int): String = sparkShims.showDataFrame(obj, maxResult, interpreterContext)
+  override def showData(obj: Any, maxResult: Int): String = sparkUtils.showDataFrame(obj, maxResult, interpreterContext)
 
   /**
    * create paragraph level of dynamic form of Select with no item selected.
@@ -237,6 +237,6 @@ class SparkZeppelinContext(val sc: SparkContext,
   }
 
   def getAsDataFrame(name: String): Object = {
-    sparkShims.getAsDataFrame(get(name).toString)
+    sparkUtils.getAsDataFrame(get(name).toString)
   }
 }

spark/scala-2.13/src/main/scala/org/apache/zeppelin/spark/SparkZeppelinContext.scala

@@ -18,6 +18,7 @@
 
 <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>spark-parent</artifactId>
     <groupId>org.apache.zeppelin</groupId>
@@ -26,21 +27,19 @@
   </parent>
 
   <modelVersion>4.0.0</modelVersion>
-  <artifactId>spark3-shims</artifactId>
+  <artifactId>spark-common</artifactId>
   <packaging>jar</packaging>
-  <name>Zeppelin: Spark3 Shims</name>
-
-  <properties>
-    <scala.binary.version>2.12</scala.binary.version>
-    <spark.version>3.4.1</spark.version>
-  </properties>
+  <name>Zeppelin: Spark Common</name>
 
   <dependencies>
-
+    <!--
+      This is for ZEPPELIN-2221 using VersionInfo for check the version of Yarn.
+      It's checked that VersionInfo is compatible at least 2.2.0 to the latest one.
+    -->
     <dependency>
-      <groupId>org.apache.zeppelin</groupId>
-      <artifactId>spark-shims</artifactId>
-      <version>${project.version}</version>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-client-api</artifactId>
+      <scope>provided</scope>
     </dependency>
 
     <dependency>