[ZEPPELIN-6700] Convert a broadcast message to JSON once instead of once per connection

[big-cir]

What is this PR for?

ConnectionManager converts a Message to JSON using serializeMessage and writes the result to a socket. A broadcast sends one message to many connections, so the two steps have different multiplicity: the JSON is one value, the writes are many. Five call sites currently perform the conversion inside the per-connection loop, causing it to run once per connection instead of once per message:

• broadcast(Message), inside the synchronized (connectedSockets) block
• broadcast(String, Message)
• broadcastToWatchers(String, String, Message), itself reached once per broadcast from broadcast, broadcastExcept, and unicast
• broadcastExcept(String, Message, NotebookSocket), the path used by collaborative patches, Angular object updates, and spell results
• unicast(Message, NotebookSocket), reached once per connection from multicastToUser, which serves both note-list updates and personalized-mode paragraphs

The message is not mutated inside these loops, and gson.toJson is deterministic. Every iteration therefore produces a byte-identical string, and all but one are discarded.

There is no functional bug being fixed here. The bytes on the wire and their order are already correct, and on a note with a single connection the current form costs nothing extra. What it does is repeat work in proportion to the number of connections attached to a note, and three paths make that repetition routine rather than occasional:

1. A paragraph state change broadcasts the whole Paragraph including its output, which zeppelin.interpreter.output.limit caps at 100 KB by default. This is the path where a single redundant conversion is expensive.
2. While a paragraph runs, streaming output is broadcast once per AppendOutputRunner flush. Each message carries only the appended chunk and the runner coalesces writes over a 100 ms window, so these are small but numerous.
3. In collaborative mode, every edit to a paragraph broadcasts a patch to the other connections on the note.

The features that make a note worth sharing are the ones that pay for this most.

This PR converts once per broadcast at those five sites and writes that one string to every connection. The resulting JSON and write order remain unchanged. Two supporting changes come with it:

• broadcastToWatchers returns early when no watcher is attached, so hoisting the conversion out of its loop does not introduce work in the common case where the loop body never ran.
• multicastToUser now sends directly rather than delegating to unicast, because unicast bundles the conversion with a watcher broadcast. The watcher broadcast stays inside the loop, so watchers receive the same messages they do today.

One behavioral detail is worth flagging for review. Hoisting moves the conversion out of the per-connection try that exists to catch IOException from NotebookSocket.send. A conversion failure was logged and skipped per connection before, and now propagates to the caller. That handler was incidental to send rather than an intentional contract for the conversion, and it never caught StackOverflowError, which is the likely failure mode for a cyclic object graph.

NotebookServer is untouched. Its own serializeMessage is used for single sends, and the one loop there that converts per iteration builds a different message each time, so there is nothing to hoist.

What type of PR is it?

Improvement

Todos

• - Convert once per broadcast in broadcast(Message), broadcast(String, Message), broadcastExcept, and broadcastToWatchers
• - Return early from broadcastToWatchers when no watcher is attached
• - Send directly from multicastToUser instead of going through unicast, keeping the watcher broadcast inside the loop
• - Add tests asserting that a single broadcast to multiple connections converts once and delivers identical payloads

What is the Jira issue?

• ZEPPELIN-6700

How should this be tested?

Three unit tests were added to ConnectionManagerTest. They subclass ConnectionManager to count how often a broadcast converts its message and assert that every connection receives an identical payload.

./mvnw package -pl zeppelin-server --am \
  -Dtest=ConnectionManagerTest,NotebookServerTest -DfailIfNoTests=false

Screenshots (if appropriate)

N/A

Questions:

• Does the license files need to update? No
• Is there breaking changes for older versions? No
• Does this needs documentation? No