[ZEPPELIN-6709] Dispatch paragraph output concurrently across notes

[miinhho]

What is this PR for?

Paragraph output currently shares a single synchronized runner. A slow output or checkpoint callback for one note blocks output delivery for unrelated notes.

This PR introduces ParagraphOutputDispatcher, which processes per-note FIFO queues through a shared worker pool. Each note has at most one active worker, while independent notes can progress concurrently. Idle note queues are removed after processing.

APPEND events remain buffered and periodically flushed. UPDATE, UPDATE_ALL, and checkpoint operations preserve ordering with earlier output and complete before their RPC returns. UPDATE_ALL keeps the clear and replacements together.

AppendOutputRunner remains responsible for merging and delivering adjacent append events. Worker count and batch size are configurable; the batch size controls scheduling fairness rather than queue capacity. No additional output limit or rejection policy is introduced.

What type of PR is it?

Bug Fix

What is the Jira issue?

• ZEPPELIN-6709

How should this be tested?

./mvnw -pl zeppelin-server \
  -Dtest=AppendOutputRunnerTest,ParagraphOutputDispatcherTest,RemoteInterpreterEventServerTest,RemoteInterpreterEventServerLibraryTest test

Questions:

• Do the license files need updating? No.
• Are there breaking changes for older versions? No
• Does this need documentation? The new worker-count and batch-size settings are documented in conf/zeppelin-site.xml.template.

[miinhho]

I'd appreciate feedback on the following defaults and scheduling choices:

1. Worker count: 4
   I chose a fixed default of four workers and made it configurable. This value has not been validated against representative workloads. Is four an appropriate default, considering that slow output or checkpoint callbacks occupy workers until they return?

2. Batch size: 1,000 events per note
   This is also a configurable default chosen without performance measurements. A note yields after processing a batch. The budget counts events, not bytes or callback duration, and an UPDATE_ALL operation counts as one event regardless of the number of replacement messages. Is this scheduling granularity appropriate, and is 1,000 a reasonable starting value?

[voidmatcha]

Approving. The per-note dispatcher allows other notes to make progress while a slow output callback is running, provided workers are available. In the benchmark, the fast notes’ p99 latency improved from 937 ms to 235 ms.

One non-blocking suggestion: please consider preserving key-based append coalescing within each batch, without crossing UPDATE, UPDATE_ALL, or checkpoint boundaries. Adjacent-only coalescing increased listener calls and latency in a high-load benchmark with interleaved paragraph output; the impact under typical usage is still unclear. A LinkedHashMap and an interleaved-output regression test could address this here or in a follow-up.

One question: checkpoint saves now occupy output workers. Since VFSNotebookRepo.save() is synchronized, one slow save with other notes’ checkpoints waiting behind it could occupy all workers and delay unrelated output. Is this an acceptable tradeoff?

[miinhho]

Thanks for pointing this out.

As you described, checkpoint saves now run on output workers. NoteManager.saveNote() holds its monitor through the repository save, so a slow save and other checkpoints waiting behind it could occupy all output workers and delay unrelated output. I hadn’t considered that path sufficiently.

After tracing the flow, I think we can keep checkpoint as a per-note ordering boundary while running the save on a separate executor. The dispatcher would deliver earlier appends, pause that note at the checkpoint, and resume its queued output when the save callback finishes. The checkpoint RPC would continue to wait for that callback.

[miinhho]

Regarding your suggestion on key-based append coalescing, I revisited the interleaved-paragraph case and updated the PR (commit) to coalesce appends by (noteId, paragraphId, index) using a LinkedHashMap, without crossing UPDATE, UPDATE_ALL, or checkpoint boundaries.

[miinhho]

Following up on the checkpoint worker concern, I put a proposed fix in this branch. The dispatcher pauses only the affected note at its checkpoint, runs the save callback on a separate fixed worker pool, and resumes that note when the callback finishes.

I checked the lock-contention path as well: with one save holding NoteManager.saveNote()'s monitor and another checkpoint waiting for it, output for an unrelated note still proceeds through processNote() on an output worker.

Would you prefer to include this change in the current PR, or handle it in a follow-up PR?