[ZEPPELIN-4763]. Use isolated interpreter for cron note job

zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/FlinkIntegrationTest.java

@@ -23,6 +23,7 @@
 import org.apache.hadoop.yarn.api.records.YarnApplicationState;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.zeppelin.interpreter.ExecutionContext;
 import org.apache.zeppelin.interpreter.Interpreter;
 import org.apache.zeppelin.interpreter.InterpreterContext;
 import org.apache.zeppelin.interpreter.InterpreterException;
@@ -99,7 +100,7 @@ public static void tearDown() throws IOException {
 
   private void testInterpreterBasics() throws IOException, InterpreterException {
     // test FlinkInterpreter
-    Interpreter flinkInterpreter = interpreterFactory.getInterpreter("user1", "note1", "flink", "flink");
+    Interpreter flinkInterpreter = interpreterFactory.getInterpreter("flink", new ExecutionContext("user1", "note1", "flink"));
 
     InterpreterContext context = new InterpreterContext.Builder().setNoteId("note1").setParagraphId("paragraph_1").build();
     InterpreterResult interpreterResult = flinkInterpreter.interpret("1+1", context);

zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/JdbcIntegrationTest.java

@@ -19,6 +19,7 @@
 
 import com.google.common.collect.Lists;
 import org.apache.zeppelin.dep.Dependency;
+import org.apache.zeppelin.interpreter.ExecutionContext;
 import org.apache.zeppelin.interpreter.Interpreter;
 import org.apache.zeppelin.interpreter.InterpreterContext;
 import org.apache.zeppelin.interpreter.InterpreterException;
@@ -69,7 +70,7 @@ public void testMySql() throws InterpreterException, InterruptedException {
     interpreterSetting.setDependencies(Lists.newArrayList(dependency));
     interpreterSettingManager.restart(interpreterSetting.getId());
     interpreterSetting.waitForReady(60 * 1000);
-    Interpreter jdbcInterpreter = interpreterFactory.getInterpreter("user1", "note1", "jdbc", "test");
+    Interpreter jdbcInterpreter = interpreterFactory.getInterpreter("jdbc", new ExecutionContext("user1", "note1", "test"));
     assertNotNull("JdbcInterpreter is null", jdbcInterpreter);
 
     InterpreterContext context = new InterpreterContext.Builder()
@@ -88,7 +89,7 @@ public void testMySql() throws InterpreterException, InterruptedException {
     assertEquals("c1\tc2\n1\t2\n", interpreterResult.message().get(0).getData());
 
     // read table_1 from python interpreter
-    Interpreter pythonInterpreter = interpreterFactory.getInterpreter("user1", "note1", "python", "test");
+    Interpreter pythonInterpreter = interpreterFactory.getInterpreter("python", new ExecutionContext("user1", "note1", "test"));
     assertNotNull("PythonInterpreter is null", pythonInterpreter);
 
     context = new InterpreterContext.Builder()

zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/SparkIntegrationTest.java

@@ -24,6 +24,7 @@
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.maven.model.Model;
 import org.apache.maven.model.io.xpp3.MavenXpp3Reader;
+import org.apache.zeppelin.interpreter.ExecutionContext;
 import org.apache.zeppelin.interpreter.Interpreter;
 import org.apache.zeppelin.interpreter.InterpreterContext;
 import org.apache.zeppelin.interpreter.InterpreterException;
@@ -96,7 +97,7 @@ private void testInterpreterBasics() throws IOException, InterpreterException, X
     sparkInterpreterSetting.setProperty("spark.jars", new File("target/zeppelin-interpreter-integration-" + model.getVersion() + ".jar").getAbsolutePath());
 
     // test SparkInterpreter
-    Interpreter sparkInterpreter = interpreterFactory.getInterpreter("user1", "note1", "spark.spark", "test");
+    Interpreter sparkInterpreter = interpreterFactory.getInterpreter("spark.spark", new ExecutionContext("user1", "note1", "test"));
 
     InterpreterContext context = new InterpreterContext.Builder().setNoteId("note1").setParagraphId("paragraph_1").build();
     InterpreterResult interpreterResult = sparkInterpreter.interpret("sc.version", context);
@@ -113,24 +114,24 @@ private void testInterpreterBasics() throws IOException, InterpreterException, X
     assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code());
 
     // test PySparkInterpreter
-    Interpreter pySparkInterpreter = interpreterFactory.getInterpreter("user1", "note1", "spark.pyspark", "test");
+    Interpreter pySparkInterpreter = interpreterFactory.getInterpreter("spark.pyspark", new ExecutionContext("user1", "note1", "test"));
     interpreterResult = pySparkInterpreter.interpret("sqlContext.createDataFrame([(1,'a'),(2,'b')], ['id','name']).registerTempTable('test')", context);
     assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code());
 
     // test IPySparkInterpreter
-    Interpreter ipySparkInterpreter = interpreterFactory.getInterpreter("user1", "note1", "spark.ipyspark", "test");
+    Interpreter ipySparkInterpreter = interpreterFactory.getInterpreter("spark.ipyspark", new ExecutionContext("user1", "note1", "test"));
     interpreterResult = ipySparkInterpreter.interpret("sqlContext.table('test').show()", context);
     assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code());
 
     // test SparkSQLInterpreter
-    Interpreter sqlInterpreter = interpreterFactory.getInterpreter("user1", "note1", "spark.sql", "test");
+    Interpreter sqlInterpreter = interpreterFactory.getInterpreter("spark.sql", new ExecutionContext("user1", "note1", "test"));
     interpreterResult = sqlInterpreter.interpret("select count(1) as c from test", context);
     assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code());
     assertEquals(InterpreterResult.Type.TABLE, interpreterResult.message().get(0).getType());
     assertEquals("c\n2\n", interpreterResult.message().get(0).getData());
 
     // test SparkRInterpreter
