diff --git a/helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinApplicationDevServer.java b/helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinApplicationDevServer.java index fe1f191a012..b349ec03815 100644 --- a/helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinApplicationDevServer.java +++ b/helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinApplicationDevServer.java @@ -136,7 +136,7 @@ ApplicationContext getApplicationContext(InterpreterContext interpreterContext) @Override protected InterpreterOutput createInterpreterOutput( - final String noteId, final String paragraphId) { + final String noteId, final String paragraphId, final String user) { if (out == null) { final RemoteInterpreterEventClient eventClient = getIntpEventClient(); try { @@ -148,14 +148,15 @@ public void onUpdateAll(InterpreterOutput out) { @Override public void onAppend(int index, InterpreterResultMessageOutput out, byte[] line) { - eventClient.onInterpreterOutputAppend(noteId, paragraphId, index, new String(line)); + eventClient.onInterpreterOutputAppend( + noteId, paragraphId, index, user, new String(line)); } @Override public void onUpdate(int index, InterpreterResultMessageOutput out) { try { eventClient.onInterpreterOutputUpdate(noteId, paragraphId, - index, out.getType(), new String(out.toByteArray())); + index, user, out.getType(), new String(out.toByteArray())); } catch (IOException e) { LOGGER.error(e.getMessage(), e); } diff --git a/helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinDevServer.java b/helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinDevServer.java index 81e0a610180..4d96612224f 100644 --- a/helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinDevServer.java +++ b/helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinDevServer.java @@ -66,7 +66,7 @@ protected Interpreter getInterpreter(String sessionId, String className) throws @Override protected InterpreterOutput createInterpreterOutput( - final String noteId, final String paragraphId) { + final String noteId, final String paragraphId, final String user) { if (out == null) { final RemoteInterpreterEventClient eventClient = getIntpEventClient(); try { @@ -78,14 +78,15 @@ public void onUpdateAll(InterpreterOutput out) { @Override public void onAppend(int index, InterpreterResultMessageOutput out, byte[] line) { - eventClient.onInterpreterOutputAppend(noteId, paragraphId, index, new String(line)); + eventClient.onInterpreterOutputAppend( + noteId, paragraphId, index, user, new String(line)); } @Override public void onUpdate(int index, InterpreterResultMessageOutput out) { try { eventClient.onInterpreterOutputUpdate(noteId, paragraphId, - index, out.getType(), new String(out.toByteArray())); + index, user, out.getType(), new String(out.toByteArray())); } catch (IOException e) { LOGGER.error(e.getMessage(), e); } 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 174de3bc194..358cc688027 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 @@ -222,11 +222,11 @@ public Resource invokeMethod( } public void onInterpreterOutputAppend( - String noteId, String paragraphId, int outputIndex, String output) { + String noteId, String paragraphId, int outputIndex, String user, String output) { try { callRemoteFunction(client -> { client.appendOutput( - new OutputAppendEvent(noteId, paragraphId, outputIndex, output, null)); + new OutputAppendEvent(noteId, paragraphId, outputIndex, output, null, user)); return null; }); } catch (Exception e) { @@ -235,12 +235,12 @@ public void onInterpreterOutputAppend( } public void onInterpreterOutputUpdate( - String noteId, String paragraphId, int outputIndex, + String noteId, String paragraphId, int outputIndex, String user, InterpreterResult.Type type, String output) { try { callRemoteFunction(client -> { - client.updateOutput( - new OutputUpdateEvent(noteId, paragraphId, outputIndex, type.name(), output, null)); + client.updateOutput(new OutputUpdateEvent( + noteId, paragraphId, outputIndex, type.name(), output, null, user)); return null; }); @@ -250,11 +250,11 @@ public void onInterpreterOutputUpdate( } public void onInterpreterOutputUpdateAll( - String noteId, String paragraphId, List messages) { + String noteId, String paragraphId, String user, List messages) { try { callRemoteFunction(client -> { client.updateAllOutput( - new OutputUpdateAllEvent(noteId, paragraphId, convertToThrift(messages))); + new OutputUpdateAllEvent(noteId, paragraphId, convertToThrift(messages), user)); return null; }); 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 e733fd57f8c..414d356fe1f 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 @@ -955,7 +955,12 @@ public List completion(String sessionId, } private InterpreterContext convert(RemoteInterpreterContext ric) { - return convert(ric, createInterpreterOutput(ric.getNoteId(), ric.getParagraphId())); + // The execution owner is fixed here, before any output is produced, so that every event + // emitted by this output carries the same owner regardless of what runs later. + AuthenticationInfo authenticationInfo = + AuthenticationInfo.fromJson(ric.getAuthenticationInfo()); + String user = authenticationInfo == null ? null : authenticationInfo.getUser(); + return convert(ric, createInterpreterOutput(ric.getNoteId(), ric.getParagraphId(), user)); } private InterpreterContext convert(RemoteInterpreterContext ric, InterpreterOutput output) { @@ -982,13 +987,13 @@ private InterpreterContext convert(RemoteInterpreterContext ric, InterpreterOutp protected InterpreterOutput createInterpreterOutput(final String noteId, final String - paragraphId) { + paragraphId, final String user) { return new InterpreterOutput(new InterpreterOutputListener() { @Override public void onUpdateAll(InterpreterOutput out) { try { intpEventClient.onInterpreterOutputUpdateAll( - noteId, paragraphId, out.toInterpreterResultMessage()); + noteId, paragraphId, user, out.toInterpreterResultMessage()); } catch (IOException e) { LOGGER.error(e.getMessage(), e); } @@ -999,7 +1004,7 @@ public void onAppend(int index, InterpreterResultMessageOutput out, byte[] line) String output = new String(line); LOGGER.debug("Output Append: {}", output); intpEventClient.onInterpreterOutputAppend( - noteId, paragraphId, index, output); + noteId, paragraphId, index, user, output); } @Override @@ -1009,7 +1014,7 @@ public void onUpdate(int index, InterpreterResultMessageOutput out) { output = new String(out.toByteArray()); LOGGER.debug("Output Update for index {}: {}", index, output); intpEventClient.onInterpreterOutputUpdate( - noteId, paragraphId, index, out.getType(), output); + noteId, paragraphId, index, user, out.getType(), output); } catch (IOException e) { LOGGER.error(e.getMessage(), e); } diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AngularObjectId.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AngularObjectId.java index 8f93f2ed444..7a62542b34e 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AngularObjectId.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AngularObjectId.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class AngularObjectId implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AngularObjectId"); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputAppendEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputAppendEvent.java index 676e46919cd..5e19d8d3154 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputAppendEvent.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputAppendEvent.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class AppOutputAppendEvent implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AppOutputAppendEvent"); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputUpdateEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputUpdateEvent.java index 46b0ac6eec0..6890bffa66f 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputUpdateEvent.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputUpdateEvent.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class AppOutputUpdateEvent implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AppOutputUpdateEvent"); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppStatusUpdateEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppStatusUpdateEvent.java index 57a18c645cb..af9c8292725 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppStatusUpdateEvent.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppStatusUpdateEvent.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class AppStatusUpdateEvent implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AppStatusUpdateEvent"); 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 7a92c079fa6..af96cedca96 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 @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class InterpreterCompletion implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("InterpreterCompletion"); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterRPCException.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterRPCException.java index f86ac4de5e6..90f1d943a41 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterRPCException.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterRPCException.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class InterpreterRPCException extends org.apache.thrift.TException implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("InterpreterRPCException"); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/LibraryMetadata.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/LibraryMetadata.java index 4299eb67117..381b8899750 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/LibraryMetadata.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/LibraryMetadata.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class LibraryMetadata implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("LibraryMetadata"); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputAppendEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputAppendEvent.java index 67a3a8f0541..a284d9fbfbf 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputAppendEvent.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputAppendEvent.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class OutputAppendEvent implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("OutputAppendEvent"); @@ -33,6 +33,7 @@ public class OutputAppendEvent implements org.apache.thrift.TBase byName = new java.util.HashMap(); @@ -75,6 +78,8 @@ public static _Fields findByThriftId(int fieldId) { return DATA; case 5: // APP_ID return APP_ID; + case 6: // USER + return USER; default: return null; } @@ -131,6 +136,8 @@ public java.lang.String getFieldName() { new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING))); tmpMap.put(_Fields.APP_ID, new org.apache.thrift.meta_data.FieldMetaData("appId", org.apache.thrift.TFieldRequirementType.DEFAULT, new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING))); + tmpMap.put(_Fields.USER, new org.apache.thrift.meta_data.FieldMetaData("user", org.apache.thrift.TFieldRequirementType.DEFAULT, + new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING))); metaDataMap = java.util.Collections.unmodifiableMap(tmpMap); org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(OutputAppendEvent.class, metaDataMap); } @@ -143,7 +150,8 @@ public OutputAppendEvent( java.lang.String paragraphId, int index, java.lang.String data, - java.lang.String appId) + java.lang.String appId, + java.lang.String user) { this(); this.noteId = noteId; @@ -152,6 +160,7 @@ public OutputAppendEvent( setIndexIsSet(true); this.data = data; this.appId = appId; + this.user = user; } /** @@ -172,6 +181,9 @@ public OutputAppendEvent(OutputAppendEvent other) { if (other.isSetAppId()) { this.appId = other.appId; } + if (other.isSetUser()) { + this.user = other.user; + } } public OutputAppendEvent deepCopy() { @@ -186,6 +198,7 @@ public void clear() { this.index = 0; this.data = null; this.appId = null; + this.user = null; } @org.apache.thrift.annotation.Nullable @@ -311,6 +324,31 @@ public void setAppIdIsSet(boolean value) { } } + @org.apache.thrift.annotation.Nullable + public java.lang.String getUser() { + return this.user; + } + + public OutputAppendEvent setUser(@org.apache.thrift.annotation.Nullable java.lang.String user) { + this.user = user; + return this; + } + + public void unsetUser() { + this.user = null; + } + + /** Returns true if field user is set (has been assigned a value) and false otherwise */ + public boolean isSetUser() { + return this.user != null; + } + + public void setUserIsSet(boolean value) { + if (!value) { + this.user = null; + } + } + public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) { switch (field) { case NOTE_ID: @@ -353,6 +391,14 @@ public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable } break; + case USER: + if (value == null) { + unsetUser(); + } else { + setUser((java.lang.String)value); + } + break; + } } @@ -374,6 +420,9 @@ public java.lang.Object getFieldValue(_Fields field) { case APP_ID: return getAppId(); + case USER: + return getUser(); + } throw new java.lang.IllegalStateException(); } @@ -395,6 +444,8 @@ public boolean isSet(_Fields field) { return isSetData(); case APP_ID: return isSetAppId(); + case USER: + return isSetUser(); } throw new java.lang.IllegalStateException(); } @@ -459,6 +510,15 @@ public boolean equals(OutputAppendEvent that) { return false; } + boolean this_present_user = true && this.isSetUser(); + boolean that_present_user = true && that.isSetUser(); + if (this_present_user || that_present_user) { + if (!(this_present_user && that_present_user)) + return false; + if (!this.user.equals(that.user)) + return false; + } + return true; } @@ -484,6 +544,10 @@ public int hashCode() { if (isSetAppId()) hashCode = hashCode * 8191 + appId.hashCode(); + hashCode = hashCode * 8191 + ((isSetUser()) ? 131071 : 524287); + if (isSetUser()) + hashCode = hashCode * 8191 + user.hashCode(); + return hashCode; } @@ -545,6 +609,16 @@ public int compareTo(OutputAppendEvent other) { return lastComparison; } } + lastComparison = java.lang.Boolean.valueOf(isSetUser()).compareTo(other.isSetUser()); + if (lastComparison != 0) { + return lastComparison; + } + if (isSetUser()) { + lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.user, other.user); + if (lastComparison != 0) { + return lastComparison; + } + } return 0; } @@ -601,6 +675,14 @@ public java.lang.String toString() { sb.append(this.appId); } first = false; + if (!first) sb.append(", "); + sb.append("user:"); + if (this.user == null) { + sb.append("null"); + } else { + sb.append(this.user); + } + first = false; sb.append(")"); return sb.toString(); } @@ -686,6 +768,14 @@ public void read(org.apache.thrift.protocol.TProtocol iprot, OutputAppendEvent s org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); } break; + case 6: // USER + if (schemeField.type == org.apache.thrift.protocol.TType.STRING) { + struct.user = iprot.readString(); + struct.setUserIsSet(true); + } else { + org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); + } + break; default: org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); } @@ -724,6 +814,11 @@ public void write(org.apache.thrift.protocol.TProtocol oprot, OutputAppendEvent oprot.writeString(struct.appId); oprot.writeFieldEnd(); } + if (struct.user != null) { + oprot.writeFieldBegin(USER_FIELD_DESC); + oprot.writeString(struct.user); + oprot.writeFieldEnd(); + } oprot.writeFieldStop(); oprot.writeStructEnd(); } @@ -757,7 +852,10 @@ public void write(org.apache.thrift.protocol.TProtocol prot, OutputAppendEvent s if (struct.isSetAppId()) { optionals.set(4); } - oprot.writeBitSet(optionals, 5); + if (struct.isSetUser()) { + optionals.set(5); + } + oprot.writeBitSet(optionals, 6); if (struct.isSetNoteId()) { oprot.writeString(struct.noteId); } @@ -773,12 +871,15 @@ public void write(org.apache.thrift.protocol.TProtocol prot, OutputAppendEvent s if (struct.isSetAppId()) { oprot.writeString(struct.appId); } + if (struct.isSetUser()) { + oprot.writeString(struct.user); + } } @Override public void read(org.apache.thrift.protocol.TProtocol prot, OutputAppendEvent struct) throws org.apache.thrift.TException { org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot; - java.util.BitSet incoming = iprot.readBitSet(5); + java.util.BitSet incoming = iprot.readBitSet(6); if (incoming.get(0)) { struct.noteId = iprot.readString(); struct.setNoteIdIsSet(true); @@ -799,6 +900,10 @@ public void read(org.apache.thrift.protocol.TProtocol prot, OutputAppendEvent st struct.appId = iprot.readString(); struct.setAppIdIsSet(true); } + if (incoming.get(5)) { + struct.user = iprot.readString(); + struct.setUserIsSet(true); + } } } diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateAllEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateAllEvent.java index 587359fffd1..46ea41c2e29 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateAllEvent.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateAllEvent.java @@ -24,13 +24,14 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class OutputUpdateAllEvent implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("OutputUpdateAllEvent"); 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 org.apache.thrift.protocol.TField MSG_FIELD_DESC = new org.apache.thrift.protocol.TField("msg", org.apache.thrift.protocol.TType.LIST, (short)3); + private static final org.apache.thrift.protocol.TField USER_FIELD_DESC = new org.apache.thrift.protocol.TField("user", org.apache.thrift.protocol.TType.STRING, (short)4); private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new OutputUpdateAllEventStandardSchemeFactory(); private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new OutputUpdateAllEventTupleSchemeFactory(); @@ -38,12 +39,14 @@ public class OutputUpdateAllEvent implements org.apache.thrift.TBase msg; // required + public @org.apache.thrift.annotation.Nullable java.lang.String user; // 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"), - MSG((short)3, "msg"); + MSG((short)3, "msg"), + USER((short)4, "user"); private static final java.util.Map byName = new java.util.HashMap(); @@ -65,6 +68,8 @@ public static _Fields findByThriftId(int fieldId) { return PARAGRAPH_ID; case 3: // MSG return MSG; + case 4: // USER + return USER; default: return null; } @@ -116,6 +121,8 @@ public java.lang.String getFieldName() { tmpMap.put(_Fields.MSG, new org.apache.thrift.meta_data.FieldMetaData("msg", org.apache.thrift.TFieldRequirementType.DEFAULT, new org.apache.thrift.meta_data.ListMetaData(org.apache.thrift.protocol.TType.LIST, new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, org.apache.zeppelin.interpreter.thrift.RemoteInterpreterResultMessage.class)))); + tmpMap.put(_Fields.USER, new org.apache.thrift.meta_data.FieldMetaData("user", org.apache.thrift.TFieldRequirementType.DEFAULT, + new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING))); metaDataMap = java.util.Collections.unmodifiableMap(tmpMap); org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(OutputUpdateAllEvent.class, metaDataMap); } @@ -126,12 +133,14 @@ public OutputUpdateAllEvent() { public OutputUpdateAllEvent( java.lang.String noteId, java.lang.String paragraphId, - java.util.List msg) + java.util.List msg, + java.lang.String user) { this(); this.noteId = noteId; this.paragraphId = paragraphId; this.msg = msg; + this.user = user; } /** @@ -151,6 +160,9 @@ public OutputUpdateAllEvent(OutputUpdateAllEvent other) { } this.msg = __this__msg; } + if (other.isSetUser()) { + this.user = other.user; + } } public OutputUpdateAllEvent deepCopy() { @@ -162,6 +174,7 @@ public void clear() { this.noteId = null; this.paragraphId = null; this.msg = null; + this.user = null; } @org.apache.thrift.annotation.Nullable @@ -255,6 +268,31 @@ public void setMsgIsSet(boolean value) { } } + @org.apache.thrift.annotation.Nullable + public java.lang.String getUser() { + return this.user; + } + + public OutputUpdateAllEvent setUser(@org.apache.thrift.annotation.Nullable java.lang.String user) { + this.user = user; + return this; + } + + public void unsetUser() { + this.user = null; + } + + /** Returns true if field user is set (has been assigned a value) and false otherwise */ + public boolean isSetUser() { + return this.user != null; + } + + public void setUserIsSet(boolean value) { + if (!value) { + this.user = null; + } + } + public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) { switch (field) { case NOTE_ID: @@ -281,6 +319,14 @@ public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable } break; + case USER: + if (value == null) { + unsetUser(); + } else { + setUser((java.lang.String)value); + } + break; + } } @@ -296,6 +342,9 @@ public java.lang.Object getFieldValue(_Fields field) { case MSG: return getMsg(); + case USER: + return getUser(); + } throw new java.lang.IllegalStateException(); } @@ -313,6 +362,8 @@ public boolean isSet(_Fields field) { return isSetParagraphId(); case MSG: return isSetMsg(); + case USER: + return isSetUser(); } throw new java.lang.IllegalStateException(); } @@ -359,6 +410,15 @@ public boolean equals(OutputUpdateAllEvent that) { return false; } + boolean this_present_user = true && this.isSetUser(); + boolean that_present_user = true && that.isSetUser(); + if (this_present_user || that_present_user) { + if (!(this_present_user && that_present_user)) + return false; + if (!this.user.equals(that.user)) + return false; + } + return true; } @@ -378,6 +438,10 @@ public int hashCode() { if (isSetMsg()) hashCode = hashCode * 8191 + msg.hashCode(); + hashCode = hashCode * 8191 + ((isSetUser()) ? 131071 : 524287); + if (isSetUser()) + hashCode = hashCode * 8191 + user.hashCode(); + return hashCode; } @@ -419,6 +483,16 @@ public int compareTo(OutputUpdateAllEvent other) { return lastComparison; } } + lastComparison = java.lang.Boolean.valueOf(isSetUser()).compareTo(other.isSetUser()); + if (lastComparison != 0) { + return lastComparison; + } + if (isSetUser()) { + lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.user, other.user); + if (lastComparison != 0) { + return lastComparison; + } + } return 0; } @@ -463,6 +537,14 @@ public java.lang.String toString() { sb.append(this.msg); } first = false; + if (!first) sb.append(", "); + sb.append("user:"); + if (this.user == null) { + sb.append("null"); + } else { + sb.append(this.user); + } + first = false; sb.append(")"); return sb.toString(); } @@ -541,6 +623,14 @@ public void read(org.apache.thrift.protocol.TProtocol iprot, OutputUpdateAllEven org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); } break; + case 4: // USER + if (schemeField.type == org.apache.thrift.protocol.TType.STRING) { + struct.user = iprot.readString(); + struct.setUserIsSet(true); + } else { + org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); + } + break; default: org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); } @@ -578,6 +668,11 @@ public void write(org.apache.thrift.protocol.TProtocol oprot, OutputUpdateAllEve } oprot.writeFieldEnd(); } + if (struct.user != null) { + oprot.writeFieldBegin(USER_FIELD_DESC); + oprot.writeString(struct.user); + oprot.writeFieldEnd(); + } oprot.writeFieldStop(); oprot.writeStructEnd(); } @@ -605,7 +700,10 @@ public void write(org.apache.thrift.protocol.TProtocol prot, OutputUpdateAllEven if (struct.isSetMsg()) { optionals.set(2); } - oprot.writeBitSet(optionals, 3); + if (struct.isSetUser()) { + optionals.set(3); + } + oprot.writeBitSet(optionals, 4); if (struct.isSetNoteId()) { oprot.writeString(struct.noteId); } @@ -621,12 +719,15 @@ public void write(org.apache.thrift.protocol.TProtocol prot, OutputUpdateAllEven } } } + if (struct.isSetUser()) { + oprot.writeString(struct.user); + } } @Override public void read(org.apache.thrift.protocol.TProtocol prot, OutputUpdateAllEvent struct) throws org.apache.thrift.TException { org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot; - java.util.BitSet incoming = iprot.readBitSet(3); + java.util.BitSet incoming = iprot.readBitSet(4); if (incoming.get(0)) { struct.noteId = iprot.readString(); struct.setNoteIdIsSet(true); @@ -649,6 +750,10 @@ public void read(org.apache.thrift.protocol.TProtocol prot, OutputUpdateAllEvent } struct.setMsgIsSet(true); } + if (incoming.get(3)) { + struct.user = iprot.readString(); + struct.setUserIsSet(true); + } } } diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateEvent.java index ef5d2f0d3b5..634c9f9beb1 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateEvent.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateEvent.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class OutputUpdateEvent implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("OutputUpdateEvent"); @@ -34,6 +34,7 @@ public class OutputUpdateEvent implements org.apache.thrift.TBase byName = new java.util.HashMap(); @@ -80,6 +83,8 @@ public static _Fields findByThriftId(int fieldId) { return DATA; case 6: // APP_ID return APP_ID; + case 7: // USER + return USER; default: return null; } @@ -138,6 +143,8 @@ public java.lang.String getFieldName() { new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING))); tmpMap.put(_Fields.APP_ID, new org.apache.thrift.meta_data.FieldMetaData("appId", org.apache.thrift.TFieldRequirementType.DEFAULT, new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING))); + tmpMap.put(_Fields.USER, new org.apache.thrift.meta_data.FieldMetaData("user", org.apache.thrift.TFieldRequirementType.DEFAULT, + new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING))); metaDataMap = java.util.Collections.unmodifiableMap(tmpMap); org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(OutputUpdateEvent.class, metaDataMap); } @@ -151,7 +158,8 @@ public OutputUpdateEvent( int index, java.lang.String type, java.lang.String data, - java.lang.String appId) + java.lang.String appId, + java.lang.String user) { this(); this.noteId = noteId; @@ -161,6 +169,7 @@ public OutputUpdateEvent( this.type = type; this.data = data; this.appId = appId; + this.user = user; } /** @@ -184,6 +193,9 @@ public OutputUpdateEvent(OutputUpdateEvent other) { if (other.isSetAppId()) { this.appId = other.appId; } + if (other.isSetUser()) { + this.user = other.user; + } } public OutputUpdateEvent deepCopy() { @@ -199,6 +211,7 @@ public void clear() { this.type = null; this.data = null; this.appId = null; + this.user = null; } @org.apache.thrift.annotation.Nullable @@ -349,6 +362,31 @@ public void setAppIdIsSet(boolean value) { } } + @org.apache.thrift.annotation.Nullable + public java.lang.String getUser() { + return this.user; + } + + public OutputUpdateEvent setUser(@org.apache.thrift.annotation.Nullable java.lang.String user) { + this.user = user; + return this; + } + + public void unsetUser() { + this.user = null; + } + + /** Returns true if field user is set (has been assigned a value) and false otherwise */ + public boolean isSetUser() { + return this.user != null; + } + + public void setUserIsSet(boolean value) { + if (!value) { + this.user = null; + } + } + public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) { switch (field) { case NOTE_ID: @@ -399,6 +437,14 @@ public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable } break; + case USER: + if (value == null) { + unsetUser(); + } else { + setUser((java.lang.String)value); + } + break; + } } @@ -423,6 +469,9 @@ public java.lang.Object getFieldValue(_Fields field) { case APP_ID: return getAppId(); + case USER: + return getUser(); + } throw new java.lang.IllegalStateException(); } @@ -446,6 +495,8 @@ public boolean isSet(_Fields field) { return isSetData(); case APP_ID: return isSetAppId(); + case USER: + return isSetUser(); } throw new java.lang.IllegalStateException(); } @@ -519,6 +570,15 @@ public boolean equals(OutputUpdateEvent that) { return false; } + boolean this_present_user = true && this.isSetUser(); + boolean that_present_user = true && that.isSetUser(); + if (this_present_user || that_present_user) { + if (!(this_present_user && that_present_user)) + return false; + if (!this.user.equals(that.user)) + return false; + } + return true; } @@ -548,6 +608,10 @@ public int hashCode() { if (isSetAppId()) hashCode = hashCode * 8191 + appId.hashCode(); + hashCode = hashCode * 8191 + ((isSetUser()) ? 131071 : 524287); + if (isSetUser()) + hashCode = hashCode * 8191 + user.hashCode(); + return hashCode; } @@ -619,6 +683,16 @@ public int compareTo(OutputUpdateEvent other) { return lastComparison; } } + lastComparison = java.lang.Boolean.valueOf(isSetUser()).compareTo(other.isSetUser()); + if (lastComparison != 0) { + return lastComparison; + } + if (isSetUser()) { + lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.user, other.user); + if (lastComparison != 0) { + return lastComparison; + } + } return 0; } @@ -683,6 +757,14 @@ public java.lang.String toString() { sb.append(this.appId); } first = false; + if (!first) sb.append(", "); + sb.append("user:"); + if (this.user == null) { + sb.append("null"); + } else { + sb.append(this.user); + } + first = false; sb.append(")"); return sb.toString(); } @@ -776,6 +858,14 @@ public void read(org.apache.thrift.protocol.TProtocol iprot, OutputUpdateEvent s org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); } break; + case 7: // USER + if (schemeField.type == org.apache.thrift.protocol.TType.STRING) { + struct.user = iprot.readString(); + struct.setUserIsSet(true); + } else { + org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); + } + break; default: org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); } @@ -819,6 +909,11 @@ public void write(org.apache.thrift.protocol.TProtocol oprot, OutputUpdateEvent oprot.writeString(struct.appId); oprot.writeFieldEnd(); } + if (struct.user != null) { + oprot.writeFieldBegin(USER_FIELD_DESC); + oprot.writeString(struct.user); + oprot.writeFieldEnd(); + } oprot.writeFieldStop(); oprot.writeStructEnd(); } @@ -855,7 +950,10 @@ public void write(org.apache.thrift.protocol.TProtocol prot, OutputUpdateEvent s if (struct.isSetAppId()) { optionals.set(5); } - oprot.writeBitSet(optionals, 6); + if (struct.isSetUser()) { + optionals.set(6); + } + oprot.writeBitSet(optionals, 7); if (struct.isSetNoteId()) { oprot.writeString(struct.noteId); } @@ -874,12 +972,15 @@ public void write(org.apache.thrift.protocol.TProtocol prot, OutputUpdateEvent s if (struct.isSetAppId()) { oprot.writeString(struct.appId); } + if (struct.isSetUser()) { + oprot.writeString(struct.user); + } } @Override public void read(org.apache.thrift.protocol.TProtocol prot, OutputUpdateEvent struct) throws org.apache.thrift.TException { org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot; - java.util.BitSet incoming = iprot.readBitSet(6); + java.util.BitSet incoming = iprot.readBitSet(7); if (incoming.get(0)) { struct.noteId = iprot.readString(); struct.setNoteIdIsSet(true); @@ -904,6 +1005,10 @@ public void read(org.apache.thrift.protocol.TProtocol prot, OutputUpdateEvent st struct.appId = iprot.readString(); struct.setAppIdIsSet(true); } + if (incoming.get(6)) { + struct.user = iprot.readString(); + struct.setUserIsSet(true); + } } } diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ParagraphInfo.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ParagraphInfo.java index 083d8003518..0d08ce08167 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ParagraphInfo.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ParagraphInfo.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class ParagraphInfo implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("ParagraphInfo"); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RegisterInfo.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RegisterInfo.java index 2c3667eb8fb..18f7fcf7556 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RegisterInfo.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RegisterInfo.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class RegisterInfo implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RegisterInfo"); 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 8aa5bc2e6ba..d3d4114eeed 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 @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class RemoteApplicationResult implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteApplicationResult"); 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 a6243c83b94..bd7ee2d868d 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 @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class RemoteInterpreterContext implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterContext"); 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 eb22200e704..b4499f3c918 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 @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class RemoteInterpreterEvent implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterEvent"); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventService.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventService.java index 9a91dbd0590..0ccd4f54227 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventService.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventService.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class RemoteInterpreterEventService { public interface Iface { 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 292c5192532..5d8244ae899 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 @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public enum RemoteInterpreterEventType implements org.apache.thrift.TEnum { NO_OP(1), ANGULAR_OBJECT_ADD(2), 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 4fce5b0b9f6..00ceeb62e5f 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 @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class RemoteInterpreterResult implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterResult"); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResultMessage.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResultMessage.java index 6ea6fa6aa7f..3a00fe14f67 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResultMessage.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResultMessage.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class RemoteInterpreterResultMessage implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterResultMessage"); 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 4cde441cdce..a6ad9d5f976 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 @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class RemoteInterpreterService { public interface Iface { diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RunParagraphsEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RunParagraphsEvent.java index 47edc59883f..d9272dd4cc4 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RunParagraphsEvent.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RunParagraphsEvent.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class RunParagraphsEvent implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RunParagraphsEvent"); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ServiceException.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ServiceException.java index 1882f45f87c..469b9d8ffea 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ServiceException.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ServiceException.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class ServiceException extends org.apache.thrift.TException implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("ServiceException"); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/WebUrlInfo.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/WebUrlInfo.java index a622c0d6f30..3dd97226515 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/WebUrlInfo.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/WebUrlInfo.java @@ -24,7 +24,7 @@ package org.apache.zeppelin.interpreter.thrift; @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) -@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2021-03-09") +@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2026-09-13") public class WebUrlInfo implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("WebUrlInfo"); diff --git a/zeppelin-interpreter/src/main/thrift/RemoteInterpreterEventService.thrift b/zeppelin-interpreter/src/main/thrift/RemoteInterpreterEventService.thrift index ad08821f41c..69eae033df4 100644 --- a/zeppelin-interpreter/src/main/thrift/RemoteInterpreterEventService.thrift +++ b/zeppelin-interpreter/src/main/thrift/RemoteInterpreterEventService.thrift @@ -36,7 +36,10 @@ struct OutputAppendEvent { 2: string paragraphId, 3: i32 index, 4: string data, - 5: string appId + 5: string appId, + // Execution owner. Null when the interpreter predates this field; see + // NotebookServer.onOutputAppend for how ownerless output is handled. + 6: string user } struct OutputUpdateEvent { @@ -45,13 +48,15 @@ struct OutputUpdateEvent { 3: i32 index, 4: string type, 5: string data, - 6: string appId + 6: string appId, + 7: string user } struct OutputUpdateAllEvent { 1: string noteId, 2: string paragraphId, 3: list msg, + 4: string user } struct RunParagraphsEvent { diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java index c8f8886e6a9..11fc186f7ae 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java @@ -218,8 +218,8 @@ public void sendWebUrl(WebUrlInfo weburlInfo) throws InterpreterRPCException, TE @Override public void appendOutput(OutputAppendEvent event) throws InterpreterRPCException, TException { if (event.getAppId() == null) { - runner.appendBuffer( - event.getNoteId(), event.getParagraphId(), event.getIndex(), event.getData()); + runner.appendBuffer(event.getNoteId(), event.getParagraphId(), event.getIndex(), + event.getUser(), event.getData()); } else { appListener.onOutputAppend(event.getNoteId(), event.getParagraphId(), event.getIndex(), event.getAppId(), event.getData()); @@ -230,7 +230,7 @@ public void appendOutput(OutputAppendEvent event) throws InterpreterRPCException public void updateOutput(OutputUpdateEvent event) throws InterpreterRPCException, TException { if (event.getAppId() == null) { runner.updateBuffer(event.getNoteId(), event.getParagraphId(), event.getIndex(), - InterpreterResult.Type.valueOf(event.getType()), event.getData()); + event.getUser(), InterpreterResult.Type.valueOf(event.getType()), event.getData()); // Complete replacements before the interpreter can publish its terminal result. runner.run(); } else { @@ -244,11 +244,11 @@ public void updateAllOutput(OutputUpdateAllEvent event) throws InterpreterRPCExc synchronized (runner) { // Finish earlier output before the clear; keep replacements ahead of the next drain. runner.run(); - listener.onOutputClear(event.getNoteId(), event.getParagraphId()); + listener.onParagraphOutputClear(event.getNoteId(), event.getParagraphId(), event.getUser()); for (int i = 0; i < event.getMsg().size(); i++) { RemoteInterpreterResultMessage msg = event.getMsg().get(i); - listener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), i, - InterpreterResult.Type.valueOf(msg.getType()), msg.getData()); + listener.onParagraphOutputUpdated(event.getNoteId(), event.getParagraphId(), i, + event.getUser(), InterpreterResult.Type.valueOf(msg.getType()), msg.getData()); } } } diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputBuffer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputBuffer.java index b13940496f4..46f275bde1e 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputBuffer.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputBuffer.java @@ -26,12 +26,15 @@ public class AppendOutputBuffer { private String noteId; private String paragraphId; private int index; + private String user; private String data; - public AppendOutputBuffer(String noteId, String paragraphId, int index, String data) { + public AppendOutputBuffer(String noteId, String paragraphId, int index, String user, + String data) { this.noteId = noteId; this.paragraphId = paragraphId; this.index = index; + this.user = user; this.data = data; } @@ -47,6 +50,11 @@ public int getIndex() { return index; } + /** The execution owner, or null when the interpreter does not report one. */ + public String getUser() { + return user; + } + public String getData() { return data; } diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java index 8bc063af106..6ca6505cd6c 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java @@ -21,11 +21,12 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.Objects; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; @@ -52,7 +53,7 @@ public AppendOutputRunner(RemoteInterpreterProcessListener listener) { @Override public synchronized void run() { - Map stringBufferMap = new HashMap<>(); + Map stringBufferMap = new LinkedHashMap<>(); List list = new LinkedList<>(); queue.drainTo(list); @@ -67,8 +68,8 @@ public synchronized void run() { sizeProcessed += flushAppendBuffers(stringBufferMap); UpdateOutputBuffer update = (UpdateOutputBuffer) buffer; try { - listener.onOutputUpdated(update.getNoteId(), update.getParagraphId(), update.getIndex(), - update.getType(), update.getData()); + listener.onParagraphOutputUpdated(update.getNoteId(), update.getParagraphId(), + update.getIndex(), update.getUser(), update.getType(), update.getData()); } catch (RuntimeException e) { // A stale callback must not abort another paragraph's synchronous drain. LOGGER.warn("Failed to update output for note {} paragraph {}", @@ -77,16 +78,13 @@ public synchronized void run() { continue; } - String noteId = buffer.getNoteId(); - String paragraphId = buffer.getParagraphId(); - int index = buffer.getIndex(); - String stringBufferKey = noteId + ":" + paragraphId + ":" + index; + // The execution owner is part of the key: two users running the same paragraph must not + // have their output folded into one chunk, because the merged chunk would have no owner. + AppendKey key = new AppendKey(buffer.getNoteId(), buffer.getParagraphId(), + buffer.getIndex(), buffer.getUser()); - StringBuilder builder = stringBufferMap.containsKey(stringBufferKey) ? - stringBufferMap.get(stringBufferKey) : new StringBuilder(); - - builder.append(buffer.getData()); - stringBufferMap.put(stringBufferKey, builder); + stringBufferMap.computeIfAbsent(key, unused -> new StringBuilder()) + .append(buffer.getData()); } sizeProcessed += flushAppendBuffers(stringBufferMap); Long processingTime = System.currentTimeMillis() - processingStartTime; @@ -104,31 +102,75 @@ public synchronized void run() { } } - private long flushAppendBuffers(Map stringBufferMap) { + private long flushAppendBuffers(Map stringBufferMap) { long sizeProcessed = 0; - for (Entry stringBufferMapEntry : stringBufferMap.entrySet()) { - String stringBufferKey = stringBufferMapEntry.getKey(); + for (Entry stringBufferMapEntry : stringBufferMap.entrySet()) { + AppendKey key = stringBufferMapEntry.getKey(); StringBuilder buffer = stringBufferMapEntry.getValue(); sizeProcessed += buffer.length(); try { - String[] keys = stringBufferKey.split(":"); - listener.onOutputAppend(keys[0], keys[1], Integer.parseInt(keys[2]), buffer.toString()); + listener.onParagraphOutputAppend(key.noteId, key.paragraphId, key.index, key.user, + buffer.toString()); } catch (RuntimeException e) { // One stale append must not abort another paragraph's synchronous drain. - LOGGER.warn("Failed to append output for {}", stringBufferKey, e); + LOGGER.warn("Failed to append output for {}", key, e); } } stringBufferMap.clear(); return sizeProcessed; } - public void appendBuffer(String noteId, String paragraphId, int index, String outputToAppend) { - queue.offer(new AppendOutputBuffer(noteId, paragraphId, index, outputToAppend)); + /** + * Identifies one stream of appended output. A user name can contain any character, so the + * parts are kept separate instead of being joined into a delimited string. + */ + private static final class AppendKey { + private final String noteId; + private final String paragraphId; + private final int index; + private final String user; + + private AppendKey(String noteId, String paragraphId, int index, String user) { + this.noteId = noteId; + this.paragraphId = paragraphId; + this.index = index; + this.user = user; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof AppendKey)) { + return false; + } + AppendKey other = (AppendKey) o; + return index == other.index + && Objects.equals(noteId, other.noteId) + && Objects.equals(paragraphId, other.paragraphId) + && Objects.equals(user, other.user); + } + + @Override + public int hashCode() { + return Objects.hash(noteId, paragraphId, index, user); + } + + @Override + public String toString() { + return "note " + noteId + " paragraph " + paragraphId + " index " + index + " user " + user; + } + } + + public void appendBuffer(String noteId, String paragraphId, int index, String user, + String outputToAppend) { + queue.offer(new AppendOutputBuffer(noteId, paragraphId, index, user, outputToAppend)); } /** Enqueues a replacement; callers needing completion must also invoke run(). */ - public void updateBuffer(String noteId, String paragraphId, int index, + public void updateBuffer(String noteId, String paragraphId, int index, String user, InterpreterResult.Type type, String output) { - queue.offer(new UpdateOutputBuffer(noteId, paragraphId, index, type, output)); + queue.offer(new UpdateOutputBuffer(noteId, paragraphId, index, user, type, output)); } } diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcessListener.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcessListener.java index e6876c72ea2..580a4642061 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcessListener.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcessListener.java @@ -33,27 +33,31 @@ public interface RemoteInterpreterProcessListener { * @param noteId * @param paragraphId * @param index + * @param user the execution owner, or null when the interpreter does not report one * @param output */ - void onOutputAppend(String noteId, String paragraphId, int index, String output); + void onParagraphOutputAppend(String noteId, String paragraphId, int index, String user, + String output); /** * Invoked when the whole output is updated * @param noteId * @param paragraphId * @param index + * @param user the execution owner, or null when the interpreter does not report one * @param type * @param output */ - void onOutputUpdated( - String noteId, String paragraphId, int index, InterpreterResult.Type type, String output); + void onParagraphOutputUpdated(String noteId, String paragraphId, int index, String user, + InterpreterResult.Type type, String output); /** * Invoked when output is cleared. * @param noteId * @param paragraphId + * @param user the execution owner, or null when the interpreter does not report one */ - void onOutputClear(String noteId, String paragraphId); + void onParagraphOutputClear(String noteId, String paragraphId, String user); /** * Run paragraphs, paragraphs can be specified via indices(paragraphIndices) or ids(paragraphIds) diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java index 15de2d5092b..9c8ee024f5f 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java @@ -28,9 +28,9 @@ public class UpdateOutputBuffer extends AppendOutputBuffer { private final InterpreterResult.Type type; - public UpdateOutputBuffer(String noteId, String paragraphId, int index, + public UpdateOutputBuffer(String noteId, String paragraphId, int index, String user, InterpreterResult.Type type, String data) { - super(noteId, paragraphId, index, data); + super(noteId, paragraphId, index, user, data); this.type = type; } 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 4d31a06558c..8ebae3d7bd9 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 @@ -1758,7 +1758,8 @@ public void onSuccess(Note note, ServiceContext context) throws IOException { * @param output output to append */ @Override - public void onOutputAppend(String noteId, String paragraphId, int index, String output) { + public void onParagraphOutputAppend(String noteId, String paragraphId, int index, String user, + String output) { if (!sendParagraphStatusToFrontend()) { return; } @@ -1772,13 +1773,19 @@ public void onOutputAppend(String noteId, String paragraphId, int index, String if (note == null) { LOGGER.warn("Note {} not found", noteId); } else if (!note.isPersonalizedMode()) { - // Streaming events do not identify the user that owns the execution. connectionManager.broadcast(noteId, msg); + } else if (user != null) { + connectionManager.multicastToUser(user, msg); + } else { + // An interpreter that predates the owner field leaves personalized output unaddressed, + // so it is dropped rather than sent to every reader of the note. + LOGGER.debug("Dropping ownerless personalized output for note {} paragraph {}", + noteId, paragraphId); } return null; }); } catch (IOException e) { - LOGGER.warn("Fail to call onOutputAppend", e); + LOGGER.warn("Fail to call onParagraphOutputAppend", e); } } @@ -1788,8 +1795,8 @@ public void onOutputAppend(String noteId, String paragraphId, int index, String * @param output output to update (replace) */ @Override - public void onOutputUpdated(String noteId, String paragraphId, int index, - InterpreterResult.Type type, String output) { + public void onParagraphOutputUpdated(String noteId, String paragraphId, int index, + String user, InterpreterResult.Type type, String output) { if (!sendParagraphStatusToFrontend()) { return; } @@ -1807,10 +1814,19 @@ public void onOutputUpdated(String noteId, String paragraphId, int index, return null; } if (note.isPersonalizedMode()) { - // Streaming events carry no owner. The shared outputBuffer is what checkpointOutput - // saves as the shared result and what other users' paragraphs are cloned from, so - // one user's output must not be written there. Personalized clients get their - // user-specific terminal snapshot instead. + if (user == null) { + LOGGER.debug("Dropping ownerless personalized output for note {} paragraph {}", + noteId, paragraphId); + return null; + } + // The shared outputBuffer is both what checkpointOutput saves and what new users' + // copies are cloned from, so one user's output goes to that user's own copy. + Paragraph userParagraph = + note.getParagraph(paragraphId).getUserParagraphMap().get(user); + if (userParagraph != null) { + userParagraph.updateOutputBuffer(index, type, output); + } + connectionManager.multicastToUser(user, msg); return null; } note.getParagraph(paragraphId).updateOutputBuffer(index, type, output); @@ -1818,7 +1834,7 @@ public void onOutputUpdated(String noteId, String paragraphId, int index, return null; }); } catch (IOException e) { - LOGGER.warn("Fail to call onOutputUpdated", e); + LOGGER.warn("Fail to call onParagraphOutputUpdated", e); } } @@ -1826,7 +1842,7 @@ public void onOutputUpdated(String noteId, String paragraphId, int index, * This callback is for the paragraph that runs on ZeppelinServer. */ @Override - public void onOutputClear(String noteId, String paragraphId) { + public void onParagraphOutputClear(String noteId, String paragraphId, String user) { if (!sendParagraphStatusToFrontend()) { return; } @@ -1840,7 +1856,17 @@ public void onOutputClear(String noteId, String paragraphId) { return null; } if (note.isPersonalizedMode()) { - // Streaming events carry no owner, so they must not mutate shared paragraph state. + if (user == null) { + LOGGER.debug("Dropping ownerless personalized clear for note {} paragraph {}", + noteId, paragraphId); + return null; + } + // Clearing the shared paragraph would discard output the other users still own. + if (note.getParagraph(paragraphId).getUserParagraphMap().containsKey(user)) { + Paragraph userParagraph = note.clearPersonalizedParagraphOutput(paragraphId, user); + connectionManager.multicastToUser(user, new Message(OP.PARAGRAPH) + .withMsgId(MSG_ID_NOT_DEFINED).put("paragraph", userParagraph)); + } return null; } note.clearParagraphOutput(paragraphId); @@ -1850,7 +1876,7 @@ public void onOutputClear(String noteId, String paragraphId) { }); } catch (IOException e) { - LOGGER.warn("Fail to call onOutputClear", e); + LOGGER.warn("Fail to call onParagraphOutputClear", e); } } diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java index db73b10a427..8fac268b9e5 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java @@ -63,9 +63,10 @@ void updateOutputCompletesBeforeReturning() throws Exception { RemoteInterpreterEventServer server = serverWithRunner(listener, new AppendOutputRunner(listener)); try { - server.updateOutput(new OutputUpdateEvent("note", "para", 0, "TEXT", "final", null)); + server.updateOutput(new OutputUpdateEvent("note", "para", 0, "TEXT", "final", null, null)); // A caller may publish terminal status as soon as the RPC returns. - verify(listener).onOutputUpdated("note", "para", 0, InterpreterResult.Type.TEXT, "final"); + verify(listener).onParagraphOutputUpdated( + "note", "para", 0, null, InterpreterResult.Type.TEXT, "final"); } finally { server.stop(); } @@ -77,10 +78,10 @@ void checkpointDrainsPendingOutputBeforeSaving() throws Exception { RemoteInterpreterEventServer server = serverWithRunner(listener, new AppendOutputRunner(listener)); try { - server.appendOutput(new OutputAppendEvent("note", "para", 0, "pending", null)); + server.appendOutput(new OutputAppendEvent("note", "para", 0, "pending", null, null)); server.checkpointOutput("note", "para"); InOrder order = inOrder(listener); - order.verify(listener).onOutputAppend("note", "para", 0, "pending"); + order.verify(listener).onParagraphOutputAppend("note", "para", 0, null, "pending"); order.verify(listener).checkpointOutput("note", "para"); } finally { server.stop(); @@ -93,19 +94,19 @@ void updateAllIsAnOrderedClearAndReplacement() throws Exception { AppendOutputRunner runner = new AppendOutputRunner(listener); RemoteInterpreterEventServer server = serverWithRunner(listener, runner); try { - runner.appendBuffer("note", "para", 0, "old"); + runner.appendBuffer("note", "para", 0, null, "old"); server.updateAllOutput(new OutputUpdateAllEvent("note", "para", Collections.singletonList( - new RemoteInterpreterResultMessage("HTML", "replacement")))); - verify(listener).onOutputUpdated("note", "para", 0, - InterpreterResult.Type.HTML, "replacement"); - runner.appendBuffer("note", "para", 0, "new"); + new RemoteInterpreterResultMessage("HTML", "replacement")), null)); + verify(listener).onParagraphOutputUpdated( + "note", "para", 0, null, InterpreterResult.Type.HTML, "replacement"); + runner.appendBuffer("note", "para", 0, null, "new"); runner.run(); InOrder order = inOrder(listener); - order.verify(listener).onOutputAppend("note", "para", 0, "old"); - order.verify(listener).onOutputClear("note", "para"); - order.verify(listener).onOutputUpdated("note", "para", 0, - InterpreterResult.Type.HTML, "replacement"); - order.verify(listener).onOutputAppend("note", "para", 0, "new"); + order.verify(listener).onParagraphOutputAppend("note", "para", 0, null, "old"); + order.verify(listener).onParagraphOutputClear("note", "para", null); + order.verify(listener).onParagraphOutputUpdated( + "note", "para", 0, null, InterpreterResult.Type.HTML, "replacement"); + order.verify(listener).onParagraphOutputAppend("note", "para", 0, null, "new"); } finally { server.stop(); } @@ -123,18 +124,18 @@ void updateAllWaitsForInFlightAppendAndCompletesBeforeReturning() throws Excepti entered.countDown(); assertTrue(release.await(5, TimeUnit.SECONDS)); return null; - }).when(listener).onOutputAppend("note", "para", 0, "old"); + }).when(listener).onParagraphOutputAppend("note", "para", 0, null, "old"); ExecutorService executor = Executors.newFixedThreadPool(2); try { - runner.appendBuffer("note", "para", 0, "old"); + runner.appendBuffer("note", "para", 0, null, "old"); Future first = executor.submit(runner); assertTrue(entered.await(5, TimeUnit.SECONDS)); Future update = executor.submit(() -> { updateStarted.countDown(); server.updateAllOutput(new OutputUpdateAllEvent("note", "para", Collections.singletonList( - new RemoteInterpreterResultMessage("HTML", "replacement")))); - verify(listener).onOutputUpdated("note", "para", 0, - InterpreterResult.Type.HTML, "replacement"); + new RemoteInterpreterResultMessage("HTML", "replacement")), null)); + verify(listener).onParagraphOutputUpdated( + "note", "para", 0, null, InterpreterResult.Type.HTML, "replacement"); return null; }); assertTrue(updateStarted.await(5, TimeUnit.SECONDS)); @@ -143,10 +144,10 @@ void updateAllWaitsForInFlightAppendAndCompletesBeforeReturning() throws Excepti first.get(5, TimeUnit.SECONDS); update.get(5, TimeUnit.SECONDS); InOrder order = inOrder(listener); - order.verify(listener).onOutputAppend("note", "para", 0, "old"); - order.verify(listener).onOutputClear("note", "para"); - order.verify(listener).onOutputUpdated("note", "para", 0, - InterpreterResult.Type.HTML, "replacement"); + order.verify(listener).onParagraphOutputAppend("note", "para", 0, null, "old"); + order.verify(listener).onParagraphOutputClear("note", "para", null); + order.verify(listener).onParagraphOutputUpdated( + "note", "para", 0, null, InterpreterResult.Type.HTML, "replacement"); } finally { release.countDown(); executor.shutdownNow(); diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java index 2d6e08beba7..786c70e23c8 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java @@ -47,6 +47,7 @@ import static org.junit.jupiter.api.Assertions.fail; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.Mockito.atMost; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.inOrder; @@ -79,8 +80,9 @@ void testSingleEvent() throws InterruptedException { String[][] buffer = {{"note", "para", "data\n"}}; loopForCompletingEvents(listener, 1, buffer); - verify(listener, times(1)).onOutputAppend(any(String.class), any(String.class), anyInt(), any(String.class)); - verify(listener, times(1)).onOutputAppend("note", "para", 0, "data\n"); + verify(listener, times(1)).onParagraphOutputAppend( + any(String.class), any(String.class), anyInt(), isNull(), any(String.class)); + verify(listener, times(1)).onParagraphOutputAppend("note", "para", 0, null, "data\n"); } @Test @@ -95,26 +97,78 @@ public void testMultipleEventsOfSameParagraph() throws InterruptedException { }; loopForCompletingEvents(listener, 1, buffer); - verify(listener, times(1)).onOutputAppend(any(String.class), any(String.class), anyInt(), any(String.class)); - verify(listener, times(1)).onOutputAppend(note1, para1, 0, "data1\ndata2\ndata3\n"); + verify(listener, times(1)).onParagraphOutputAppend( + any(String.class), any(String.class), anyInt(), isNull(), any(String.class)); + verify(listener, times(1)).onParagraphOutputAppend( + note1, para1, 0, null, "data1\ndata2\ndata3\n"); + } + + // A paragraph in shared mode is one Job, so one execution -- one user -- produces the output + // for a given index at a time, and keying by user must leave that batching alone. + @Test + void appendsFromOneExecutionAreStillBatchedIntoOneChunk() { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + AppendOutputRunner runner = new AppendOutputRunner(listener); + runner.appendBuffer("note", "para", 0, "user1", "line-1\n"); + runner.appendBuffer("note", "para", 0, "user1", "line-2\n"); + + runner.run(); + + verify(listener, times(1)).onParagraphOutputAppend( + "note", "para", 0, "user1", "line-1\nline-2\n"); + } + + // Two executions only overlap in personalized mode, where each user runs their own copy. + // A merged chunk would have no single owner and could not be routed to either user. + @Test + void appendsFromDifferentExecutionsAreNotBatchedTogether() { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + AppendOutputRunner runner = new AppendOutputRunner(listener); + runner.appendBuffer("note", "para", 0, "user1", "mine\n"); + runner.appendBuffer("note", "para", 0, "user2", "theirs\n"); + + runner.run(); + + verify(listener, times(1)).onParagraphOutputAppend("note", "para", 0, "user1", "mine\n"); + verify(listener, times(1)).onParagraphOutputAppend("note", "para", 0, "user2", "theirs\n"); + } + + // Keying by user splits the buffer, so the per-owner ordering the shared queue guarantees + // must still hold once another user's output is interleaved. + @Test + void updatesDoNotOvertakeQueuedAppendsOfTheSameOwner() { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + AppendOutputRunner runner = new AppendOutputRunner(listener); + runner.appendBuffer("note", "para", 0, "owner", "before\n"); + runner.appendBuffer("note", "para", 0, "other", "theirs\n"); + runner.updateBuffer("note", "para", 0, "owner", InterpreterResult.Type.TEXT, "replacement\n"); + runner.appendBuffer("note", "para", 0, "owner", "after\n"); + + runner.run(); + + InOrder order = inOrder(listener); + order.verify(listener).onParagraphOutputAppend("note", "para", 0, "owner", "before\n"); + order.verify(listener).onParagraphOutputUpdated( + "note", "para", 0, "owner", InterpreterResult.Type.TEXT, "replacement\n"); + order.verify(listener).onParagraphOutputAppend("note", "para", 0, "owner", "after\n"); } @Test void testUpdateDoesNotOvertakeQueuedAppend() { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); AppendOutputRunner runner = new AppendOutputRunner(listener); - runner.appendBuffer("note", "para", 0, "before-1\n"); - runner.appendBuffer("note", "para", 0, "before-2\n"); - runner.updateBuffer("note", "para", 0, InterpreterResult.Type.TEXT, "replacement\n"); - runner.appendBuffer("note", "para", 0, "after\n"); + runner.appendBuffer("note", "para", 0, null, "before-1\n"); + runner.appendBuffer("note", "para", 0, null, "before-2\n"); + runner.updateBuffer("note", "para", 0, null, InterpreterResult.Type.TEXT, "replacement\n"); + runner.appendBuffer("note", "para", 0, null, "after\n"); runner.run(); InOrder order = inOrder(listener); - order.verify(listener).onOutputAppend("note", "para", 0, "before-1\nbefore-2\n"); - order.verify(listener).onOutputUpdated( - "note", "para", 0, InterpreterResult.Type.TEXT, "replacement\n"); - order.verify(listener).onOutputAppend("note", "para", 0, "after\n"); + order.verify(listener).onParagraphOutputAppend("note", "para", 0, null, "before-1\nbefore-2\n"); + order.verify(listener).onParagraphOutputUpdated( + "note", "para", 0, null, InterpreterResult.Type.TEXT, "replacement\n"); + order.verify(listener).onParagraphOutputAppend("note", "para", 0, null, "after\n"); } @Test @@ -132,11 +186,12 @@ void testMultipleEventsOfDifferentParagraphs() throws InterruptedException { }; loopForCompletingEvents(listener, 4, buffer); - verify(listener, times(4)).onOutputAppend(any(String.class), any(String.class), anyInt(), any(String.class)); - verify(listener, times(1)).onOutputAppend(note1, para1, 0, "data1\n"); - verify(listener, times(1)).onOutputAppend(note1, para2, 0, "data2\n"); - verify(listener, times(1)).onOutputAppend(note2, para1, 0, "data3\n"); - verify(listener, times(1)).onOutputAppend(note2, para2, 0, "data4\n"); + verify(listener, times(4)).onParagraphOutputAppend( + any(String.class), any(String.class), anyInt(), isNull(), any(String.class)); + verify(listener, times(1)).onParagraphOutputAppend(note1, para1, 0, null, "data1\n"); + verify(listener, times(1)).onParagraphOutputAppend(note1, para2, 0, null, "data2\n"); + verify(listener, times(1)).onParagraphOutputAppend(note2, para1, 0, null, "data3\n"); + verify(listener, times(1)).onParagraphOutputAppend(note2, para2, 0, null, "data4\n"); } @Test @@ -155,7 +210,8 @@ void testClubbedData() throws InterruptedException { * calls, 30-40 Web-socket calls are made. Keeping * the unit-test to a pessimistic 100 web-socket calls. */ - verify(listener, atMost(NUM_CLUBBED_EVENTS)).onOutputAppend(any(String.class), any(String.class), anyInt(), any(String.class)); + verify(listener, atMost(NUM_CLUBBED_EVENTS)).onParagraphOutputAppend( + any(String.class), any(String.class), anyInt(), isNull(), any(String.class)); } @Test @@ -166,7 +222,7 @@ void testWarnLoggerForLargeData() throws InterruptedException { int numEvents = 100000; for (int i=0; i first = executor.submit(runner); assertTrue(entered.await(5, TimeUnit.SECONDS)); - runner.updateBuffer("note", "para", 0, InterpreterResult.Type.TEXT, "new"); + runner.updateBuffer("note", "para", 0, null, InterpreterResult.Type.TEXT, "new"); Future second = executor.submit(runner); assertThrows(TimeoutException.class, () -> second.get(100, TimeUnit.MILLISECONDS)); release.countDown(); first.get(5, TimeUnit.SECONDS); second.get(5, TimeUnit.SECONDS); InOrder order = inOrder(listener); - order.verify(listener).onOutputAppend("note", "para", 0, "old"); - order.verify(listener).onOutputUpdated("note", "para", 0, - InterpreterResult.Type.TEXT, "new"); + order.verify(listener).onParagraphOutputAppend("note", "para", 0, null, "old"); + order.verify(listener).onParagraphOutputUpdated( + "note", "para", 0, null, InterpreterResult.Type.TEXT, "new"); } finally { release.countDown(); executor.shutdownNow(); @@ -268,7 +324,7 @@ public void run() { String noteId = "noteId"; String paraId = "paraId"; for (int i=0; i { + const frames: string[] = []; + page.on('websocket', socket => { + socket.on('framereceived', frame => { + const payload = typeof frame.payload === 'string' ? frame.payload : frame.payload.toString(); + if (payload.includes('PARAGRAPH_APPEND_OUTPUT') || payload.includes('PARAGRAPH_UPDATE_OUTPUT')) { + frames.push(payload); + } + }); + }); + return frames; +}; + +/** The pid the shell printed, which differs for every execution of the same code. */ +const markerIn = (frames: string[]): string => { + const marker = /MARK-\d+/.exec(frames.join('\n')); + expect(marker, `no marker in ${frames.length} streaming frames`).not.toBeNull(); + return marker![0]; +}; + +test.describe('Personalized streaming scope', () => { + // JUSTIFIED: one shared notebook and two signed-in principals must stay within one worker. + test.describe.configure({ mode: 'default' }); + addPageAnnotationBeforeEach(PAGES.WORKSPACE.NOTEBOOK); + + let ownerContext: BrowserContext; + let otherContext: BrowserContext; + + test.afterEach(async () => { + await ownerContext?.close(); + await otherContext?.close(); + }); + + test('sends each user only the streaming output of their own run', async ({ browser }) => { + const accounts = await getTwoTestAccounts(); + test.skip(!accounts, 'Two shiro accounts are required to tell personalized users apart'); + const [ownerAccount, otherAccount] = accounts!; + + ownerContext = await browser.newContext({ storageState: SIGNED_OUT }); + otherContext = await browser.newContext({ storageState: SIGNED_OUT }); + const ownerPage = await ownerContext.newPage(); + const otherPage = await otherContext.newPage(); + const ownerFrames = recordStreamingFrames(ownerPage); + const otherFrames = recordStreamingFrames(otherPage); + + const ownerNote = new CollaborationPage(ownerPage); + const otherNote = new CollaborationPage(otherPage); + const ownerParagraph = new NotebookParagraphPage(ownerPage); + const otherParagraph = new NotebookParagraphPage(otherPage); + + await test.step('Given a personalized note two signed-in users share', async () => { + await loginAs(ownerPage, ownerAccount); + await loginAs(otherPage, otherAccount); + + const { noteId } = await createTestNotebook(ownerPage); + await ownerNote.openNotebook(noteId); + await ownerNote.switchToPersonalModeButton.click(); + await ownerNote.confirmPersonalizedModeChange(); + await expect(ownerNote.switchToCollaborationModeButton).toBeVisible({ timeout: 15000 }); + + await otherNote.openNotebook(noteId); + }); + + await test.step('When both run that paragraph', async () => { + // Personalized mode shares and live-syncs the paragraph text, so the two users cannot hold + // different code. The shell pid differs per run, giving each execution its own marker. + await ownerNote.typeInEditor('%sh\necho MARK-$$'); + await expect(otherNote.editorText).toContainText('MARK-', { timeout: 15000 }); + + await ownerParagraph.runParagraph(); + await otherParagraph.runParagraph(); + + await expect(ownerParagraph.status).toHaveText('FINISHED'); + await expect(otherParagraph.status).toHaveText('FINISHED'); + }); + + await test.step('Then neither socket carried the other run output', async () => { + const ownerMarker = markerIn(ownerFrames); + const otherMarker = markerIn(otherFrames); + expect(ownerMarker).not.toEqual(otherMarker); + + expect(ownerFrames.join('\n')).not.toContain(otherMarker); + expect(otherFrames.join('\n')).not.toContain(ownerMarker); + }); + + await test.step('And the terminal result each user sees matches their own run', async () => { + const ownerMarker = markerIn(ownerFrames); + const otherMarker = markerIn(otherFrames); + + await expect(ownerParagraph.resultDisplay).toContainText(ownerMarker); + await expect(ownerParagraph.resultDisplay).not.toContainText(otherMarker); + await expect(otherParagraph.resultDisplay).toContainText(otherMarker); + await expect(otherParagraph.resultDisplay).not.toContainText(ownerMarker); + }); + }); + + test('sends no streaming output to a user who only opened the note', async ({ browser }) => { + const accounts = await getTwoTestAccounts(); + test.skip(!accounts, 'Two shiro accounts are required to tell personalized users apart'); + const [ownerAccount, otherAccount] = accounts!; + + ownerContext = await browser.newContext({ storageState: SIGNED_OUT }); + otherContext = await browser.newContext({ storageState: SIGNED_OUT }); + const ownerPage = await ownerContext.newPage(); + const watcherPage = await otherContext.newPage(); + const ownerFrames = recordStreamingFrames(ownerPage); + const watcherFrames = recordStreamingFrames(watcherPage); + + await loginAs(ownerPage, ownerAccount); + await loginAs(watcherPage, otherAccount); + + const { noteId } = await createTestNotebook(ownerPage); + const ownerNote = new CollaborationPage(ownerPage); + await ownerNote.openNotebook(noteId); + await ownerNote.switchToPersonalModeButton.click(); + await ownerNote.confirmPersonalizedModeChange(); + await expect(ownerNote.switchToCollaborationModeButton).toBeVisible({ timeout: 15000 }); + await new CollaborationPage(watcherPage).openNotebook(noteId); + + await ownerNote.typeInEditor('%sh\necho RUNNER-ONLY'); + const ownerParagraph = new NotebookParagraphPage(ownerPage); + await ownerParagraph.runParagraph(); + await expect(ownerParagraph.status).toHaveText('FINISHED'); + await expect(ownerParagraph.resultDisplay).toContainText('RUNNER-ONLY'); + + // Both halves matter. On their own "the watcher got nothing" also holds when the server + // sends the output to nobody, so it would pass against a build that simply drops it. + expect(ownerFrames.join('\n')).toContain('RUNNER-ONLY'); + expect(watcherFrames).toHaveLength(0); + await expect(new NotebookParagraphPage(watcherPage).resultDisplay).toHaveCount(0); + }); +}); diff --git a/zeppelin-web-angular/e2e/utils.ts b/zeppelin-web-angular/e2e/utils.ts index dd1c6262431..06cacc8ee0d 100644 --- a/zeppelin-web-angular/e2e/utils.ts +++ b/zeppelin-web-angular/e2e/utils.ts @@ -13,7 +13,7 @@ import { globSync } from 'fs'; import { join, sep } from 'path'; import { test, expect, Page, TestInfo } from '@playwright/test'; -import { LoginTestUtil } from './models/login-page.util'; +import { LoginTestUtil, TestCredentials } from './models/login-page.util'; import { E2E_TEST_FOLDER } from './models/base-page'; import { LoginPage } from './models/login-page'; @@ -298,6 +298,41 @@ export const performLoginIfRequired = async (page: Page): Promise => { return false; }; +/** + * Signs in as one specific account. performLoginIfRequired always takes the first configured + * user, which cannot express a test that needs two distinct principals at once. + */ +export const loginAs = async (page: Page, credentials: TestCredentials): Promise => { + const loginPage = new LoginPage(page); + await loginPage.navigate(); + // Wait for the form rather than probing visibility, which resolves before Angular renders it. + await loginPage.userNameInput.waitFor({ state: 'visible', timeout: 30000 }); + await loginPage.login(credentials.username, credentials.password); + await page.waitForSelector('zeppelin-login', { state: 'hidden', timeout: 30000 }); + await page.evaluate(() => { + if (window.location.hash.includes('login')) { + window.location.hash = '#/'; + } + }); + // The note list only renders once the authenticated socket has delivered it, so it is the + // signal that this principal can open a notebook -- not just that the form was accepted. + await page.waitForSelector('zeppelin-node-list', { timeout: 30000 }); + await waitForZeppelinReady(page); +}; + +/** + * Two distinct shiro accounts, or null when the deployment cannot provide them. + */ +export const getTwoTestAccounts = async (): Promise<[TestCredentials, TestCredentials] | null> => { + if (!(await LoginTestUtil.isShiroEnabled())) { + return null; + } + const accounts = Object.values(await LoginTestUtil.getTestCredentials()).filter( + account => account.username && account.password + ); + return accounts.length >= 2 ? [accounts[0], accounts[1]] : null; +}; + export const skipWhenAuthenticationIsStillRequired = async (page: Page): Promise => { const loginStillVisible = await page .locator('zeppelin-login')