Skip to content

[ZEPPELIN-6704] Scope personalized streaming output to its execution user #5479

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
big-cir wants to merge 3 commits into
base: master
Choose a base branch
from
+929 −176
Open

[ZEPPELIN-6704] Scope personalized streaming output to its execution user #5479

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3977d06
[ZEPPELIN-6704] Carry the execution owner through interpreter output …
big-cir Sep 13, 2026
c2a1f36
[ZEPPELIN-6704] Deliver personalized streaming output to its executio…
big-cir Sep 13, 2026
9ca307d
[ZEPPELIN-6704] Add two-user e2e coverage for personalized streaming …
big-cir Sep 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinApplicationDevServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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);
}
Expand Down
7 changes: 4 additions & 3 deletions helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinDevServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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);
}
Expand Down
14 changes: 7 additions & 7 deletions ...er/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand All @@ -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;
});

Expand All @@ -250,11 +250,11 @@ public void onInterpreterOutputUpdate(
}

public void onInterpreterOutputUpdateAll(
String noteId, String paragraphId, List<InterpreterResultMessage> messages) {
String noteId, String paragraphId, String user, List<InterpreterResultMessage> messages) {
try {
callRemoteFunction(client -> {
client.updateAllOutput(
new OutputUpdateAllEvent(noteId, paragraphId, convertToThrift(messages)));
new OutputUpdateAllEvent(noteId, paragraphId, convertToThrift(messages), user));
return null;
});

Expand Down
15 changes: 10 additions & 5 deletions ...rpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -955,7 +955,12 @@ public List<InterpreterCompletion> 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.
Comment on lines +958 to +959

@tbonelee tbonelee Sep 20, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Two of the three things this comment says are already in the code. The value is an argument to createInterpreterOutput, so it cannot be computed after the output exists, and the listener captures it as a final parameter exactly the way it captures noteId and paragraphId, neither of which carries such a comment.

What is left, that this is the execution owner, fits in a name:

private InterpreterContext convert(RemoteInterpreterContext ric) {
  return convert(ric, createInterpreterOutput(
      ric.getNoteId(), ric.getParagraphId(), executionOwnerOf(ric)));
}

private static String executionOwnerOf(RemoteInterpreterContext ric) {
  AuthenticationInfo authenticationInfo =
      AuthenticationInfo.fromJson(ric.getAuthenticationInfo());
  return authenticationInfo == null ? null : authenticationInfo.getUser();
}

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) {
Expand All @@ -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);
}
Expand All @@ -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
Expand All @@ -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);
}
Expand Down
2 changes: 1 addition & 1 deletion ...lin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AngularObjectId.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<AngularObjectId, AngularObjectId._Fields>, java.io.Serializable, Cloneable, Comparable<AngularObjectId> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AngularObjectId");

Expand Down
2 changes: 1 addition & 1 deletion ...nterpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputAppendEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<AppOutputAppendEvent, AppOutputAppendEvent._Fields>, java.io.Serializable, Cloneable, Comparable<AppOutputAppendEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AppOutputAppendEvent");

Expand Down
2 changes: 1 addition & 1 deletion ...nterpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputUpdateEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<AppOutputUpdateEvent, AppOutputUpdateEvent._Fields>, java.io.Serializable, Cloneable, Comparable<AppOutputUpdateEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AppOutputUpdateEvent");

Expand Down
2 changes: 1 addition & 1 deletion ...nterpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppStatusUpdateEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<AppStatusUpdateEvent, AppStatusUpdateEvent._Fields>, java.io.Serializable, Cloneable, Comparable<AppStatusUpdateEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AppStatusUpdateEvent");

Expand Down
2 changes: 1 addition & 1 deletion ...terpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterCompletion.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<InterpreterCompletion, InterpreterCompletion._Fields>, java.io.Serializable, Cloneable, Comparable<InterpreterCompletion> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("InterpreterCompletion");

Expand Down
2 changes: 1 addition & 1 deletion ...rpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterRPCException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<InterpreterRPCException, InterpreterRPCException._Fields>, java.io.Serializable, Cloneable, Comparable<InterpreterRPCException> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("InterpreterRPCException");

Expand Down
2 changes: 1 addition & 1 deletion ...lin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/LibraryMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<LibraryMetadata, LibraryMetadata._Fields>, java.io.Serializable, Cloneable, Comparable<LibraryMetadata> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("LibraryMetadata");

Expand Down
Loading
Loading