-    Interpreter sparkrInterpreter = interpreterFactory.getInterpreter("user1", "note1", "spark.r", "test");
+    Interpreter sparkrInterpreter = interpreterFactory.getInterpreter("spark.r", new ExecutionContext("user1", "note1", "test"));
     if (isSpark2()) {
       interpreterResult = sparkrInterpreter.interpret("df <- as.DataFrame(faithful)\nhead(df)", context);
     } else {

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/ExecutionContext.java

@@ -0,0 +1,67 @@
+/*
+ * 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.
+ */
+
+package org.apache.zeppelin.interpreter;
+
+public class ExecutionContext {
+
+  private final String user;
+  private final String noteId;
+  private final String defaultInterpreterGroup;
+  private final boolean inCronMode;
+
+  public ExecutionContext(String user, String noteId) {
+    this(user, noteId, "", false);
+  }
+
+  public ExecutionContext(String user, String noteId, String defaultInterpreterGroup) {
+    this(user, noteId, defaultInterpreterGroup, false);
+  }
+
+  public ExecutionContext(String user, String noteId, String defaultInterpreterGroup, boolean inCronMode) {
+    this.user = user;
+    this.noteId = noteId;
+    this.defaultInterpreterGroup = defaultInterpreterGroup;
+    this.inCronMode = inCronMode;
+  }
+
+  public String getUser() {
+    return user;
+  }
+
+  public String getNoteId() {
+    return noteId;
+  }
+
+  public String getDefaultInterpreterGroup() {
+    return defaultInterpreterGroup;
+  }
+
+  public boolean isInCronMode() {
+    return inCronMode;
+  }
+
+  @Override
+  public String toString() {
+    return "ExecutionContext{" +
+            "user='" + user + '\'' +
+            ", noteId='" + noteId + '\'' +
+            ", defaultInterpreterGroup='" + defaultInterpreterGroup + '\'' +
+            ", inCronMode=" + inCronMode +
+            '}';
+  }
+}

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactoryInterface.java

@@ -26,9 +26,7 @@
  * So access through the interface.
  */
 public interface InterpreterFactoryInterface {
-  Interpreter getInterpreter(String user,
-                             String noteId,
-                             String replName,
-                             String defaultInterpreterSetting)
-      throws InterpreterNotFoundException;
+
+  Interpreter getInterpreter(String replName, ExecutionContext executionContext)
+          throws InterpreterNotFoundException;
 }

zeppelin-server/src/main/java/org/apache/zeppelin/rest/InterpreterRestApi.java

@@ -23,6 +23,7 @@
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.apache.zeppelin.annotation.ZeppelinApi;
 import org.apache.zeppelin.dep.Repository;
+import org.apache.zeppelin.interpreter.ExecutionContext;
 import org.apache.zeppelin.interpreter.InterpreterException;
 import org.apache.zeppelin.interpreter.InterpreterPropertyType;
 import org.apache.zeppelin.interpreter.InterpreterSetting;
@@ -200,7 +201,8 @@ public Response restartSetting(String message, @PathParam("settingId") String se
       if (null == noteId) {
         interpreterSettingManager.close(settingId);
       } else {
-        interpreterSettingManager.restart(settingId, noteId, authenticationService.getPrincipal());
+        interpreterSettingManager.restart(settingId,
+                new ExecutionContext(authenticationService.getPrincipal(), noteId));
       }
 
     } catch (InterpreterException e) {

zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java

@@ -100,7 +100,7 @@ public void setUp() throws Exception {
     InterpreterSettingManager mockInterpreterSettingManager = mock(InterpreterSettingManager.class);
     InterpreterFactory mockInterpreterFactory = mock(InterpreterFactory.class);
     Interpreter mockInterpreter = mock(Interpreter.class);
-    when(mockInterpreterFactory.getInterpreter(any(), any(), any(), any()))
+    when(mockInterpreterFactory.getInterpreter(any(), any()))
         .thenReturn(mockInterpreter);
     when(mockInterpreter.interpret(eq("invalid_code"), any()))
         .thenReturn(new InterpreterResult(Code.ERROR, "failed"));

zeppelin-web/src/app/notebook/notebook-actionBar.html

@@ -300,12 +300,6 @@ <h3>
                   {{note.info.cron}}
                 </p>
               </div>
-              <div>
-                <span>- After execution stop the interpreter </span>
-                <input type="checkbox"
-                       ng-model="note.config.releaseresource"
-                       ng-click="setReleaseResource(note.config.releaseresource)"/>
-              </div>
             </div>
           </li>
         </ul>

zeppelin-web/src/app/notebook/notebook.controller.js

@@ -512,12 +512,6 @@ function NotebookCtrl($scope, $route, $routeParams, $location, $rootScope,
     $scope.setConfig();
   };
 
-  /** Set release resource for this note **/
-  $scope.setReleaseResource = function(value) {
-    $scope.note.config.releaseresource = value;
-    $scope.setConfig();
-  };
-
   /** Update note config **/
   $scope.setConfig = function(config) {
     if (config) {

zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java

@@ -17,14 +17,11 @@
 
 package org.apache.zeppelin.interpreter;
 
-import com.google.common.base.Preconditions;
 import javax.inject.Inject;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.List;
-
 /**
  * //TODO(zjffdu) considering to move to InterpreterSettingManager
  *
@@ -42,17 +39,15 @@ public InterpreterFactory(InterpreterSettingManager interpreterSettingManager) {
   }
 
   @Override
-  public Interpreter getInterpreter(String user,
-                                    String noteId,
-                                    String replName,
-                                    String defaultInterpreterSetting)
+  public Interpreter getInterpreter(String replName,
+                                    ExecutionContext executionContext)
       throws InterpreterNotFoundException {
 
     if (StringUtils.isBlank(replName)) {
       // Get the default interpreter of the defaultInterpreterSetting
       InterpreterSetting defaultSetting =
-          interpreterSettingManager.getByName(defaultInterpreterSetting);
-      return defaultSetting.getDefaultInterpreter(user, noteId);
+          interpreterSettingManager.getByName(executionContext.getDefaultInterpreterGroup());
+      return defaultSetting.getDefaultInterpreter(executionContext);
     }
 
     String[] replNameSplits = replName.split("\\.");
@@ -61,7 +56,7 @@ public Interpreter getInterpreter(String user,
       String name = replNameSplits[1];
       InterpreterSetting setting = interpreterSettingManager.getByName(group);
       if (null != setting) {
-        Interpreter interpreter = setting.getInterpreter(user, noteId, name);
+        Interpreter interpreter = setting.getInterpreter(executionContext, name);
         if (null != interpreter) {
           return interpreter;
         }
@@ -72,9 +67,9 @@ public Interpreter getInterpreter(String user,
     } else if (replNameSplits.length == 1){
       // first assume group is omitted
       InterpreterSetting setting =
-          interpreterSettingManager.getByName(defaultInterpreterSetting);
+          interpreterSettingManager.getByName(executionContext.getDefaultInterpreterGroup());
       if (setting != null) {
-        Interpreter interpreter = setting.getInterpreter(user, noteId, replName);
+        Interpreter interpreter = setting.getInterpreter(executionContext, replName);
         if (null != interpreter) {
           return interpreter;
         }
@@ -83,7 +78,7 @@ public Interpreter getInterpreter(String user,
       // then assume interpreter name is omitted
       setting = interpreterSettingManager.getByName(replName);
       if (null != setting) {
-        return setting.getDefaultInterpreter(user, noteId);
+        return setting.getDefaultInterpreter(executionContext);
       }
     }