[ZEPPELIN-713] Expand ResourceId and put InterpreterResult into ResourcePool

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventPoller.java

@@ -260,7 +260,10 @@ private Object getResource(ResourceId resourceId) {
       boolean broken = false;
       try {
         client = remoteInterpreterProcess.getClient();
-        ByteBuffer res = client.resourceGet(resourceId.getName());
+        ByteBuffer res = client.resourceGet(
+            resourceId.getNoteId(),
+            resourceId.getParagraphId(),
+            resourceId.getName());
         Object o = Resource.deserializeObject(res);
         return o;
       } catch (Exception e) {

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java

@@ -343,12 +343,22 @@ protected Object jobRun() throws Throwable {
         }
 
         String interpreterResultMessage = result.message();
+
+        InterpreterResult combinedResult;
         if (interpreterResultMessage != null && !interpreterResultMessage.isEmpty()) {
           message += interpreterResultMessage;
-          return new InterpreterResult(result.code(), result.type(), message);
+          combinedResult = new InterpreterResult(result.code(), result.type(), message);
         } else {
-          return new InterpreterResult(result.code(), outputType, message);
+          combinedResult = new InterpreterResult(result.code(), outputType, message);
         }
+
+        // put result into resource pool
+        context.getResourcePool().put(
+            context.getNoteId(),
+            context.getParagraphId(),
+            WellKnownResourceName.ParagraphResult.toString(),
+            combinedResult);
+        return combinedResult;
       } finally {
         InterpreterContext.remove();
       }
@@ -651,9 +661,17 @@ public List<String> resoucePoolGetAll() throws TException {
   }
 
   @Override
-  public ByteBuffer resourceGet(String resourceName) throws TException {
+  public boolean resourceRemove(String noteId, String paragraphId, String resourceName)
+      throws TException {
+    Resource resource = resourcePool.remove(noteId, paragraphId, resourceName);
+    return resource != null;
+  }
+
+  @Override
+  public ByteBuffer resourceGet(String noteId, String paragraphId, String resourceName)
+      throws TException {
     logger.debug("Request resourceGet {} from ZeppelinServer", resourceName);
-    Resource resource = resourcePool.get(resourceName, false);
+    Resource resource = resourcePool.get(noteId, paragraphId, resourceName, false);
 
     if (resource == null || resource.get() == null || !resource.isSerializable()) {
       return ByteBuffer.allocate(0);

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterContext.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 /**
- * Autogenerated by Thrift Compiler (0.9.3)
+ * Autogenerated by Thrift Compiler (0.9.2)
  *
  * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
  *  @generated
@@ -51,7 +51,7 @@
 import org.slf4j.LoggerFactory;
 
 @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.3)", date = "2016-02-16")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-3-4")
 public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteInterpreterContext, RemoteInterpreterContext._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteInterpreterContext> {
   private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterContext");
 

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEvent.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 /**
- * Autogenerated by Thrift Compiler (0.9.3)
+ * Autogenerated by Thrift Compiler (0.9.2)
  *
  * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
  *  @generated
@@ -51,7 +51,7 @@
 import org.slf4j.LoggerFactory;
 
 @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.3)", date = "2016-02-16")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-3-4")
 public class RemoteInterpreterEvent implements org.apache.thrift.TBase<RemoteInterpreterEvent, RemoteInterpreterEvent._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteInterpreterEvent> {
   private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterEvent");
 

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventType.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 /**
- * Autogenerated by Thrift Compiler (0.9.3)
+ * Autogenerated by Thrift Compiler (0.9.2)
  *
  * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
  *  @generated

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResult.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 /**
- * Autogenerated by Thrift Compiler (0.9.3)
+ * Autogenerated by Thrift Compiler (0.9.2)
  *
  * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
  *  @generated
@@ -51,7 +51,7 @@
 import org.slf4j.LoggerFactory;
 
 @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.3)", date = "2016-02-16")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-3-4")
 public class RemoteInterpreterResult implements org.apache.thrift.TBase<RemoteInterpreterResult, RemoteInterpreterResult._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteInterpreterResult> {
   private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterResult");