You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zeppelin.apache.org by mo...@apache.org on 2016/11/29 16:07:21 UTC
[1/2] zeppelin git commit: [ZEPPELIN-1665] Z.run with external note
executable and access resource for zeppelin in each interpreter
Repository: zeppelin
Updated Branches:
refs/heads/master 6a2a341b0 -> 9db840c65
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ZeppelinServerResourceParagraphRunner.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ZeppelinServerResourceParagraphRunner.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ZeppelinServerResourceParagraphRunner.java
new file mode 100644
index 0000000..da2f703
--- /dev/null
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ZeppelinServerResourceParagraphRunner.java
@@ -0,0 +1,520 @@
+/**
+ * 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.
+ */
+/**
+ * Autogenerated by Thrift Compiler (0.9.2)
+ *
+ * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
+ * @generated
+ */
+package org.apache.zeppelin.interpreter.thrift;
+
+import org.apache.thrift.scheme.IScheme;
+import org.apache.thrift.scheme.SchemeFactory;
+import org.apache.thrift.scheme.StandardScheme;
+
+import org.apache.thrift.scheme.TupleScheme;
+import org.apache.thrift.protocol.TTupleProtocol;
+import org.apache.thrift.protocol.TProtocolException;
+import org.apache.thrift.EncodingUtils;
+import org.apache.thrift.TException;
+import org.apache.thrift.async.AsyncMethodCallback;
+import org.apache.thrift.server.AbstractNonblockingServer.*;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.EnumMap;
+import java.util.Set;
+import java.util.HashSet;
+import java.util.EnumSet;
+import java.util.Collections;
+import java.util.BitSet;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import javax.annotation.Generated;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-25")
+public class ZeppelinServerResourceParagraphRunner implements org.apache.thrift.TBase<ZeppelinServerResourceParagraphRunner, ZeppelinServerResourceParagraphRunner._Fields>, java.io.Serializable, Cloneable, Comparable<ZeppelinServerResourceParagraphRunner> {
+ private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("ZeppelinServerResourceParagraphRunner");
+
+ private static final org.apache.thrift.protocol.TField NOTE_ID_FIELD_DESC = new org.apache.thrift.protocol.TField("noteId", org.apache.thrift.protocol.TType.STRING, (short)1);
+ private static final org.apache.thrift.protocol.TField PARAGRAPH_ID_FIELD_DESC = new org.apache.thrift.protocol.TField("paragraphId", org.apache.thrift.protocol.TType.STRING, (short)2);
+
+ private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
+ static {
+ schemes.put(StandardScheme.class, new ZeppelinServerResourceParagraphRunnerStandardSchemeFactory());
+ schemes.put(TupleScheme.class, new ZeppelinServerResourceParagraphRunnerTupleSchemeFactory());
+ }
+
+ public String noteId; // required
+ public String paragraphId; // required
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements org.apache.thrift.TFieldIdEnum {
+ NOTE_ID((short)1, "noteId"),
+ PARAGRAPH_ID((short)2, "paragraphId");
+
+ private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+ static {
+ for (_Fields field : EnumSet.allOf(_Fields.class)) {
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ public static _Fields findByThriftId(int fieldId) {
+ switch(fieldId) {
+ case 1: // NOTE_ID
+ return NOTE_ID;
+ case 2: // PARAGRAPH_ID
+ return PARAGRAPH_ID;
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, throwing an exception
+ * if it is not found.
+ */
+ public static _Fields findByThriftIdOrThrow(int fieldId) {
+ _Fields fields = findByThriftId(fieldId);
+ if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+ return fields;
+ }
+
+ /**
+ * Find the _Fields constant that matches name, or null if its not found.
+ */
+ public static _Fields findByName(String name) {
+ return byName.get(name);
+ }
+
+ private final short _thriftId;
+ private final String _fieldName;
+
+ _Fields(short thriftId, String fieldName) {
+ _thriftId = thriftId;
+ _fieldName = fieldName;
+ }
+
+ public short getThriftFieldId() {
+ return _thriftId;
+ }
+
+ public String getFieldName() {
+ return _fieldName;
+ }
+ }
+
+ // isset id assignments
+ public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+ static {
+ Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
+ tmpMap.put(_Fields.NOTE_ID, new org.apache.thrift.meta_data.FieldMetaData("noteId", org.apache.thrift.TFieldRequirementType.DEFAULT,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
+ tmpMap.put(_Fields.PARAGRAPH_ID, new org.apache.thrift.meta_data.FieldMetaData("paragraphId", org.apache.thrift.TFieldRequirementType.DEFAULT,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
+ metaDataMap = Collections.unmodifiableMap(tmpMap);
+ org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(ZeppelinServerResourceParagraphRunner.class, metaDataMap);
+ }
+
+ public ZeppelinServerResourceParagraphRunner() {
+ }
+
+ public ZeppelinServerResourceParagraphRunner(
+ String noteId,
+ String paragraphId)
+ {
+ this();
+ this.noteId = noteId;
+ this.paragraphId = paragraphId;
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public ZeppelinServerResourceParagraphRunner(ZeppelinServerResourceParagraphRunner other) {
+ if (other.isSetNoteId()) {
+ this.noteId = other.noteId;
+ }
+ if (other.isSetParagraphId()) {
+ this.paragraphId = other.paragraphId;
+ }
+ }
+
+ public ZeppelinServerResourceParagraphRunner deepCopy() {
+ return new ZeppelinServerResourceParagraphRunner(this);
+ }
+
+ @Override
+ public void clear() {
+ this.noteId = null;
+ this.paragraphId = null;
+ }
+
+ public String getNoteId() {
+ return this.noteId;
+ }
+
+ public ZeppelinServerResourceParagraphRunner setNoteId(String noteId) {
+ this.noteId = noteId;
+ return this;
+ }
+
+ public void unsetNoteId() {
+ this.noteId = null;
+ }
+
+ /** Returns true if field noteId is set (has been assigned a value) and false otherwise */
+ public boolean isSetNoteId() {
+ return this.noteId != null;
+ }
+
+ public void setNoteIdIsSet(boolean value) {
+ if (!value) {
+ this.noteId = null;
+ }
+ }
+
+ public String getParagraphId() {
+ return this.paragraphId;
+ }
+
+ public ZeppelinServerResourceParagraphRunner setParagraphId(String paragraphId) {
+ this.paragraphId = paragraphId;
+ return this;
+ }
+
+ public void unsetParagraphId() {
+ this.paragraphId = null;
+ }
+
+ /** Returns true if field paragraphId is set (has been assigned a value) and false otherwise */
+ public boolean isSetParagraphId() {
+ return this.paragraphId != null;
+ }
+
+ public void setParagraphIdIsSet(boolean value) {
+ if (!value) {
+ this.paragraphId = null;
+ }
+ }
+
+ public void setFieldValue(_Fields field, Object value) {
+ switch (field) {
+ case NOTE_ID:
+ if (value == null) {
+ unsetNoteId();
+ } else {
+ setNoteId((String)value);
+ }
+ break;
+
+ case PARAGRAPH_ID:
+ if (value == null) {
+ unsetParagraphId();
+ } else {
+ setParagraphId((String)value);
+ }
+ break;
+
+ }
+ }
+
+ public Object getFieldValue(_Fields field) {
+ switch (field) {
+ case NOTE_ID:
+ return getNoteId();
+
+ case PARAGRAPH_ID:
+ return getParagraphId();
+
+ }
+ throw new IllegalStateException();
+ }
+
+ /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+ public boolean isSet(_Fields field) {
+ if (field == null) {
+ throw new IllegalArgumentException();
+ }
+
+ switch (field) {
+ case NOTE_ID:
+ return isSetNoteId();
+ case PARAGRAPH_ID:
+ return isSetParagraphId();
+ }
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ if (that == null)
+ return false;
+ if (that instanceof ZeppelinServerResourceParagraphRunner)
+ return this.equals((ZeppelinServerResourceParagraphRunner)that);
+ return false;
+ }
+
+ public boolean equals(ZeppelinServerResourceParagraphRunner that) {
+ if (that == null)
+ return false;
+
+ boolean this_present_noteId = true && this.isSetNoteId();
+ boolean that_present_noteId = true && that.isSetNoteId();
+ if (this_present_noteId || that_present_noteId) {
+ if (!(this_present_noteId && that_present_noteId))
+ return false;
+ if (!this.noteId.equals(that.noteId))
+ return false;
+ }
+
+ boolean this_present_paragraphId = true && this.isSetParagraphId();
+ boolean that_present_paragraphId = true && that.isSetParagraphId();
+ if (this_present_paragraphId || that_present_paragraphId) {
+ if (!(this_present_paragraphId && that_present_paragraphId))
+ return false;
+ if (!this.paragraphId.equals(that.paragraphId))
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ List<Object> list = new ArrayList<Object>();
+
+ boolean present_noteId = true && (isSetNoteId());
+ list.add(present_noteId);
+ if (present_noteId)
+ list.add(noteId);
+
+ boolean present_paragraphId = true && (isSetParagraphId());
+ list.add(present_paragraphId);
+ if (present_paragraphId)
+ list.add(paragraphId);
+
+ return list.hashCode();
+ }
+
+ @Override
+ public int compareTo(ZeppelinServerResourceParagraphRunner other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+
+ lastComparison = Boolean.valueOf(isSetNoteId()).compareTo(other.isSetNoteId());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetNoteId()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.noteId, other.noteId);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ lastComparison = Boolean.valueOf(isSetParagraphId()).compareTo(other.isSetParagraphId());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetParagraphId()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.paragraphId, other.paragraphId);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ return 0;
+ }
+
+ public _Fields fieldForId(int fieldId) {
+ return _Fields.findByThriftId(fieldId);
+ }
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+ schemes.get(iprot.getScheme()).getScheme().read(iprot, this);
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+ schemes.get(oprot.getScheme()).getScheme().write(oprot, this);
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("ZeppelinServerResourceParagraphRunner(");
+ boolean first = true;
+
+ sb.append("noteId:");
+ if (this.noteId == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.noteId);
+ }
+ first = false;
+ if (!first) sb.append(", ");
+ sb.append("paragraphId:");
+ if (this.paragraphId == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.paragraphId);
+ }
+ first = false;
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws org.apache.thrift.TException {
+ // check for required fields
+ // check for sub-struct validity
+ }
+
+ private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+ try {
+ write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
+ try {
+ read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private static class ZeppelinServerResourceParagraphRunnerStandardSchemeFactory implements SchemeFactory {
+ public ZeppelinServerResourceParagraphRunnerStandardScheme getScheme() {
+ return new ZeppelinServerResourceParagraphRunnerStandardScheme();
+ }
+ }
+
+ private static class ZeppelinServerResourceParagraphRunnerStandardScheme extends StandardScheme<ZeppelinServerResourceParagraphRunner> {
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot, ZeppelinServerResourceParagraphRunner struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TField schemeField;
+ iprot.readStructBegin();
+ while (true)
+ {
+ schemeField = iprot.readFieldBegin();
+ if (schemeField.type == org.apache.thrift.protocol.TType.STOP) {
+ break;
+ }
+ switch (schemeField.id) {
+ case 1: // NOTE_ID
+ if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
+ struct.noteId = iprot.readString();
+ struct.setNoteIdIsSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ break;
+ case 2: // PARAGRAPH_ID
+ if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
+ struct.paragraphId = iprot.readString();
+ struct.setParagraphIdIsSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ break;
+ default:
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ iprot.readFieldEnd();
+ }
+ iprot.readStructEnd();
+
+ // check for required fields of primitive type, which can't be checked in the validate method
+ struct.validate();
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot, ZeppelinServerResourceParagraphRunner struct) throws org.apache.thrift.TException {
+ struct.validate();
+
+ oprot.writeStructBegin(STRUCT_DESC);
+ if (struct.noteId != null) {
+ oprot.writeFieldBegin(NOTE_ID_FIELD_DESC);
+ oprot.writeString(struct.noteId);
+ oprot.writeFieldEnd();
+ }
+ if (struct.paragraphId != null) {
+ oprot.writeFieldBegin(PARAGRAPH_ID_FIELD_DESC);
+ oprot.writeString(struct.paragraphId);
+ oprot.writeFieldEnd();
+ }
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ }
+
+ private static class ZeppelinServerResourceParagraphRunnerTupleSchemeFactory implements SchemeFactory {
+ public ZeppelinServerResourceParagraphRunnerTupleScheme getScheme() {
+ return new ZeppelinServerResourceParagraphRunnerTupleScheme();
+ }
+ }
+
+ private static class ZeppelinServerResourceParagraphRunnerTupleScheme extends TupleScheme<ZeppelinServerResourceParagraphRunner> {
+
+ @Override
+ public void write(org.apache.thrift.protocol.TProtocol prot, ZeppelinServerResourceParagraphRunner struct) throws org.apache.thrift.TException {
+ TTupleProtocol oprot = (TTupleProtocol) prot;
+ BitSet optionals = new BitSet();
+ if (struct.isSetNoteId()) {
+ optionals.set(0);
+ }
+ if (struct.isSetParagraphId()) {
+ optionals.set(1);
+ }
+ oprot.writeBitSet(optionals, 2);
+ if (struct.isSetNoteId()) {
+ oprot.writeString(struct.noteId);
+ }
+ if (struct.isSetParagraphId()) {
+ oprot.writeString(struct.paragraphId);
+ }
+ }
+
+ @Override
+ public void read(org.apache.thrift.protocol.TProtocol prot, ZeppelinServerResourceParagraphRunner struct) throws org.apache.thrift.TException {
+ TTupleProtocol iprot = (TTupleProtocol) prot;
+ BitSet incoming = iprot.readBitSet(2);
+ if (incoming.get(0)) {
+ struct.noteId = iprot.readString();
+ struct.setNoteIdIsSet(true);
+ }
+ if (incoming.get(1)) {
+ struct.paragraphId = iprot.readString();
+ struct.setParagraphIdIsSet(true);
+ }
+ }
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/thrift/RemoteInterpreterService.thrift
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/thrift/RemoteInterpreterService.thrift b/zeppelin-interpreter/src/main/thrift/RemoteInterpreterService.thrift
index 1626d6d..fdcf21b 100644
--- a/zeppelin-interpreter/src/main/thrift/RemoteInterpreterService.thrift
+++ b/zeppelin-interpreter/src/main/thrift/RemoteInterpreterService.thrift
@@ -50,9 +50,11 @@ enum RemoteInterpreterEventType {
OUTPUT_UPDATE = 9,
ANGULAR_REGISTRY_PUSH = 10,
APP_STATUS_UPDATE = 11,
- META_INFOS = 12
+ META_INFOS = 12,
+ REMOTE_ZEPPELIN_SERVER_RESOURCE = 13
}
+
struct RemoteInterpreterEvent {
1: RemoteInterpreterEventType type,
2: string data // json serialized data
@@ -63,6 +65,11 @@ struct RemoteApplicationResult {
2: string msg
}
+struct ZeppelinServerResourceParagraphRunner {
+ 1: string noteId,
+ 2: string paragraphId
+}
+
/*
* The below variables(name, value) will be connected to getCompletions in paragraph.controller.js
*
@@ -110,4 +117,6 @@ service RemoteInterpreterService {
RemoteApplicationResult loadApplication(1: string applicationInstanceId, 2: string packageInfo, 3: string sessionKey, 4: string paragraphId);
RemoteApplicationResult unloadApplication(1: string applicationInstanceId);
RemoteApplicationResult runApplication(1: string applicationInstanceId);
+
+ void onReceivedZeppelinResource(1: string object);
}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterOutputTestStream.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterOutputTestStream.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterOutputTestStream.java
index c505255..9922d57 100644
--- a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterOutputTestStream.java
+++ b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterOutputTestStream.java
@@ -163,4 +163,16 @@ public class RemoteInterpreterOutputTestStream implements RemoteInterpreterProce
public void onMetaInfosReceived(String settingId, Map<String, String> metaInfos) {
}
+
+ @Override
+ public void onGetParagraphRunners(String noteId, String paragraphId, RemoteWorksEventListener callback) {
+ if (callback != null) {
+ callback.onFinished(new LinkedList<>());
+ }
+ }
+
+ @Override
+ public void onRemoteRunParagraph(String noteId, String ParagraphID) throws Exception {
+
+ }
}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java
index 99c75a8..0c739ee 100644
--- a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java
+++ b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java
@@ -81,16 +81,16 @@ public class RemoteInterpreterTest {
private RemoteInterpreter createMockInterpreterA(Properties p, String noteId) {
return new RemoteInterpreter(
- p,
- noteId,
- MockInterpreterA.class.getName(),
- new File(INTERPRETER_SCRIPT).getAbsolutePath(),
- "fake",
- "fakeRepo",
- env,
- 10 * 1000,
- null,
- null,
+ p,
+ noteId,
+ MockInterpreterA.class.getName(),
+ new File(INTERPRETER_SCRIPT).getAbsolutePath(),
+ "fake",
+ "fakeRepo",
+ env,
+ 10 * 1000,
+ null,
+ null,
"anonymous",
false);
}
@@ -101,16 +101,16 @@ public class RemoteInterpreterTest {
private RemoteInterpreter createMockInterpreterB(Properties p, String noteId) {
return new RemoteInterpreter(
- p,
- noteId,
- MockInterpreterB.class.getName(),
- new File(INTERPRETER_SCRIPT).getAbsolutePath(),
- "fake",
- "fakeRepo",
- env,
- 10 * 1000,
- null,
- null,
+ p,
+ noteId,
+ MockInterpreterB.class.getName(),
+ new File(INTERPRETER_SCRIPT).getAbsolutePath(),
+ "fake",
+ "fakeRepo",
+ env,
+ 10 * 1000,
+ null,
+ null,
"anonymous",
false);
}
@@ -217,7 +217,6 @@ public class RemoteInterpreterTest {
"anonymous",
false);
-
intpGroup.get("note").add(intpA);
intpA.setInterpreterGroup(intpGroup);
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/test/java/org/apache/zeppelin/scheduler/RemoteSchedulerTest.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/scheduler/RemoteSchedulerTest.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/scheduler/RemoteSchedulerTest.java
index 4d64a3a..4ddb53d 100644
--- a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/scheduler/RemoteSchedulerTest.java
+++ b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/scheduler/RemoteSchedulerTest.java
@@ -311,4 +311,16 @@ public class RemoteSchedulerTest implements RemoteInterpreterProcessListener {
public void onMetaInfosReceived(String settingId, Map<String, String> metaInfos) {
}
+
+ @Override
+ public void onGetParagraphRunners(String noteId, String paragraphId, RemoteWorksEventListener callback) {
+ if (callback != null) {
+ callback.onFinished(new LinkedList<>());
+ }
+ }
+
+ @Override
+ public void onRemoteRunParagraph(String noteId, String PsaragraphID) throws Exception {
+
+ }
}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
----------------------------------------------------------------------
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
index c5a39b3..7094345 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
@@ -32,6 +32,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import javax.servlet.http.HttpServletRequest;
+import com.google.common.collect.Sets;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.vfs2.FileSystemException;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
@@ -41,6 +42,7 @@ import org.apache.zeppelin.display.AngularObjectRegistry;
import org.apache.zeppelin.display.AngularObjectRegistryListener;
import org.apache.zeppelin.helium.ApplicationEventListener;
import org.apache.zeppelin.helium.HeliumPackage;
+import org.apache.zeppelin.interpreter.InterpreterContextRunner;
import org.apache.zeppelin.interpreter.InterpreterGroup;
import org.apache.zeppelin.interpreter.InterpreterOutput;
import org.apache.zeppelin.interpreter.InterpreterResult;
@@ -59,6 +61,7 @@ import org.apache.zeppelin.notebook.repo.NotebookRepo.Revision;
import org.apache.zeppelin.notebook.socket.Message;
import org.apache.zeppelin.notebook.socket.Message.OP;
import org.apache.zeppelin.notebook.socket.WatcherMessage;
+import org.apache.zeppelin.rest.exception.ForbiddenException;
import org.apache.zeppelin.scheduler.Job;
import org.apache.zeppelin.scheduler.Job.Status;
import org.apache.zeppelin.server.ZeppelinServer;
@@ -1506,6 +1509,72 @@ public class NotebookServer extends WebSocketServlet implements
broadcast(noteId, msg);
}
+ @Override
+ public void onGetParagraphRunners(
+ String noteId, String paragraphId, RemoteWorksEventListener callback) {
+ Notebook notebookIns = notebook();
+ List<InterpreterContextRunner> runner = new LinkedList<>();
+
+ if (notebookIns == null) {
+ LOG.info("intepreter request notebook instance is null");
+ callback.onFinished(notebookIns);
+ }
+
+ try {
+ Note note = notebookIns.getNote(noteId);
+ if (note != null) {
+ if (paragraphId != null) {
+ Paragraph paragraph = note.getParagraph(paragraphId);
+ if (paragraph != null) {
+ runner.add(paragraph.getInterpreterContextRunner());
+ }
+ } else {
+ for (Paragraph p : note.getParagraphs()) {
+ runner.add(p.getInterpreterContextRunner());
+ }
+ }
+ }
+ callback.onFinished(runner);
+ } catch (NullPointerException e) {
+ LOG.warn(e.getMessage());
+ callback.onError();
+ }
+ }
+
+ @Override
+ public void onRemoteRunParagraph(String noteId, String paragraphId) throws Exception {
+ Notebook notebookIns = notebook();
+ try {
+ if (notebookIns == null) {
+ throw new Exception("onRemoteRunParagraph notebook instance is null");
+ }
+ Note noteIns = notebookIns.getNote(noteId);
+ if (noteIns == null) {
+ throw new Exception(String.format("Can't found note id %s", noteId));
+ }
+
+ Paragraph paragraph = noteIns.getParagraph(paragraphId);
+ if (paragraph == null) {
+ throw new Exception(String.format("Can't found paragraph %s %s", noteId, paragraphId));
+ }
+
+ Set<String> userAndRoles = Sets.newHashSet();
+ userAndRoles.add(SecurityUtils.getPrincipal());
+ userAndRoles.addAll(SecurityUtils.getRoles());
+ if (!notebookIns.getNotebookAuthorization().hasWriteAuthorization(userAndRoles, noteId)) {
+ throw new ForbiddenException(String.format("can't execute note %s", noteId));
+ }
+
+ AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
+ paragraph.setAuthenticationInfo(subject);
+
+ noteIns.run(paragraphId);
+
+ } catch (Exception e) {
+ throw e;
+ }
+ }
+
/**
* Notebook Information Change event
*/
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-server/src/test/java/org/apache/zeppelin/integration/ZeppelinIT.java
----------------------------------------------------------------------
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/integration/ZeppelinIT.java b/zeppelin-server/src/test/java/org/apache/zeppelin/integration/ZeppelinIT.java
index 1d60fce..49e1a9f 100644
--- a/zeppelin-server/src/test/java/org/apache/zeppelin/integration/ZeppelinIT.java
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/integration/ZeppelinIT.java
@@ -137,7 +137,7 @@ public class ZeppelinIT extends AbstractZeppelinIT {
* z.run(2, context)
* }
*/
- setTextOfParagraph(4, "z.angularWatch(\"myVar\", (before:Object, after:Object, context:org.apache.zeppelin.interpreter.InterpreterContext)=>{ z.run(2, context)})");
+ setTextOfParagraph(4, "z.angularWatch(\"myVar\", (before:Object, after:Object, context:org.apache.zeppelin.interpreter.InterpreterContext)=>{ z.run(2)})");
runParagraph(4);
waitForParagraph(4, "FINISHED");
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java
----------------------------------------------------------------------
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java
index 740ef40..cb1625a 100644
--- a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java
@@ -17,6 +17,7 @@
package org.apache.zeppelin.rest;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
import java.io.File;
@@ -346,6 +347,34 @@ public class ZeppelinSparkClusterTest extends AbstractTestRestApi {
assertEquals(Status.FINISHED, p2.getStatus());
assertEquals("10", p2.getResult().message());
+ Paragraph p3 = note.addParagraph();
+ Map config3 = p3.getConfig();
+ config3.put("enabled", true);
+ p3.setConfig(config3);
+ p3.setText("%spark println(new java.util.Date())");
+ p3.setAuthenticationInfo(anonymous);
+
+ p0.setText(String.format("%%spark z.runNote(\"%s\")", note.getId()));
+ note.run(p0.getId());
+ waitForFinish(p0);
+ waitForFinish(p1);
+ waitForFinish(p2);
+ waitForFinish(p3);
+
+ assertEquals(Status.FINISHED, p3.getStatus());
+ String p3result = p3.getResult().message();
+ assertNotEquals(null, p3result);
+ assertNotEquals("", p3result);
+
+ p0.setText(String.format("%%spark z.run(\"%s\", \"%s\")", note.getId(), p3.getId()));
+ p3.setText("%%spark println(\"END\")");
+
+ note.run(p0.getId());
+ waitForFinish(p0);
+ waitForFinish(p3);
+
+ assertNotEquals(p3result, p3.getResult().message());
+
ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
----------------------------------------------------------------------
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
index 20fe133..56d64ef 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
@@ -486,6 +486,11 @@ public class Paragraph extends Job implements Serializable, Cloneable {
return interpreterContext;
}
+ public InterpreterContextRunner getInterpreterContextRunner() {
+
+ return new ParagraphRunner(note, note.getId(), getId());
+ }
+
static class ParagraphRunner extends InterpreterContextRunner {
private transient Note note;
[2/2] zeppelin git commit: [ZEPPELIN-1665] Z.run with external note
executable and access resource for zeppelin in each interpreter
Posted by mo...@apache.org.
[ZEPPELIN-1665] Z.run with external note executable and access resource for zeppelin in each interpreter
### What is this PR for?
Currently, the z.run command is restricted.
Only paragraphs in a single note can be executed.
I have modified this to allow you to freely execute paragraphs of other notes.
This PR provides the basis for the freeful use of Zeppelin's resources at each Interpreter implementation.
### What type of PR is it?
Improvement, Feature
### Todos
- [x] extends z.run
- [x] run all paragraph in external note
- [x] run paragraph for external note.
- [x] get resource for zeppelin in each interpreter.
- [x] improve test case.
- [x] how to use docuement
### What is the Jira issue?
https://issues.apache.org/jira/browse/ZEPPELIN-1665
### How should this be tested?
Currently under development.
run paragraph in same note
```
%spark
z.run("paragraphID")
```
run paragraph with external note
```
z.run("noteid", "paragraphid");
```
all note run
```
z.runNote("noteid");
```
### Screenshots (if appropriate)
- paragraph run
- noterun
### Questions:
* Does the licenses files need update? no
* Is there breaking changes for older versions? no
* Does this needs documentation? yes
Author: CloverHearts <cl...@gmail.com>
Closes #1637 from cloverhearts/extends-zrun-remote-transaction and squashes the following commits:
41fa9d7 [CloverHearts] restore unless changed and import
113b475 [CloverHearts] Merge branch 'master' into extends-zrun-remote-transaction
03a3a2b [CloverHearts] testcase change z.run(2, context) to z.run(2)
2a2c173 [CloverHearts] Merge branch 'master' into extends-zrun-remote-transaction
f2e3bcf [CloverHearts] fix TestCase
5a80a5a [CloverHearts] last test case time check to print string
e6cd82c [CloverHearts] Merge branch 'master' into extends-zrun-remote-transaction
3862166 [CloverHearts] regenerate thrfit class
5ec4640 [CloverHearts] change defined protocol for thrift
7562535 [CloverHearts] remove unused import and asterisk import
8a54917 [CloverHearts] Merge branch 'master' into extends-zrun-remote-transaction
342752d [CloverHearts] add document for extends z.run and z.runNote
292319a [CloverHearts] add test case for extends z.run and z.runNote
10c2a47 [CloverHearts] Implement runNote and re implement run method
f9661c8 [CloverHearts] Merge branch 'master' into extends-zrun-remote-transaction
9ab05af [CloverHearts] Change structure and remove remoteWorksManager
8cbe46c [CloverHearts] remote remoteworksController in interpreter.java
8d42c16 [CloverHearts] Merge branch 'master' into extends-zrun-remote-transaction
f11fed4 [CloverHearts] Merge branch 'workflow' into extends-zrun-remote-transaction
c074f07 [CloverHearts] fix sio support
4b1ef08 [CloverHearts] fix thrift interface
2628a20 [CloverHearts] fix thrift
6fbe08a [CloverHearts] Merge branch 'master' into workflow
3f75bd5 [CloverHearts] support scald
55e8704 [CloverHearts] support spark r
5a7886f [CloverHearts] fix sio support
afb9db7 [CloverHearts] Merge branch 'master' into workflow
3ed556c [CloverHearts] remove debug console message.
3d34f9e [CloverHearts] Implement getParagraphRunner transaction.
2523238 [CloverHearts] Implement eventForWait class
0570ae8 [CloverHearts] add remote works controller class and include interpreter factory
6e1f219 [CloverHearts] code base workflow for remote zeppelin server control default thrift transaction.
Project: http://git-wip-us.apache.org/repos/asf/zeppelin/repo
Commit: http://git-wip-us.apache.org/repos/asf/zeppelin/commit/9db840c6
Tree: http://git-wip-us.apache.org/repos/asf/zeppelin/tree/9db840c6
Diff: http://git-wip-us.apache.org/repos/asf/zeppelin/diff/9db840c6
Branch: refs/heads/master
Commit: 9db840c6555e0fae60b3eca8f2af59d1c87ed036
Parents: 6a2a341
Author: CloverHearts <cl...@gmail.com>
Authored: Tue Nov 29 00:09:33 2016 +0900
Committer: Lee moon soo <mo...@apache.org>
Committed: Tue Nov 29 08:07:15 2016 -0800
----------------------------------------------------------------------
docs/displaysystem/front-end-angular.md | 12 +-
.../apache/zeppelin/spark/ZeppelinContext.java | 119 ++-
.../interpreter/InterpreterContext.java | 32 +-
.../interpreter/RemoteWorksController.java | 29 +
.../RemoteZeppelinServerResource.java | 58 ++
.../remote/RemoteInterpreterEventClient.java | 18 +
.../remote/RemoteInterpreterEventPoller.java | 87 ++-
.../RemoteInterpreterProcessListener.java | 11 +
.../remote/RemoteInterpreterServer.java | 121 ++-
.../thrift/InterpreterCompletion.java | 19 +-
.../thrift/RemoteApplicationResult.java | 19 +-
.../thrift/RemoteInterpreterContext.java | 19 +-
.../thrift/RemoteInterpreterEvent.java | 19 +-
.../thrift/RemoteInterpreterEventType.java | 22 +-
.../thrift/RemoteInterpreterResult.java | 19 +-
.../thrift/RemoteInterpreterService.java | 756 ++++++++++++++++++-
.../ZeppelinServerResourceParagraphRunner.java | 520 +++++++++++++
.../main/thrift/RemoteInterpreterService.thrift | 11 +-
.../RemoteInterpreterOutputTestStream.java | 12 +
.../remote/RemoteInterpreterTest.java | 41 +-
.../zeppelin/scheduler/RemoteSchedulerTest.java | 12 +
.../apache/zeppelin/socket/NotebookServer.java | 69 ++
.../apache/zeppelin/integration/ZeppelinIT.java | 2 +-
.../zeppelin/rest/ZeppelinSparkClusterTest.java | 29 +
.../org/apache/zeppelin/notebook/Paragraph.java | 5 +
25 files changed, 2000 insertions(+), 61 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/docs/displaysystem/front-end-angular.md
----------------------------------------------------------------------
diff --git a/docs/displaysystem/front-end-angular.md b/docs/displaysystem/front-end-angular.md
index 0e64aae..973ed88 100644
--- a/docs/displaysystem/front-end-angular.md
+++ b/docs/displaysystem/front-end-angular.md
@@ -150,7 +150,17 @@ How does the front-end AngularJS API compares to the [back-end API](./back-end-a
<td>Executing Paragraph</td>
<td>z.runParagraph(paragraphId)</td>
<td>z.run(paragraphId)</td>
- </tr>
+ </tr>
+ <tr>
+ <td>Executing Paragraph (Specific paragraphs in other notes) (</td>
+ <td></td>
+ <td>z.run(noteid, paragraphId)</td>
+ </tr>
+ <tr>
+ <td>Executing note</td>
+ <td></td>
+ <td>z.runNote(noteId)</td>
+ </tr>
<tbody>
<tbody>
</table>
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinContext.java
----------------------------------------------------------------------
diff --git a/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinContext.java b/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinContext.java
index ff21c7a..281a8f8 100644
--- a/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinContext.java
+++ b/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinContext.java
@@ -45,6 +45,7 @@ import org.apache.zeppelin.interpreter.InterpreterContext;
import org.apache.zeppelin.interpreter.InterpreterContextRunner;
import org.apache.zeppelin.interpreter.InterpreterException;
import org.apache.zeppelin.interpreter.InterpreterHookRegistry;
+import org.apache.zeppelin.interpreter.RemoteWorksController;
import org.apache.zeppelin.spark.dep.SparkDependencyResolver;
import org.apache.zeppelin.resource.Resource;
import org.apache.zeppelin.resource.ResourcePool;
@@ -300,32 +301,103 @@ public class ZeppelinContext {
/**
* Run paragraph by id
- * @param id
+ * @param noteId
+ * @param paragraphId
*/
@ZeppelinApi
- public void run(String id) {
- run(id, interpreterContext);
+ public void run(String noteId, String paragraphId) {
+ run(noteId, paragraphId, interpreterContext);
}
/**
* Run paragraph by id
- * @param id
+ * @param paragraphId
+ */
+ @ZeppelinApi
+ public void run(String paragraphId) {
+ String noteId = interpreterContext.getNoteId();
+ run(noteId, paragraphId, interpreterContext);
+ }
+
+ /**
+ * Run paragraph by id
+ * @param noteId
* @param context
*/
@ZeppelinApi
- public void run(String id, InterpreterContext context) {
- if (id.equals(context.getParagraphId())) {
+ public void run(String noteId, String paragraphId, InterpreterContext context) {
+ if (paragraphId.equals(context.getParagraphId())) {
throw new InterpreterException("Can not run current Paragraph");
}
- for (InterpreterContextRunner r : context.getRunners()) {
- if (id.equals(r.getParagraphId())) {
- r.run();
- return;
+ List<InterpreterContextRunner> runners =
+ getInterpreterContextRunner(noteId, paragraphId, context);
+
+ if (runners.size() <= 0) {
+ throw new InterpreterException("Paragraph " + paragraphId + " not found " + runners.size());
+ }
+
+ for (InterpreterContextRunner r : runners) {
+ r.run();
+ }
+
+ }
+
+ public void runNote(String noteId) {
+ runNote(noteId, interpreterContext);
+ }
+
+ public void runNote(String noteId, InterpreterContext context) {
+ String runningNoteId = context.getNoteId();
+ String runningParagraphId = context.getParagraphId();
+ List<InterpreterContextRunner> runners = getInterpreterContextRunner(noteId, context);
+
+ if (runners.size() <= 0) {
+ throw new InterpreterException("Note " + noteId + " not found " + runners.size());
+ }
+
+ for (InterpreterContextRunner r : runners) {
+ if (r.getNoteId().equals(runningNoteId) && r.getParagraphId().equals(runningParagraphId)) {
+ continue;
}
+ r.run();
}
+ }
+
+
+ /**
+ * get Zeppelin Paragraph Runner from zeppelin server
+ * @param noteId
+ */
+ @ZeppelinApi
+ public List<InterpreterContextRunner> getInterpreterContextRunner(
+ String noteId, InterpreterContext interpreterContext) {
+ List<InterpreterContextRunner> runners = new LinkedList<>();
+ RemoteWorksController remoteWorksController = interpreterContext.getRemoteWorksController();
- throw new InterpreterException("Paragraph " + id + " not found");
+ if (remoteWorksController != null) {
+ runners = remoteWorksController.getRemoteContextRunner(noteId);
+ }
+
+ return runners;
+ }
+
+ /**
+ * get Zeppelin Paragraph Runner from zeppelin server
+ * @param noteId
+ * @param paragraphId
+ */
+ @ZeppelinApi
+ public List<InterpreterContextRunner> getInterpreterContextRunner(
+ String noteId, String paragraphId, InterpreterContext interpreterContext) {
+ List<InterpreterContextRunner> runners = new LinkedList<>();
+ RemoteWorksController remoteWorksController = interpreterContext.getRemoteWorksController();
+
+ if (remoteWorksController != null) {
+ runners = remoteWorksController.getRemoteContextRunner(noteId, paragraphId);
+ }
+
+ return runners;
}
/**
@@ -334,7 +406,8 @@ public class ZeppelinContext {
*/
@ZeppelinApi
public void run(int idx) {
- run(idx, interpreterContext);
+ String noteId = interpreterContext.getNoteId();
+ run(noteId, idx, interpreterContext);
}
/**
@@ -342,12 +415,13 @@ public class ZeppelinContext {
* @param idx index starting from 0
* @param context interpreter context
*/
- public void run(int idx, InterpreterContext context) {
- if (idx >= context.getRunners().size()) {
+ public void run(String noteId, int idx, InterpreterContext context) {
+ List<InterpreterContextRunner> runners = getInterpreterContextRunner(noteId, context);
+ if (idx >= runners.size()) {
throw new InterpreterException("Index out of bound");
}
- InterpreterContextRunner runner = context.getRunners().get(idx);
+ InterpreterContextRunner runner = runners.get(idx);
if (runner.getParagraphId().equals(context.getParagraphId())) {
throw new InterpreterException("Can not run current Paragraph");
}
@@ -366,13 +440,14 @@ public class ZeppelinContext {
*/
@ZeppelinApi
public void run(List<Object> paragraphIdOrIdx, InterpreterContext context) {
+ String noteId = context.getNoteId();
for (Object idOrIdx : paragraphIdOrIdx) {
if (idOrIdx instanceof String) {
- String id = (String) idOrIdx;
- run(id, context);
+ String paragraphId = (String) idOrIdx;
+ run(noteId, paragraphId, context);
} else if (idOrIdx instanceof Integer) {
Integer idx = (Integer) idOrIdx;
- run(idx, context);
+ run(noteId, idx, context);
} else {
throw new InterpreterException("Paragraph " + idOrIdx + " not found");
}
@@ -389,13 +464,7 @@ public class ZeppelinContext {
*/
@ZeppelinApi
public void runAll(InterpreterContext context) {
- for (InterpreterContextRunner r : context.getRunners()) {
- if (r.getParagraphId().equals(context.getParagraphId())) {
- // skip itself
- continue;
- }
- r.run();
- }
+ runNote(context.getNoteId());
}
@ZeppelinApi
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterContext.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterContext.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterContext.java
index db540aa..3f36405 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterContext.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterContext.java
@@ -61,6 +61,7 @@ public class InterpreterContext {
private List<InterpreterContextRunner> runners;
private String className;
private RemoteEventClientWrapper client;
+ private RemoteWorksController remoteWorksController;
public InterpreterContext(String noteId,
String paragraphId,
@@ -75,6 +76,24 @@ public class InterpreterContext {
List<InterpreterContextRunner> runners,
InterpreterOutput out
) {
+ this(noteId, paragraphId, replName, paragraphTitle, paragraphText, authenticationInfo,
+ config, gui, angularObjectRegistry, resourcePool, runners, out, null);
+ }
+
+ public InterpreterContext(String noteId,
+ String paragraphId,
+ String replName,
+ String paragraphTitle,
+ String paragraphText,
+ AuthenticationInfo authenticationInfo,
+ Map<String, Object> config,
+ GUI gui,
+ AngularObjectRegistry angularObjectRegistry,
+ ResourcePool resourcePool,
+ List<InterpreterContextRunner> runners,
+ InterpreterOutput out,
+ RemoteWorksController remoteWorksController
+ ) {
this.noteId = noteId;
this.paragraphId = paragraphId;
this.replName = replName;
@@ -87,6 +106,7 @@ public class InterpreterContext {
this.resourcePool = resourcePool;
this.runners = runners;
this.out = out;
+ this.remoteWorksController = remoteWorksController;
}
public InterpreterContext(String noteId,
@@ -101,9 +121,11 @@ public class InterpreterContext {
ResourcePool resourcePool,
List<InterpreterContextRunner> contextRunners,
InterpreterOutput output,
+ RemoteWorksController remoteWorksController,
RemoteInterpreterEventClient eventClient) {
this(noteId, paragraphId, replName, paragraphTitle, paragraphText, authenticationInfo,
- config, gui, angularObjectRegistry, resourcePool, contextRunners, output);
+ config, gui, angularObjectRegistry, resourcePool, contextRunners, output,
+ remoteWorksController);
this.client = new RemoteEventClient(eventClient);
}
@@ -162,4 +184,12 @@ public class InterpreterContext {
public RemoteEventClientWrapper getClient() {
return client;
}
+
+ public RemoteWorksController getRemoteWorksController() {
+ return remoteWorksController;
+ }
+
+ public void setRemoteWorksController(RemoteWorksController remoteWorksController) {
+ this.remoteWorksController = remoteWorksController;
+ }
}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/RemoteWorksController.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/RemoteWorksController.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/RemoteWorksController.java
new file mode 100644
index 0000000..e1410d6
--- /dev/null
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/RemoteWorksController.java
@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+import java.util.List;
+
+/**
+ * zeppelin job for Remote works controller by interpreter
+ *
+ */
+public interface RemoteWorksController {
+ List<InterpreterContextRunner> getRemoteContextRunner(String noteId);
+ List<InterpreterContextRunner> getRemoteContextRunner(String noteId, String paragraphId);
+}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/RemoteZeppelinServerResource.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/RemoteZeppelinServerResource.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/RemoteZeppelinServerResource.java
new file mode 100644
index 0000000..b2a87aa
--- /dev/null
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/RemoteZeppelinServerResource.java
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+/**
+ * Remote Zeppelin Server Resource
+ */
+public class RemoteZeppelinServerResource {
+ /**
+ * Resource Type for Zeppelin Server
+ */
+ public enum Type{
+ PARAGRAPH_RUNNERS
+ }
+
+ private String ownerKey;
+ private Type resourceType;
+ private Object data;
+
+ public Type getResourceType() {
+ return resourceType;
+ }
+
+ public String getOwnerKey() {
+ return ownerKey;
+ }
+
+ public void setOwnerKey(String ownerKey) {
+ this.ownerKey = ownerKey;
+ }
+
+ public void setResourceType(Type resourceType) {
+ this.resourceType = resourceType;
+ }
+
+ public Object getData() {
+ return data;
+ }
+
+ public void setData(Object data) {
+ this.data = data;
+ }
+}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java
index ae38ee8..867b726 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java
@@ -19,8 +19,10 @@ package org.apache.zeppelin.interpreter.remote;
import com.google.gson.Gson;
import org.apache.zeppelin.display.AngularObject;
import org.apache.zeppelin.interpreter.InterpreterContextRunner;
+import org.apache.zeppelin.interpreter.RemoteZeppelinServerResource;
import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterEvent;
import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterEventType;
+import org.apache.zeppelin.interpreter.thrift.ZeppelinServerResourceParagraphRunner;
import org.apache.zeppelin.resource.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,6 +54,22 @@ public class RemoteInterpreterEventClient implements ResourcePoolConnector {
* Run paragraph
* @param runner
*/
+ public void getZeppelinServerNoteRunner(
+ String eventOwnerKey, ZeppelinServerResourceParagraphRunner runner) {
+ RemoteZeppelinServerResource eventBody = new RemoteZeppelinServerResource();
+ eventBody.setResourceType(RemoteZeppelinServerResource.Type.PARAGRAPH_RUNNERS);
+ eventBody.setOwnerKey(eventOwnerKey);
+ eventBody.setData(runner);
+
+ sendEvent(new RemoteInterpreterEvent(
+ RemoteInterpreterEventType.REMOTE_ZEPPELIN_SERVER_RESOURCE,
+ gson.toJson(eventBody)));
+ }
+
+ /**
+ * Run paragraph
+ * @param runner
+ */
public void run(InterpreterContextRunner runner) {
sendEvent(new RemoteInterpreterEvent(
RemoteInterpreterEventType.RUN_INTERPRETER_CONTEXT_RUNNER,
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventPoller.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventPoller.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventPoller.java
index b75e5fa..33cc8e4 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventPoller.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventPoller.java
@@ -25,9 +25,11 @@ import org.apache.zeppelin.display.AngularObjectRegistry;
import org.apache.zeppelin.helium.ApplicationEventListener;
import org.apache.zeppelin.interpreter.InterpreterContextRunner;
import org.apache.zeppelin.interpreter.InterpreterGroup;
+import org.apache.zeppelin.interpreter.RemoteZeppelinServerResource;
import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterEvent;
import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterEventType;
import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterService.Client;
+import org.apache.zeppelin.interpreter.thrift.ZeppelinServerResourceParagraphRunner;
import org.apache.zeppelin.resource.Resource;
import org.apache.zeppelin.resource.ResourceId;
import org.apache.zeppelin.resource.ResourcePool;
@@ -145,8 +147,9 @@ public class RemoteInterpreterEventPoller extends Thread {
InterpreterContextRunner runnerFromRemote = gson.fromJson(
event.getData(), RemoteInterpreterContextRunner.class);
- interpreterProcess.getInterpreterContextRunnerPool().run(
+ listener.onRemoteRunParagraph(
runnerFromRemote.getNoteId(), runnerFromRemote.getParagraphId());
+
} else if (event.getType() == RemoteInterpreterEventType.RESOURCE_POOL_GET_ALL) {
ResourceSet resourceSet = getAllResourcePoolExcept();
sendResourcePoolResponseGetAll(resourceSet);
@@ -195,6 +198,12 @@ public class RemoteInterpreterEventPoller extends Thread {
String status = appStatusUpdate.get("status");
appListener.onStatusChange(noteId, paragraphId, appId, status);
+ } else if (event.getType() == RemoteInterpreterEventType.REMOTE_ZEPPELIN_SERVER_RESOURCE) {
+ RemoteZeppelinServerResource reqResourceBody = gson.fromJson(
+ event.getData(), RemoteZeppelinServerResource.class);
+ progressRemoteZeppelinControlEvent(
+ reqResourceBody.getResourceType(), listener, reqResourceBody);
+
} else if (event.getType() == RemoteInterpreterEventType.META_INFOS) {
Map<String, String> metaInfos = gson.fromJson(event.getData(),
new TypeToken<Map<String, String>>() {
@@ -214,6 +223,82 @@ public class RemoteInterpreterEventPoller extends Thread {
}
}
+ private void progressRemoteZeppelinControlEvent(
+ RemoteZeppelinServerResource.Type resourceType,
+ RemoteInterpreterProcessListener remoteWorksEventListener,
+ RemoteZeppelinServerResource reqResourceBody) throws Exception {
+ boolean broken = false;
+ final Gson gson = new Gson();
+ final String eventOwnerKey = reqResourceBody.getOwnerKey();
+ Client interpreterServerMain = null;
+ try {
+ interpreterServerMain = interpreterProcess.getClient();
+ final Client eventClient = interpreterServerMain;
+ if (resourceType == RemoteZeppelinServerResource.Type.PARAGRAPH_RUNNERS) {
+ final List<ZeppelinServerResourceParagraphRunner> remoteRunners = new LinkedList<>();
+
+ ZeppelinServerResourceParagraphRunner reqRunnerContext =
+ new ZeppelinServerResourceParagraphRunner();
+
+ Map<String, Object> reqResourceMap = (Map<String, Object>) reqResourceBody.getData();
+ String noteId = (String) reqResourceMap.get("noteId");
+ String paragraphId = (String) reqResourceMap.get("paragraphId");
+
+ reqRunnerContext.setNoteId(noteId);
+ reqRunnerContext.setParagraphId(paragraphId);
+
+ RemoteInterpreterProcessListener.RemoteWorksEventListener callBackEvent =
+ new RemoteInterpreterProcessListener.RemoteWorksEventListener() {
+
+ @Override
+ public void onFinished(Object resultObject) {
+ boolean clientBroken = false;
+ if (resultObject != null && resultObject instanceof List) {
+ List<InterpreterContextRunner> runnerList =
+ (List<InterpreterContextRunner>) resultObject;
+ for (InterpreterContextRunner r : runnerList) {
+ remoteRunners.add(
+ new ZeppelinServerResourceParagraphRunner(r.getNoteId(), r.getParagraphId())
+ );
+ }
+
+ final RemoteZeppelinServerResource resResource =
+ new RemoteZeppelinServerResource();
+ resResource.setOwnerKey(eventOwnerKey);
+ resResource.setResourceType(RemoteZeppelinServerResource.Type.PARAGRAPH_RUNNERS);
+ resResource.setData(remoteRunners);
+
+ try {
+ eventClient.onReceivedZeppelinResource(gson.toJson(resResource));
+ } catch (Exception e) {
+ clientBroken = true;
+ logger.error("Can't get RemoteInterpreterEvent", e);
+ waitQuietly();
+ } finally {
+ interpreterProcess.releaseClient(eventClient, clientBroken);
+ }
+ }
+ }
+
+ @Override
+ public void onError() {
+ logger.info("onGetParagraphRunners onError");
+ }
+ };
+
+ remoteWorksEventListener.onGetParagraphRunners(
+ reqRunnerContext.getNoteId(), reqRunnerContext.getParagraphId(), callBackEvent);
+ }
+ } catch (Exception e) {
+ broken = true;
+ logger.error("Can't get RemoteInterpreterEvent", e);
+ waitQuietly();
+
+ } finally {
+ interpreterProcess.releaseClient(interpreterServerMain, broken);
+ }
+ }
+
private void sendResourcePoolResponseGetAll(ResourceSet resourceSet) {
Client client = null;
boolean broken = false;
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcessListener.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcessListener.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcessListener.java
index d25683f..763e14c 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcessListener.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcessListener.java
@@ -25,4 +25,15 @@ public interface RemoteInterpreterProcessListener {
public void onOutputAppend(String noteId, String paragraphId, String output);
public void onOutputUpdated(String noteId, String paragraphId, String output);
public void onMetaInfosReceived(String settingId, Map<String, String> metaInfos);
+ public void onRemoteRunParagraph(String noteId, String ParagraphID) throws Exception;
+ public void onGetParagraphRunners(
+ String noteId, String paragraphId, RemoteWorksEventListener callback);
+
+ /**
+ * Remote works for Interpreter callback listener
+ */
+ public interface RemoteWorksEventListener {
+ public void onFinished(Object resultObject);
+ public void onError();
+ }
}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
index 4e2d5bf..50a1f7c 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
@@ -23,7 +23,10 @@ import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.net.URL;
import java.nio.ByteBuffer;
+import java.rmi.server.RemoteServer;
import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeoutException;
import org.apache.thrift.TException;
import org.apache.thrift.server.TThreadPoolServer;
@@ -78,6 +81,9 @@ public class RemoteInterpreterServer
private final Map<String, RunningApplication> runningApplications =
Collections.synchronizedMap(new HashMap<String, RunningApplication>());
+ private Map<String, Object> remoteWorksResponsePool;
+ private ZeppelinRemoteWorksController remoteWorksController;
+
public RemoteInterpreterServer(int port) throws TTransportException {
this.port = port;
@@ -85,6 +91,8 @@ public class RemoteInterpreterServer
TServerSocket serverTransport = new TServerSocket(port);
server = new TThreadPoolServer(
new TThreadPoolServer.Args(serverTransport).processor(processor));
+ remoteWorksResponsePool = Collections.synchronizedMap(new HashMap<String, Object>());
+ remoteWorksController = new ZeppelinRemoteWorksController(this, remoteWorksResponsePool);
}
@Override
@@ -174,7 +182,6 @@ public class RemoteInterpreterServer
Constructor<Interpreter> constructor =
replClass.getConstructor(new Class[] {Properties.class});
Interpreter repl = constructor.newInstance(p);
-
repl.setClassloaderUrls(new URL[]{});
synchronized (interpreterGroup) {
@@ -335,6 +342,42 @@ public class RemoteInterpreterServer
context.getGui());
}
+ @Override
+ public void onReceivedZeppelinResource(String responseJson) throws TException {
+ RemoteZeppelinServerResource response = gson.fromJson(
+ responseJson, RemoteZeppelinServerResource.class);
+
+ if (response == null) {
+ throw new TException("Bad response for remote resource");
+ }
+
+ try {
+ if (response.getResourceType() == RemoteZeppelinServerResource.Type.PARAGRAPH_RUNNERS) {
+ List<InterpreterContextRunner> intpContextRunners = new LinkedList<>();
+ List<Map<String, Object>> remoteRunnersMap =
+ (List<Map<String, Object>>) response.getData();
+
+ String noteId = null;
+ String paragraphId = null;
+
+ for (Map<String, Object> runnerItem : remoteRunnersMap) {
+ noteId = (String) runnerItem.get("noteId");
+ paragraphId = (String) runnerItem.get("paragraphId");
+ intpContextRunners.add(
+ new ParagraphRunner(this, noteId, paragraphId)
+ );
+ }
+
+ synchronized (this.remoteWorksResponsePool) {
+ this.remoteWorksResponsePool.put(
+ response.getOwnerKey(),
+ intpContextRunners);
+ }
+ }
+ } catch (Exception e) {
+ throw e;
+ }
+ }
class InterpretJobListener implements JobListener {
@@ -552,7 +595,7 @@ public class RemoteInterpreterServer
gson.fromJson(ric.getGui(), GUI.class),
interpreterGroup.getAngularObjectRegistry(),
interpreterGroup.getResourcePool(),
- contextRunners, output, eventClient);
+ contextRunners, output, remoteWorksController, eventClient);
}
@@ -575,7 +618,7 @@ public class RemoteInterpreterServer
static class ParagraphRunner extends InterpreterContextRunner {
-
+ Logger logger = LoggerFactory.getLogger(ParagraphRunner.class);
private transient RemoteInterpreterServer server;
public ParagraphRunner(RemoteInterpreterServer server, String noteId, String paragraphId) {
@@ -589,6 +632,78 @@ public class RemoteInterpreterServer
}
}
+ static class ZeppelinRemoteWorksController implements RemoteWorksController{
+ Logger logger = LoggerFactory.getLogger(ZeppelinRemoteWorksController.class);
+
+ private final long DEFAULT_TIMEOUT_VALUE = 300000;
+ private final Map<String, Object> remoteWorksResponsePool;
+ private RemoteInterpreterServer server;
+ public ZeppelinRemoteWorksController(
+ RemoteInterpreterServer server, Map<String, Object> remoteWorksResponsePool) {
+ this.remoteWorksResponsePool = remoteWorksResponsePool;
+ this.server = server;
+ }
+
+ public String generateOwnerKey() {
+ String hashKeyText = new String("ownerKey" + System.currentTimeMillis());
+ String hashKey = String.valueOf(hashKeyText.hashCode());
+ return hashKey;
+ }
+
+ public boolean waitForEvent(String eventOwnerKey) throws InterruptedException {
+ return waitForEvent(eventOwnerKey, DEFAULT_TIMEOUT_VALUE);
+ }
+
+ public boolean waitForEvent(String eventOwnerKey, long timeout) throws InterruptedException {
+ boolean wasGetData = false;
+ long now = System.currentTimeMillis();
+ long endTime = System.currentTimeMillis() + timeout;
+
+ while (endTime >= now) {
+ synchronized (this.remoteWorksResponsePool) {
+ wasGetData = this.remoteWorksResponsePool.containsKey(eventOwnerKey);
+ }
+ if (wasGetData == true) {
+ break;
+ }
+ now = System.currentTimeMillis();
+ sleep(500);
+ }
+
+ return wasGetData;
+ }
+
+ @Override
+ public List<InterpreterContextRunner> getRemoteContextRunner(String noteId) {
+ return getRemoteContextRunner(noteId, null);
+ }
+
+ public List<InterpreterContextRunner> getRemoteContextRunner(
+ String noteId, String paragraphID) {
+
+ List<InterpreterContextRunner> runners = null;
+ String ownerKey = generateOwnerKey();
+
+ ZeppelinServerResourceParagraphRunner resource = new ZeppelinServerResourceParagraphRunner();
+ resource.setNoteId(noteId);
+ resource.setParagraphId(paragraphID);
+ server.eventClient.getZeppelinServerNoteRunner(ownerKey, resource);
+
+ try {
+ this.waitForEvent(ownerKey);
+ } catch (Exception e) {
+ return new LinkedList<>();
+ }
+ synchronized (this.remoteWorksResponsePool) {
+ runners = (List<InterpreterContextRunner>) this.remoteWorksResponsePool.get(ownerKey);
+ this.remoteWorksResponsePool.remove(ownerKey);
+ }
+ return runners;
+ }
+
+
+ }
+
private RemoteInterpreterResult convert(InterpreterResult result,
Map<String, Object> config, GUI gui) {
return new RemoteInterpreterResult(
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterCompletion.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterCompletion.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterCompletion.java
index 04d345a..2514cb0 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterCompletion.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterCompletion.java
@@ -1,4 +1,21 @@
/**
+ * 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.
+ */
+/**
* Autogenerated by Thrift Compiler (0.9.2)
*
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
@@ -34,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-18")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-25")
public class InterpreterCompletion implements org.apache.thrift.TBase<InterpreterCompletion, InterpreterCompletion._Fields>, java.io.Serializable, Cloneable, Comparable<InterpreterCompletion> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("InterpreterCompletion");
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteApplicationResult.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteApplicationResult.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteApplicationResult.java
index fc0670c..3d9aae5 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteApplicationResult.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteApplicationResult.java
@@ -1,4 +1,21 @@
/**
+ * 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.
+ */
+/**
* Autogenerated by Thrift Compiler (0.9.2)
*
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
@@ -34,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-18")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-25")
public class RemoteApplicationResult implements org.apache.thrift.TBase<RemoteApplicationResult, RemoteApplicationResult._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteApplicationResult> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteApplicationResult");
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterContext.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterContext.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterContext.java
index 4c4439a..eb7c2a0 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterContext.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterContext.java
@@ -1,4 +1,21 @@
/**
+ * 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.
+ */
+/**
* Autogenerated by Thrift Compiler (0.9.2)
*
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
@@ -34,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-18")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-25")
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");
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEvent.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEvent.java
index 4cb32bc..9ff5a34 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEvent.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEvent.java
@@ -1,4 +1,21 @@
/**
+ * 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.
+ */
+/**
* Autogenerated by Thrift Compiler (0.9.2)
*
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
@@ -34,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-18")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-25")
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");
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventType.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventType.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventType.java
index 0c20e55..13f1236 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventType.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventType.java
@@ -1,4 +1,21 @@
/**
+ * 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.
+ */
+/**
* Autogenerated by Thrift Compiler (0.9.2)
*
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
@@ -23,7 +40,8 @@ public enum RemoteInterpreterEventType implements org.apache.thrift.TEnum {
OUTPUT_UPDATE(9),
ANGULAR_REGISTRY_PUSH(10),
APP_STATUS_UPDATE(11),
- META_INFOS(12);
+ META_INFOS(12),
+ REMOTE_ZEPPELIN_SERVER_RESOURCE(13);
private final int value;
@@ -68,6 +86,8 @@ public enum RemoteInterpreterEventType implements org.apache.thrift.TEnum {
return APP_STATUS_UPDATE;
case 12:
return META_INFOS;
+ case 13:
+ return REMOTE_ZEPPELIN_SERVER_RESOURCE;
default:
return null;
}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResult.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResult.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResult.java
index 2ed58ad..1a20f39 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResult.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResult.java
@@ -1,4 +1,21 @@
/**
+ * 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.
+ */
+/**
* Autogenerated by Thrift Compiler (0.9.2)
*
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
@@ -34,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-18")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-25")
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");
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9db840c6/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterService.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterService.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterService.java
index a562bf4..0b7930d 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterService.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterService.java
@@ -1,4 +1,21 @@
/**
+ * 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.
+ */
+/**
* Autogenerated by Thrift Compiler (0.9.2)
*
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
@@ -34,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-18")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2016-11-25")
public class RemoteInterpreterService {
public interface Iface {
@@ -85,6 +102,8 @@ public class RemoteInterpreterService {
public RemoteApplicationResult runApplication(String applicationInstanceId) throws org.apache.thrift.TException;
+ public void onReceivedZeppelinResource(String object) throws org.apache.thrift.TException;
+
}
public interface AsyncIface {
@@ -135,6 +154,8 @@ public class RemoteInterpreterService {
public void runApplication(String applicationInstanceId, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException;
+ public void onReceivedZeppelinResource(String object, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException;
+
}
public static class Client extends org.apache.thrift.TServiceClient implements Iface {
@@ -683,6 +704,26 @@ public class RemoteInterpreterService {
throw new org.apache.thrift.TApplicationException(org.apache.thrift.TApplicationException.MISSING_RESULT, "runApplication failed: unknown result");
}
+ public void onReceivedZeppelinResource(String object) throws org.apache.thrift.TException
+ {
+ send_onReceivedZeppelinResource(object);
+ recv_onReceivedZeppelinResource();
+ }
+
+ public void send_onReceivedZeppelinResource(String object) throws org.apache.thrift.TException
+ {
+ onReceivedZeppelinResource_args args = new onReceivedZeppelinResource_args();
+ args.setObject(object);
+ sendBase("onReceivedZeppelinResource", args);
+ }
+
+ public void recv_onReceivedZeppelinResource() throws org.apache.thrift.TException
+ {
+ onReceivedZeppelinResource_result result = new onReceivedZeppelinResource_result();
+ receiveBase(result, "onReceivedZeppelinResource");
+ return;
+ }
+
}
public static class AsyncClient extends org.apache.thrift.async.TAsyncClient implements AsyncIface {
public static class Factory implements org.apache.thrift.async.TAsyncClientFactory<AsyncClient> {
@@ -1527,6 +1568,38 @@ public class RemoteInterpreterService {
}
}
+ public void onReceivedZeppelinResource(String object, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException {
+ checkReady();
+ onReceivedZeppelinResource_call method_call = new onReceivedZeppelinResource_call(object, resultHandler, this, ___protocolFactory, ___transport);
+ this.___currentMethod = method_call;
+ ___manager.call(method_call);
+ }
+
+ public static class onReceivedZeppelinResource_call extends org.apache.thrift.async.TAsyncMethodCall {
+ private String object;
+ public onReceivedZeppelinResource_call(String object, org.apache.thrift.async.AsyncMethodCallback resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException {
+ super(client, protocolFactory, transport, resultHandler, false);
+ this.object = object;
+ }
+
+ public void write_args(org.apache.thrift.protocol.TProtocol prot) throws org.apache.thrift.TException {
+ prot.writeMessageBegin(new org.apache.thrift.protocol.TMessage("onReceivedZeppelinResource", org.apache.thrift.protocol.TMessageType.CALL, 0));
+ onReceivedZeppelinResource_args args = new onReceivedZeppelinResource_args();
+ args.setObject(object);
+ args.write(prot);
+ prot.writeMessageEnd();
+ }
+
+ public void getResult() throws org.apache.thrift.TException {
+ if (getState() != org.apache.thrift.async.TAsyncMethodCall.State.RESPONSE_READ) {
+ throw new IllegalStateException("Method call not finished!");
+ }
+ org.apache.thrift.transport.TMemoryInputTransport memoryTransport = new org.apache.thrift.transport.TMemoryInputTransport(getFrameBuffer().array());
+ org.apache.thrift.protocol.TProtocol prot = client.getProtocolFactory().getProtocol(memoryTransport);
+ (new Client(prot)).recv_onReceivedZeppelinResource();
+ }
+ }
+
}
public static class Processor<I extends Iface> extends org.apache.thrift.TBaseProcessor<I> implements org.apache.thrift.TProcessor {
@@ -1563,6 +1636,7 @@ public class RemoteInterpreterService {
processMap.put("loadApplication", new loadApplication());
processMap.put("unloadApplication", new unloadApplication());
processMap.put("runApplication", new runApplication());
+ processMap.put("onReceivedZeppelinResource", new onReceivedZeppelinResource());
return processMap;
}
@@ -2028,6 +2102,26 @@ public class RemoteInterpreterService {
}
}
+ public static class onReceivedZeppelinResource<I extends Iface> extends org.apache.thrift.ProcessFunction<I, onReceivedZeppelinResource_args> {
+ public onReceivedZeppelinResource() {
+ super("onReceivedZeppelinResource");
+ }
+
+ public onReceivedZeppelinResource_args getEmptyArgsInstance() {
+ return new onReceivedZeppelinResource_args();
+ }
+
+ protected boolean isOneway() {
+ return false;
+ }
+
+ public onReceivedZeppelinResource_result getResult(I iface, onReceivedZeppelinResource_args args) throws org.apache.thrift.TException {
+ onReceivedZeppelinResource_result result = new onReceivedZeppelinResource_result();
+ iface.onReceivedZeppelinResource(args.object);
+ return result;
+ }
+ }
+
}
public static class AsyncProcessor<I extends AsyncIface> extends org.apache.thrift.TBaseAsyncProcessor<I> {
@@ -2064,6 +2158,7 @@ public class RemoteInterpreterService {
processMap.put("loadApplication", new loadApplication());
processMap.put("unloadApplication", new unloadApplication());
processMap.put("runApplication", new runApplication());
+ processMap.put("onReceivedZeppelinResource", new onReceivedZeppelinResource());
return processMap;
}
@@ -3231,6 +3326,56 @@ public class RemoteInterpreterService {
}
}
+ public static class onReceivedZeppelinResource<I extends AsyncIface> extends org.apache.thrift.AsyncProcessFunction<I, onReceivedZeppelinResource_args, Void> {
+ public onReceivedZeppelinResource() {
+ super("onReceivedZeppelinResource");
+ }
+
+ public onReceivedZeppelinResource_args getEmptyArgsInstance() {
+ return new onReceivedZeppelinResource_args();
+ }
+
+ public AsyncMethodCallback<Void> getResultHandler(final AsyncFrameBuffer fb, final int seqid) {
+ final org.apache.thrift.AsyncProcessFunction fcall = this;
+ return new AsyncMethodCallback<Void>() {
+ public void onComplete(Void o) {
+ onReceivedZeppelinResource_result result = new onReceivedZeppelinResource_result();
+ try {
+ fcall.sendResponse(fb,result, org.apache.thrift.protocol.TMessageType.REPLY,seqid);
+ return;
+ } catch (Exception e) {
+ LOGGER.error("Exception writing to internal frame buffer", e);
+ }
+ fb.close();
+ }
+ public void onError(Exception e) {
+ byte msgType = org.apache.thrift.protocol.TMessageType.REPLY;
+ org.apache.thrift.TBase msg;
+ onReceivedZeppelinResource_result result = new onReceivedZeppelinResource_result();
+ {
+ msgType = org.apache.thrift.protocol.TMessageType.EXCEPTION;
+ msg = (org.apache.thrift.TBase)new org.apache.thrift.TApplicationException(org.apache.thrift.TApplicationException.INTERNAL_ERROR, e.getMessage());
+ }
+ try {
+ fcall.sendResponse(fb,msg,msgType,seqid);
+ return;
+ } catch (Exception ex) {
+ LOGGER.error("Exception writing to internal frame buffer", ex);
+ }
+ fb.close();
+ }
+ };
+ }
+
+ protected boolean isOneway() {
+ return false;
+ }
+
+ public void start(I iface, onReceivedZeppelinResource_args args, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws TException {
+ iface.onReceivedZeppelinResource(args.object,resultHandler);
+ }
+ }
+
}
public static class createInterpreter_args implements org.apache.thrift.TBase<createInterpreter_args, createInterpreter_args._Fields>, java.io.Serializable, Cloneable, Comparable<createInterpreter_args> {
@@ -21984,4 +22129,613 @@ public class RemoteInterpreterService {
}
+ public static class onReceivedZeppelinResource_args implements org.apache.thrift.TBase<onReceivedZeppelinResource_args, onReceivedZeppelinResource_args._Fields>, java.io.Serializable, Cloneable, Comparable<onReceivedZeppelinResource_args> {
+ private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("onReceivedZeppelinResource_args");
+
+ private static final org.apache.thrift.protocol.TField OBJECT_FIELD_DESC = new org.apache.thrift.protocol.TField("object", org.apache.thrift.protocol.TType.STRING, (short)1);
+
+ private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
+ static {
+ schemes.put(StandardScheme.class, new onReceivedZeppelinResource_argsStandardSchemeFactory());
+ schemes.put(TupleScheme.class, new onReceivedZeppelinResource_argsTupleSchemeFactory());
+ }
+
+ public String object; // required
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements org.apache.thrift.TFieldIdEnum {
+ OBJECT((short)1, "object");
+
+ private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+ static {
+ for (_Fields field : EnumSet.allOf(_Fields.class)) {
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ public static _Fields findByThriftId(int fieldId) {
+ switch(fieldId) {
+ case 1: // OBJECT
+ return OBJECT;
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, throwing an exception
+ * if it is not found.
+ */
+ public static _Fields findByThriftIdOrThrow(int fieldId) {
+ _Fields fields = findByThriftId(fieldId);
+ if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+ return fields;
+ }
+
+ /**
+ * Find the _Fields constant that matches name, or null if its not found.
+ */
+ public static _Fields findByName(String name) {
+ return byName.get(name);
+ }
+
+ private final short _thriftId;
+ private final String _fieldName;
+
+ _Fields(short thriftId, String fieldName) {
+ _thriftId = thriftId;
+ _fieldName = fieldName;
+ }
+
+ public short getThriftFieldId() {
+ return _thriftId;
+ }
+
+ public String getFieldName() {
+ return _fieldName;
+ }
+ }
+
+ // isset id assignments
+ public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+ static {
+ Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
+ tmpMap.put(_Fields.OBJECT, new org.apache.thrift.meta_data.FieldMetaData("object", org.apache.thrift.TFieldRequirementType.DEFAULT,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
+ metaDataMap = Collections.unmodifiableMap(tmpMap);
+ org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(onReceivedZeppelinResource_args.class, metaDataMap);
+ }
+
+ public onReceivedZeppelinResource_args() {
+ }
+
+ public onReceivedZeppelinResource_args(
+ String object)
+ {
+ this();
+ this.object = object;
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public onReceivedZeppelinResource_args(onReceivedZeppelinResource_args other) {
+ if (other.isSetObject()) {
+ this.object = other.object;
+ }
+ }
+
+ public onReceivedZeppelinResource_args deepCopy() {
+ return new onReceivedZeppelinResource_args(this);
+ }
+
+ @Override
+ public void clear() {
+ this.object = null;
+ }
+
+ public String getObject() {
+ return this.object;
+ }
+
+ public onReceivedZeppelinResource_args setObject(String object) {
+ this.object = object;
+ return this;
+ }
+
+ public void unsetObject() {
+ this.object = null;
+ }
+
+ /** Returns true if field object is set (has been assigned a value) and false otherwise */
+ public boolean isSetObject() {
+ return this.object != null;
+ }
+
+ public void setObjectIsSet(boolean value) {
+ if (!value) {
+ this.object = null;
+ }
+ }
+
+ public void setFieldValue(_Fields field, Object value) {
+ switch (field) {
+ case OBJECT:
+ if (value == null) {
+ unsetObject();
+ } else {
+ setObject((String)value);
+ }
+ break;
+
+ }
+ }
+
+ public Object getFieldValue(_Fields field) {
+ switch (field) {
+ case OBJECT:
+ return getObject();
+
+ }
+ throw new IllegalStateException();
+ }
+
+ /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+ public boolean isSet(_Fields field) {
+ if (field == null) {
+ throw new IllegalArgumentException();
+ }
+
+ switch (field) {
+ case OBJECT:
+ return isSetObject();
+ }
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ if (that == null)
+ return false;
+ if (that instanceof onReceivedZeppelinResource_args)
+ return this.equals((onReceivedZeppelinResource_args)that);
+ return false;
+ }
+
+ public boolean equals(onReceivedZeppelinResource_args that) {
+ if (that == null)
+ return false;
+
+ boolean this_present_object = true && this.isSetObject();
+ boolean that_present_object = true && that.isSetObject();
+ if (this_present_object || that_present_object) {
+ if (!(this_present_object && that_present_object))
+ return false;
+ if (!this.object.equals(that.object))
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ List<Object> list = new ArrayList<Object>();
+
+ boolean present_object = true && (isSetObject());
+ list.add(present_object);
+ if (present_object)
+ list.add(object);
+
+ return list.hashCode();
+ }
+
+ @Override
+ public int compareTo(onReceivedZeppelinResource_args other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+
+ lastComparison = Boolean.valueOf(isSetObject()).compareTo(other.isSetObject());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetObject()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.object, other.object);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ return 0;
+ }
+
+ public _Fields fieldForId(int fieldId) {
+ return _Fields.findByThriftId(fieldId);
+ }
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+ schemes.get(iprot.getScheme()).getScheme().read(iprot, this);
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+ schemes.get(oprot.getScheme()).getScheme().write(oprot, this);
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("onReceivedZeppelinResource_args(");
+ boolean first = true;
+
+ sb.append("object:");
+ if (this.object == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.object);
+ }
+ first = false;
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws org.apache.thrift.TException {
+ // check for required fields
+ // check for sub-struct validity
+ }
+
+ private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+ try {
+ write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
+ try {
+ read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private static class onReceivedZeppelinResource_argsStandardSchemeFactory implements SchemeFactory {
+ public onReceivedZeppelinResource_argsStandardScheme getScheme() {
+ return new onReceivedZeppelinResource_argsStandardScheme();
+ }
+ }
+
+ private static class onReceivedZeppelinResource_argsStandardScheme extends StandardScheme<onReceivedZeppelinResource_args> {
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot, onReceivedZeppelinResource_args struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TField schemeField;
+ iprot.readStructBegin();
+ while (true)
+ {
+ schemeField = iprot.readFieldBegin();
+ if (schemeField.type == org.apache.thrift.protocol.TType.STOP) {
+ break;
+ }
+ switch (schemeField.id) {
+ case 1: // OBJECT
+ if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
+ struct.object = iprot.readString();
+ struct.setObjectIsSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ break;
+ default:
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ iprot.readFieldEnd();
+ }
+ iprot.readStructEnd();
+
+ // check for required fields of primitive type, which can't be checked in the validate method
+ struct.validate();
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot, onReceivedZeppelinResource_args struct) throws org.apache.thrift.TException {
+ struct.validate();
+
+ oprot.writeStructBegin(STRUCT_DESC);
+ if (struct.object != null) {
+ oprot.writeFieldBegin(OBJECT_FIELD_DESC);
+ oprot.writeString(struct.object);
+ oprot.writeFieldEnd();
+ }
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ }
+
+ private static class onReceivedZeppelinResource_argsTupleSchemeFactory implements SchemeFactory {
+ public onReceivedZeppelinResource_argsTupleScheme getScheme() {
+ return new onReceivedZeppelinResource_argsTupleScheme();
+ }
+ }
+
+ private static class onReceivedZeppelinResource_argsTupleScheme extends TupleScheme<onReceivedZeppelinResource_args> {
+
+ @Override
+ public void write(org.apache.thrift.protocol.TProtocol prot, onReceivedZeppelinResource_args struct) throws org.apache.thrift.TException {
+ TTupleProtocol oprot = (TTupleProtocol) prot;
+ BitSet optionals = new BitSet();
+ if (struct.isSetObject()) {
+ optionals.set(0);
+ }
+ oprot.writeBitSet(optionals, 1);
+ if (struct.isSetObject()) {
+ oprot.writeString(struct.object);
+ }
+ }
+
+ @Override
+ public void read(org.apache.thrift.protocol.TProtocol prot, onReceivedZeppelinResource_args struct) throws org.apache.thrift.TException {
+ TTupleProtocol iprot = (TTupleProtocol) prot;
+ BitSet incoming = iprot.readBitSet(1);
+ if (incoming.get(0)) {
+ struct.object = iprot.readString();
+ struct.setObjectIsSet(true);
+ }
+ }
+ }
+
+ }
+
+ public static class onReceivedZeppelinResource_result implements org.apache.thrift.TBase<onReceivedZeppelinResource_result, onReceivedZeppelinResource_result._Fields>, java.io.Serializable, Cloneable, Comparable<onReceivedZeppelinResource_result> {
+ private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("onReceivedZeppelinResource_result");
+
+
+ private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
+ static {
+ schemes.put(StandardScheme.class, new onReceivedZeppelinResource_resultStandardSchemeFactory());
+ schemes.put(TupleScheme.class, new onReceivedZeppelinResource_resultTupleSchemeFactory());
+ }
+
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements org.apache.thrift.TFieldIdEnum {
+;
+
+ private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+ static {
+ for (_Fields field : EnumSet.allOf(_Fields.class)) {
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ public static _Fields findByThriftId(int fieldId) {
+ switch(fieldId) {
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, throwing an exception
+ * if it is not found.
+ */
+ public static _Fields findByThriftIdOrThrow(int fieldId) {
+ _Fields fields = findByThriftId(fieldId);
+ if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+ return fields;
+ }
+
+ /**
+ * Find the _Fields constant that matches name, or null if its not found.
+ */
+ public static _Fields findByName(String name) {
+ return byName.get(name);
+ }
+
+ private final short _thriftId;
+ private final String _fieldName;
+
+ _Fields(short thriftId, String fieldName) {
+ _thriftId = thriftId;
+ _fieldName = fieldName;
+ }
+
+ public short getThriftFieldId() {
+ return _thriftId;
+ }
+
+ public String getFieldName() {
+ return _fieldName;
+ }
+ }
+ public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+ static {
+ Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
+ metaDataMap = Collections.unmodifiableMap(tmpMap);
+ org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(onReceivedZeppelinResource_result.class, metaDataMap);
+ }
+
+ public onReceivedZeppelinResource_result() {
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public onReceivedZeppelinResource_result(onReceivedZeppelinResource_result other) {
+ }
+
+ public onReceivedZeppelinResource_result deepCopy() {
+ return new onReceivedZeppelinResource_result(this);
+ }
+
+ @Override
+ public void clear() {
+ }
+
+ public void setFieldValue(_Fields field, Object value) {
+ switch (field) {
+ }
+ }
+
+ public Object getFieldValue(_Fields field) {
+ switch (field) {
+ }
+ throw new IllegalStateException();
+ }
+
+ /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+ public boolean isSet(_Fields field) {
+ if (field == null) {
+ throw new IllegalArgumentException();
+ }
+
+ switch (field) {
+ }
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ if (that == null)
+ return false;
+ if (that instanceof onReceivedZeppelinResource_result)
+ return this.equals((onReceivedZeppelinResource_result)that);
+ return false;
+ }
+
+ public boolean equals(onReceivedZeppelinResource_result that) {
+ if (that == null)
+ return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ List<Object> list = new ArrayList<Object>();
+
+ return list.hashCode();
+ }
+
+ @Override
+ public int compareTo(onReceivedZeppelinResource_result other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+
+ return 0;
+ }
+
+ public _Fields fieldForId(int fieldId) {
+ return _Fields.findByThriftId(fieldId);
+ }
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+ schemes.get(iprot.getScheme()).getScheme().read(iprot, this);
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+ schemes.get(oprot.getScheme()).getScheme().write(oprot, this);
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("onReceivedZeppelinResource_result(");
+ boolean first = true;
+
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws org.apache.thrift.TException {
+ // check for required fields
+ // check for sub-struct validity
+ }
+
+ private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+ try {
+ write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
+ try {
+ read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private static class onReceivedZeppelinResource_resultStandardSchemeFactory implements SchemeFactory {
+ public onReceivedZeppelinResource_resultStandardScheme getScheme() {
+ return new onReceivedZeppelinResource_resultStandardScheme();
+ }
+ }
+
+ private static class onReceivedZeppelinResource_resultStandardScheme extends StandardScheme<onReceivedZeppelinResource_result> {
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot, onReceivedZeppelinResource_result struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TField schemeField;
+ iprot.readStructBegin();
+ while (true)
+ {
+ schemeField = iprot.readFieldBegin();
+ if (schemeField.type == org.apache.thrift.protocol.TType.STOP) {
+ break;
+ }
+ switch (schemeField.id) {
+ default:
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ iprot.readFieldEnd();
+ }
+ iprot.readStructEnd();
+
+ // check for required fields of primitive type, which can't be checked in the validate method
+ struct.validate();
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot, onReceivedZeppelinResource_result struct) throws org.apache.thrift.TException {
+ struct.validate();
+
+ oprot.writeStructBegin(STRUCT_DESC);
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ }
+
+ private static class onReceivedZeppelinResource_resultTupleSchemeFactory implements SchemeFactory {
+ public onReceivedZeppelinResource_resultTupleScheme getScheme() {
+ return new onReceivedZeppelinResource_resultTupleScheme();
+ }
+ }
+
+ private static class onReceivedZeppelinResource_resultTupleScheme extends TupleScheme<onReceivedZeppelinResource_result> {
+
+ @Override
+ public void write(org.apache.thrift.protocol.TProtocol prot, onReceivedZeppelinResource_result struct) throws org.apache.thrift.TException {
+ TTupleProtocol oprot = (TTupleProtocol) prot;
+ }
+
+ @Override
+ public void read(org.apache.thrift.protocol.TProtocol prot, onReceivedZeppelinResource_result struct) throws org.apache.thrift.TException {
+ TTupleProtocol iprot = (TTupleProtocol) prot;
+ }
+ }
+
+ }
+
}