Address review comments

Signed-off-by: Karup <karuppayya@outlook.com>

Author: Karup

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

@@ -377,7 +377,7 @@ public InterpreterResult interpret(String st, InterpreterContext context) {
           "pyspark " + sparkInterpreter.getSparkContext().version() + " is not supported"));
       return new InterpreterResult(Code.ERROR, errorMessage);
     }
-    String jobGroup = sparkInterpreter.getJobGroup(context);
+    String jobGroup = Utils.buildJobGroupId(context);
     ZeppelinContext z = sparkInterpreter.getZeppelinContext();
     z.setInterpreterContext(context);
     z.setGui(context.getGui());

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

@@ -172,6 +172,7 @@ public synchronized void onJobStart(SparkListenerJobStart jobStart) {
           RemoteEventClientWrapper eventClient = ZeppelinContext.getEventClient();
           Map<String, String> infos = new java.util.HashMap<>();
           infos.put("jobUrl", jobUrl);
+          infos.put("label", "SPARK JOB");
           if (eventClient != null) {
             eventClient.onParaInfosReceived(noteId, paragraphId, infos);
           }
@@ -1147,10 +1148,6 @@ public Object getLastObject() {
     return obj;
   }
 
-  String getJobGroup(InterpreterContext context) {
-    return "zeppelin-" + context.getNoteId() + "-" + context.getParagraphId();
-  }
-
   /**
    * Interpret a single line.
    */
@@ -1171,7 +1168,7 @@ public InterpreterResult interpret(String line, InterpreterContext context) {
   public InterpreterResult interpret(String[] lines, InterpreterContext context) {
     synchronized (this) {
       z.setGui(context.getGui());
-      sc.setJobGroup(getJobGroup(context), "Zeppelin", false);
+      sc.setJobGroup(Utils.buildJobGroupId(context), "Zeppelin", false);
       InterpreterResult r = interpretInput(lines, context);
       sc.clearJobGroup();
       return r;
@@ -1294,12 +1291,12 @@ private void putLatestVarInResourcePool(InterpreterContext context) {
 
   @Override
   public void cancel(InterpreterContext context) {
-    sc.cancelJobGroup(getJobGroup(context));
+    sc.cancelJobGroup(Utils.buildJobGroupId(context));
   }
 
   @Override
   public int getProgress(InterpreterContext context) {
-    String jobGroup = getJobGroup(context);
+    String jobGroup = Utils.buildJobGroupId(context);
     int completedTasks = 0;
     int totalTasks = 0;
 

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

@@ -105,7 +105,7 @@ public InterpreterResult interpret(String lines, InterpreterContext interpreterC
     SparkInterpreter sparkInterpreter = getSparkInterpreter();
     sparkInterpreter.populateSparkWebUrl(interpreterContext);
 
-    String jobGroup = sparkInterpreter.getJobGroup(interpreterContext);
+    String jobGroup = Utils.buildJobGroupId(interpreterContext);
     sparkInterpreter.getSparkContext().setJobGroup(jobGroup, "Zeppelin", false);
 
     String imageWidth = getProperty("zeppelin.R.image.width");

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

@@ -101,7 +101,7 @@ public InterpreterResult interpret(String st, InterpreterContext context) {
       sc.setLocalProperty("spark.scheduler.pool", null);
     }
 
-    sc.setJobGroup(sparkInterpreter.getJobGroup(context), "Zeppelin", false);
+    sc.setJobGroup(Utils.buildJobGroupId(context), "Zeppelin", false);
     Object rdd = null;
     try {
       // method signature of sqlc.sql() is changed
@@ -134,7 +134,7 @@ public void cancel(InterpreterContext context) {
     SQLContext sqlc = sparkInterpreter.getSQLContext();
     SparkContext sc = sqlc.sparkContext();
 
-    sc.cancelJobGroup(sparkInterpreter.getJobGroup(context));
+    sc.cancelJobGroup(Utils.buildJobGroupId(context));
   }
 
   @Override

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

@@ -17,6 +17,7 @@
 
 package org.apache.zeppelin.spark;
 
+import org.apache.zeppelin.interpreter.InterpreterContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -106,4 +107,8 @@ static boolean isSpark2() {
       return false;
     }
   }
+  
+  public static String buildJobGroupId(InterpreterContext context) {
+    return "zeppelin-" + context.getNoteId() + "-" + context.getParagraphId();
+  }
 }

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

@@ -223,8 +223,7 @@ public static String showDF(SparkContext sc,
       Object df, int maxResult) {
     Object[] rows = null;
     Method take;
-    String jobGroup = "zeppelin-" + interpreterContext.getNoteId() + "-"
-        + interpreterContext.getParagraphId();
+    String jobGroup = Utils.buildJobGroupId(interpreterContext);
     sc.setJobGroup(jobGroup, "Zeppelin", false);
 
     try {

zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java

@@ -2327,7 +2327,9 @@ public void onParaInfosReceived(String noteId, String paragraphId,
         setting.addNoteToPara(noteId, paragraphId);
         metaInfos.remove("noteId");
         metaInfos.remove("paraId");
-        paragraph.updateRuntimeInfos(metaInfos);
+        String label = metaInfos.get("label");
+        metaInfos.remove("label");
+        paragraph.updateRuntimeInfos(label, metaInfos, setting.getGroup());
         broadcast(note.getId(), new Message(OP.PARAGRAPH).put("paragraph", paragraph));
       }
     }

zeppelin-web/src/app/notebook/paragraph/paragraph-control.html

@@ -13,7 +13,24 @@
 -->
 
 <div id="{{paragraph.id}}_control" class="control" ng-show="!asIframe">
-
+  <span ng-show="paragraph.runtimeInfos.jobUrl.values.length == 1 && paragraph.runtimeInfos.jobUrl.group == 'spark'">
+    <a href="{{paragraph.runtimeInfos.jobUrl.values[0]}}" target="_blank" style="text-decoration: none;">
+      <span class="fa fa-tasks"></span>
+      {{paragraph.runtimeInfos.jobUrl.label}}
+    </a>
+  </span>
+  <span class="dropdown" ng-show="paragraph.runtimeInfos.jobUrl.values.length > 1 && paragraph.runtimeInfos.jobUrl.group == 'spark'">
+    <span style="cursor:pointer;color:#3071A9" tooltip-placement="top" tooltip="View in Spark web UI"
+          data-toggle="dropdown" type="button">
+      <span class="fa fa-tasks"></span>
+      {{paragraph.runtimeInfos.jobUrl.label}}S
+    </span>
+    <ul class="dropdown-menu" role="menu" style="width:200px;z-index:1002">
+      <li ng-class="{'option-disabled': !noteOperationsAllowed()}" ng-repeat="url in paragraph.runtimeInfos.jobUrl.values">
+        <a href="{{url}}" target="_blank"><span class="fa fa-external-link-square"></span> Jobs #{{$index}}</a>
+      </li>
+    </ul>
+  </span>
   <span>
     {{paragraph.status}}
   </span>

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

@@ -127,7 +127,7 @@ public String getName() {
     return name;
   }
 
-  String getGroup() {
+  public String getGroup() {
     return group;
   }
 

zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java

@@ -71,6 +71,7 @@ public class Paragraph extends Job implements Serializable, Cloneable {
   // For backward compatibility of note.json format after ZEPPELIN-212
   Object result;
   private Map<String, Set<String>> runtimeInfos;
+  private Map<String, ParagraphRuntimeInfos> runtimeInfos;
 
   /**
    * Applicaiton states in this paragraph
@@ -678,19 +679,19 @@ private boolean isValidInterpreter(String replName) {
     }
   }
 
-  public void updateRuntimeInfos(Map<String, String> infos) {
+  public void updateRuntimeInfos(String label, Map<String, String> infos, String group) {
     if (this.runtimeInfos == null) {
-      this.runtimeInfos = new HashMap<String, Set<String>>();
+      this.runtimeInfos = new HashMap<String, ParagraphRuntimeInfos>();
     }
 
     if (infos != null) {
       for (String key : infos.keySet()) {
-        Set<String> values = this.runtimeInfos.get(key);
-        if (values == null) {
-          values = new HashSet<>();
-          this.runtimeInfos.put(key, values);
+        ParagraphRuntimeInfos info = this.runtimeInfos.get(key);
+        if (info == null) {
+          info = new ParagraphRuntimeInfos(key, label, group);
+          this.runtimeInfos.put(key, info);
         }
-        values.add(infos.get(key));
+        info.addValue(infos.get(key));
       }
     }
   }