ZEPPELIN-2324. Add property zeppelin.spark.unSupportedVersionCheck for trying new spark version

docs/interpreter/spark.md

@@ -140,6 +140,11 @@ You can also set other Spark properties which are not listed in the table. For a
     <td>true</td>
     <td>Import implicits, UDF collection, and sql if set true.</td>
   </tr>
+  <tr>
+    <td>zeppelin.spark.unSupportedVersionCheck</td>
+    <td>true</td>
+    <td>Don't change it, It is only for zeppelin developer use, not safe or use in production</td>
+  </tr>
 </table>
 
 Without any configuration, Spark interpreter works out of box in local mode. But if you want to connect to your Spark cluster, you'll need to follow below two simple steps.

spark/src/main/java/org/apache/zeppelin/spark/PySparkInterpreter.java

@@ -337,7 +337,8 @@ public class PySparkInterpreter extends Interpreter implements ExecuteResultHand
   public InterpreterResult interpret(String st, InterpreterContext context) {
     SparkInterpreter sparkInterpreter = getSparkInterpreter();
     sparkInterpreter.populateSparkWebUrl(context);
-    if (sparkInterpreter.getSparkVersion().isUnsupportedVersion()) {
+    if (sparkInterpreter.isUnsupportedVersionCheck
+      && sparkInterpreter.getSparkVersion().isUnsupportedVersion()) {
       return new InterpreterResult(Code.ERROR, "Spark "
           + sparkInterpreter.getSparkVersion().toString() + " is not supported");
     }
@@ -466,7 +467,8 @@ public class PySparkInterpreter extends Interpreter implements ExecuteResultHand
 
     //start code for completion
     SparkInterpreter sparkInterpreter = getSparkInterpreter();
-    if (sparkInterpreter.getSparkVersion().isUnsupportedVersion() == false
+    if (sparkInterpreter.isUnsupportedVersionCheck
+        && sparkInterpreter.getSparkVersion().isUnsupportedVersion() == false
             && pythonscriptRunning == false) {
       return new LinkedList<>();
     }

spark/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java

@@ -128,7 +128,7 @@ public class SparkInterpreter extends Interpreter {
   private static File outputDir;          // class outputdir for scala 2.11
   private Object classServer;      // classserver for scala 2.11
   private JavaSparkContext jsc;
-
+  boolean isUnsupportedVersionCheck;
 
   public SparkInterpreter(Properties property) {
     super(property);
@@ -609,6 +609,9 @@ public class SparkInterpreter extends Interpreter {
 
   @Override
   public void open() {
+    this.isUnsupportedVersionCheck = java.lang.Boolean.parseBoolean(
+            property.getProperty("zeppelin.spark.unSupportedVersionCheck", "true"));
+
     // set properties and do login before creating any spark stuff for secured cluster
     if (isYarnMode()) {
       System.setProperty("SPARK_YARN_MODE", "true");
@@ -1158,7 +1161,7 @@ public class SparkInterpreter extends Interpreter {
    */
   @Override
   public InterpreterResult interpret(String line, InterpreterContext context) {
-    if (sparkVersion.isUnsupportedVersion()) {
+    if (isUnsupportedVersionCheck && sparkVersion.isUnsupportedVersion()) {
       return new InterpreterResult(Code.ERROR, "Spark " + sparkVersion.toString()
           + " is not supported");
     }

spark/src/main/java/org/apache/zeppelin/spark/SparkRInterpreter.java

@@ -108,6 +108,11 @@ public class SparkRInterpreter extends Interpreter {
 
     SparkInterpreter sparkInterpreter = getSparkInterpreter();
     sparkInterpreter.populateSparkWebUrl(interpreterContext);
+    if (sparkInterpreter.isUnsupportedVersionCheck
+      && sparkInterpreter.getSparkVersion().isUnsupportedVersion()) {
+      return new InterpreterResult(InterpreterResult.Code.ERROR, "Spark "
+          + sparkInterpreter.getSparkVersion().toString() + " is not supported");
+    }
 
     String jobGroup = Utils.buildJobGroupId(interpreterContext);
     sparkInterpreter.getSparkContext().setJobGroup(jobGroup, "Zeppelin", false);

spark/src/main/java/org/apache/zeppelin/spark/SparkSqlInterpreter.java

@@ -87,7 +87,8 @@ public class SparkSqlInterpreter extends Interpreter {
     SQLContext sqlc = null;
     SparkInterpreter sparkInterpreter = getSparkInterpreter();
 
-    if (sparkInterpreter.getSparkVersion().isUnsupportedVersion()) {
+    if (sparkInterpreter.isUnsupportedVersionCheck
+        && sparkInterpreter.getSparkVersion().isUnsupportedVersion()) {
       return new InterpreterResult(Code.ERROR, "Spark "
           + sparkInterpreter.getSparkVersion().toString() + " is not supported");
     }

spark/src/main/resources/interpreter-setting.json

@@ -53,6 +53,12 @@
         "propertyName": "spark.master",
         "defaultValue": "local[*]",
         "description": "Spark master uri. ex) spark://masterhost:7077"
+      },
+      "zeppelin.spark.unSupportedVersionCheck": {
+        "envName": null,
+        "propertyName": "zeppelin.spark.unSupportedVersionCheck",
+        "defaultValue": "true",
+        "description": "Don't change it, It is only for zeppelin developer use, not safe to use in production"
       }
     },
     "editor": {

spark/src/main/sparkr-resources/interpreter-setting.json

@@ -53,6 +53,12 @@
         "propertyName": "spark.master",
         "defaultValue": "local[*]",
         "description": "Spark master uri. ex) spark://masterhost:7077"
+      },
+      "zeppelin.spark.unSupportedVersionCheck": {
+        "envName": null,
+        "propertyName": "zeppelin.spark.unSupportedVersionCheck",
+        "defaultValue": "true",
+        "description": "Don't change it, It is only for zeppelin developer use, not safe to use in production"
       }
     },
     "editor": {