[ZEPPELIN-6737] Connect a private read-only Angular notebook view to Shared Notebook Core

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

@@ -30,6 +30,7 @@
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -147,6 +148,10 @@ String getKey() {
       .setPrettyPrinting()
       .registerTypeAdapterFactory(Input.TypeAdapterFactory).create();
   private static final AtomicReference<NotebookServer> self = new AtomicReference<>();
+  // Read requests whose direct reply and failures carry the request msgId back to the sender.
+  // Broadcasts of the same ops stay uncorrelated.
+  private static final Set<OP> MSG_ID_CORRELATED_READS = Collections.unmodifiableSet(
+      new HashSet<>(Arrays.asList(OP.GET_NOTE, OP.NOTE_REVISION)));
 
   private final ExecutorService executorService = Executors.newFixedThreadPool(10);
 
@@ -596,7 +601,9 @@ public void onMessage(NotebookSocket conn, String msg) {
               + "success=false, errorType={}",
           operation, principal, e.getClass().getSimpleName());
       try {
-        conn.send(serializeMessage(new Message(OP.ERROR_INFO).put("info", e.getMessage())));
+        conn.send(serializeMessage(new Message(OP.ERROR_INFO)
+            .withMsgId(correlatedReadMsgId(receivedMessage))
+            .put("info", e.getMessage())));
       } catch (IOException iox) {
         LOGGER.error("Fail to send error info", iox);
       }
@@ -901,13 +908,13 @@ private void getNote(NotebookSocket conn, ServiceContext context, Message fromMe
       return;
     }
     getNotebookService().getNote(noteId, context,
-        new WebSocketServiceCallback<Note>(conn) {
+        new WebSocketServiceCallback<Note>(conn, fromMessage.msgId) {
           @Override
           public void onSuccess(Note note, ServiceContext context)
               throws IOException {
             connectionManager.addNoteConnection(note.getId(), conn);
-            conn.send(
-                serializeMessage(new Message(OP.NOTE).put("note", note)));
+            conn.send(serializeMessage(
+                new Message(OP.NOTE).withMsgId(fromMessage.msgId).put("note", note)));
             updateAngularObjectRegistry(conn, note);
             sendAllAngularObjects(note, context.getAutheInfo().getUser(),
                 conn);
@@ -1796,12 +1803,13 @@ private void getNoteByRevision(NotebookSocket conn,
     String noteId = (String) fromMessage.get("noteId");
     String revisionId = (String) fromMessage.get("revisionId");
     getNotebookService().getNotebyRevision(noteId, revisionId, context,
-        new WebSocketServiceCallback<Note>(conn) {
+        new WebSocketServiceCallback<Note>(conn, fromMessage.msgId) {
           @Override
           public void onSuccess(Note note, ServiceContext context) throws IOException {
             super.onSuccess(note, context);
             conn.send(serializeMessage(
                 new Message(OP.NOTE_REVISION)
+                    .withMsgId(fromMessage.msgId)
                     .put("noteId", noteId)
                     .put("revisionId", revisionId)
                     .put("note", note)));
@@ -2499,12 +2507,24 @@ private ServiceContext getServiceContext(TicketContainer.Entry ticketEntry) {
     return new ServiceContext(authInfo, userAndRoles);
   }
 
+  private static String correlatedReadMsgId(Message receivedMessage) {
+    return receivedMessage != null && MSG_ID_CORRELATED_READS.contains(receivedMessage.op)
+        ? receivedMessage.msgId : MSG_ID_NOT_DEFINED;
+  }
+
   public class WebSocketServiceCallback<T> extends SimpleServiceCallback<T> {
 
     private final NotebookSocket conn;
+    private final String msgId;
 
     WebSocketServiceCallback(NotebookSocket conn) {
+      this(conn, MSG_ID_NOT_DEFINED);
+    }
+
+    // msgId of the request this callback answers; failures are sent back with it.
+    WebSocketServiceCallback(NotebookSocket conn, String msgId) {
       this.conn = conn;
+      this.msgId = msgId;
     }
 
     @Override
@@ -2516,13 +2536,16 @@ public void onFailure(Exception ex, ServiceContext context) throws IOException {
         Map<String, String> jsonObject =
             gson.fromJson(((ForbiddenException) ex).getResponse().getEntity().toString(), type);
         conn.send(serializeMessage(new Message(OP.AUTH_INFO)
+            .withMsgId(msgId)
             .put("info", jsonObject.get("message"))));
       } else {
         String message = ex.getMessage();
         if (ex.getCause() != null) {
           message += ", cause: " + ex.getCause().getMessage();
         }
-        conn.send(serializeMessage(new Message(OP.ERROR_INFO).put("info", message)));
+        conn.send(serializeMessage(new Message(OP.ERROR_INFO)
+            .withMsgId(msgId)
+            .put("info", message)));
       }
     }
   }

zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java

@@ -49,6 +49,8 @@
 import java.util.concurrent.Callable;
 import java.util.stream.Collectors;
 
+import com.google.gson.JsonParser;
+
 import org.apache.commons.io.IOUtils;
 import org.apache.thrift.TException;
 import org.apache.zeppelin.conf.ZeppelinConfiguration;
@@ -1250,6 +1252,170 @@ void setNoteRevisionIncludesNoteId() throws IOException {
     }
   }
 
+  @Test
+  void getNoteReplyCarriesTheRequestMsgId() throws IOException {
+    String noteId = notebook.createNote("read-reply-msg-id", anonymous);
+    try {
+      NotebookSocket socket = createWebSocket();
+
+      notebookServer.onMessage(socket,
+          new Message(OP.GET_NOTE).withMsgId("read-1").put("id", noteId).toJson());
+
+      Message reply = onlySent(socket, OP.NOTE);
+      assertEquals("read-1", reply.msgId);
+    } finally {
+      notebook.removeNote(noteId, anonymous);
+    }
+  }
+
+  @Test
+  void getNoteWithoutMsgIdKeepsTheReplyUncorrelated() throws IOException {
+    String noteId = notebook.createNote("read-reply-without-msg-id", anonymous);
+    try {
+      NotebookSocket socket = createWebSocket();
+
+      notebookServer.onMessage(socket, new Message(OP.GET_NOTE).put("id", noteId).toJson());
+
+      String reply = onlySentPayload(socket, OP.NOTE);
+      assertNull(notebookServer.deserializeMessage(reply).msgId);
+      assertFalse(JsonParser.parseString(reply).getAsJsonObject().has("msgId"));
+    } finally {
+      notebook.removeNote(noteId, anonymous);
+    }
+  }
+
+  @Test
+  void getNoteFailuresCarryTheRequestMsgId() throws IOException {
+    NotebookSocket socket = createWebSocket();
+
+    notebookServer.onMessage(socket,
+        new Message(OP.GET_NOTE).withMsgId("read-missing").put("id", "missing-note").toJson());
+
+    assertEquals("read-missing", onlySent(socket, OP.ERROR_INFO).msgId);
+
+    String noteId = notebook.createNote("read-forbidden-msg-id", anonymous);
+    try {
+      setNotePermissions(noteId, "read-owner", "read-owner");
+      reset(socket);
+
+      notebookServer.onMessage(socket,
+          new Message(OP.GET_NOTE).withMsgId("read-forbidden").put("id", noteId).toJson());
+
+      assertEquals("read-forbidden", onlySent(socket, OP.AUTH_INFO).msgId);
+      assertTrue(sentOps(socket).stream().noneMatch(op -> op == OP.NOTE));
+    } finally {
+      notebook.removeNote(noteId, anonymous);
+    }
+  }
+
+  @Test
+  void noteRevisionReplyAndFailuresCarryTheRequestMsgId() throws IOException {
+    String noteId = notebook.createNote("revision-reply-msg-id", anonymous);
+    try {
+      NotebookRepoWithVersionControl.Revision revision =
+          notebook.processNote(noteId,
+              note -> notebook.checkpointNote(note.getId(), note.getPath(), "revision", anonymous));
+      NotebookSocket socket = createWebSocket();
+
+      notebookServer.onMessage(socket, new Message(OP.NOTE_REVISION).withMsgId("revision-1")
+          .put("noteId", noteId).put("revisionId", revision.id).toJson());
+
+      Message reply = onlySent(socket, OP.NOTE_REVISION);
+      assertEquals("revision-1", reply.msgId);
+      assertEquals(revision.id, reply.data.get("revisionId"));
+
+      reset(socket);
+      notebookServer.onMessage(socket, new Message(OP.NOTE_REVISION).withMsgId("revision-missing")
+          .put("noteId", "missing-note").put("revisionId", revision.id).toJson());
+
+      assertEquals("revision-missing", onlySent(socket, OP.ERROR_INFO).msgId);
+
+      setNotePermissions(noteId, "revision-owner", "revision-owner");
+      reset(socket);
+      notebookServer.onMessage(socket, new Message(OP.NOTE_REVISION).withMsgId("revision-forbidden")
+          .put("noteId", noteId).put("revisionId", revision.id).toJson());
+
+      assertEquals("revision-forbidden", onlySent(socket, OP.AUTH_INFO).msgId);
+    } finally {
+      notebook.removeNote(noteId, anonymous);
+    }
+  }
+
+  @Test
+  void readRequestExceptionsReachTheSenderWithTheRequestMsgId() throws IOException {
+    NotebookSocket socket = createWebSocket();
+    // A non-string id fails the cast in the handler before any service callback exists, so the
+    // error comes from the top-level exception handler of onMessage.
+    notebookServer.onMessage(socket,
+        new Message(OP.GET_NOTE).withMsgId("read-throws").put("id", 42).toJson());
+
+    assertEquals("read-throws", onlySent(socket, OP.ERROR_INFO).msgId);
+
+    reset(socket);
+    notebookServer.onMessage(socket, new Message(OP.NOTE_REVISION).withMsgId("revision-throws")
+        .put("noteId", 42).put("revisionId", "some-revision").toJson());
+
+    assertEquals("revision-throws", onlySent(socket, OP.ERROR_INFO).msgId);
+  }
+
+  @Test
+  void unrelatedFailuresAndNoteBroadcastsStayUncorrelated() throws IOException {
+    NotebookSocket socket = createWebSocket();
+    notebookServer.onMessage(socket,
+        new Message(OP.RELOAD_NOTE).withMsgId("reload-missing").put("id", "missing-note").toJson());
+
+    assertNull(onlySent(socket, OP.ERROR_INFO).msgId);
+
+    reset(socket);
+    notebookServer.onMessage(socket,
+        new Message(OP.RELOAD_NOTE).withMsgId("reload-throws").put("id", 42).toJson());
+
+    assertNull(onlySent(socket, OP.ERROR_INFO).msgId);
+
+    String noteId = notebook.createNote("read-broadcast-msg-id", anonymous);
+    try {
+      NotebookSocket reader = createWebSocket();
+      NotebookSocket writer = createWebSocket();
+      notebookServer.onMessage(reader,
+          new Message(OP.GET_NOTE).withMsgId("reader-read").put("id", noteId).toJson());
+      notebookServer.onMessage(writer,
+          new Message(OP.GET_NOTE).withMsgId("writer-read").put("id", noteId).toJson());
+      reset(reader, writer);
+
+      notebookServer.onMessage(writer, new Message(OP.PARAGRAPH_CLEAR_ALL_OUTPUT)
+          .withMsgId("writer-clear").put("id", noteId).toJson());
+
+      assertNull(onlySent(reader, OP.NOTE).msgId);
+      assertNull(onlySent(writer, OP.NOTE).msgId);
+    } finally {
+      notebook.removeNote(noteId, anonymous);
+    }
+  }
+
+  private List<String> sentPayloads(NotebookSocket socket) throws IOException {
+    ArgumentCaptor<String> sent = ArgumentCaptor.forClass(String.class);
+    verify(socket, Mockito.atLeast(0)).send(sent.capture());
+    return sent.getAllValues();
+  }
+
+  private List<OP> sentOps(NotebookSocket socket) throws IOException {
+    return sentPayloads(socket).stream()
+        .map(payload -> notebookServer.deserializeMessage(payload).op)
+        .collect(Collectors.toList());
+  }
+
+  private String onlySentPayload(NotebookSocket socket, OP op) throws IOException {
+    List<String> matches = sentPayloads(socket).stream()
+        .filter(payload -> notebookServer.deserializeMessage(payload).op == op)
+        .collect(Collectors.toList());
+    assertEquals(1, matches.size(), "expected exactly one " + op + " but sent " + sentOps(socket));
+    return matches.get(0);
+  }
+
+  private Message onlySent(NotebookSocket socket, OP op) throws IOException {
+    return notebookServer.deserializeMessage(onlySentPayload(socket, op));
+  }
+
   private NotebookSocket createWebSocket() {
     NotebookSocket sock = mock(NotebookSocket.class);
     return sock;

zeppelin-web-angular/e2e/core-contract/README.md

@@ -351,3 +351,36 @@ already rejects stale revision and interpreter-binding replies inside the Angula
 reconnect recovery remains outside it until the lifecycle rules have an enforceable
 stale-reply mechanism. The harness does not implement those lifecycle rules, switch
 the production renderer, or move production notebook state out of Angular.
+
+## Private Core read view captures
+
+[notebook-read](../fixtures/notebook-read/) holds the server replies read by the private Core read
+view (ZEPPELIN-6737). They are separate from the seven lifecycle captures and their inventory.
+
+| Fixture               | Behavior covered                                                                                   |
+| --------------------- | -------------------------------------------------------------------------------------------------- |
+| `read-anonymous.json` | `GET_NOTE` and `NOTE_REVISION` replies, a missing note's `ERROR_INFO`, and note permission reads.  |
+| `read-denied.json`    | An authenticated read the note permissions deny: `AUTH_INFO` and a `403` permissions response.     |
+
+Each read reply and each failure carries the msgId of its request. That depends on a server change
+that is not on Apache master yet, so the committed captures name the branch commit that was built and
+served in `metadata.source`. Recapture from Apache master once that change is merged.
+
+Capture with the server running and its UI served on `PLAYWRIGHT_BASE_URL`; the denied read
+needs an authenticated server with two Shiro accounts:
+
+```bash
+export ZEPPELIN_READ_CAPTURE_COMMIT="$(git rev-parse HEAD)"
+npm run e2e:core-contract:live -- read-fixtures
+npx prettier --write e2e/fixtures/notebook-read
+```
+
+The capture fails unless `/api/version` reports that commit (as its abbreviated id), so build the server in a regular
+clone with `./mvnw package -pl zeppelin-server --am -DskipTests`; a git worktree stamped the main checkout's
+commit when tried. It records that id as `metadata.source.serverCommit`. It also fails when the recorded authentication
+does not match the captured page's session: `read-anonymous.json` needs an anonymous session, `read-denied.json` a
+signed-in one. `metadata.source.repository` is `apache/zeppelin`, so capture from an Apache master commit.
+
+The capture replaces the account names that `AUTH_INFO` prints in its text with `<user>`.
+`src/app/pages/workspace/notebook/core-read/notebook-read-fixture.spec.ts` replays both captures
+through the production read transport and Core.