[ZEPPELIN-6709] Dispatch paragraph output concurrently across notes

conf/zeppelin-site.xml.template

@@ -439,6 +439,21 @@
   <description>Output message from interpreter exceeding the limit will be truncated</description>
 </property>
 
+<property>
+  <name>zeppelin.interpreter.output.worker.count</name>
+  <value>4</value>
+  <description>Number of workers in each of the server output and checkpoint pools. Each note is
+  processed serially; independent notes may run concurrently, and checkpoint saves do not occupy
+  output workers. Tune for listener and repository latency and available server resources.</description>
+</property>
+
+<property>
+  <name>zeppelin.interpreter.output.events.per.batch</name>
+  <value>1000</value>
+  <description>Maximum output events processed for one note before yielding to other ready notes.
+  This controls scheduling granularity, not queue capacity or output truncation.</description>
+</property>
+
 <property>
   <name>zeppelin.ssl</name>
   <value>false</value>

zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java

@@ -1021,6 +1021,9 @@ public enum ConfVars {
     ZEPPELIN_INTERPRETER_CONNECTION_POOL_SIZE("zeppelin.interpreter.connection.poolsize", 100),
     ZEPPELIN_INTERPRETER_GROUP_DEFAULT("zeppelin.interpreter.group.default", "spark"),
     ZEPPELIN_INTERPRETER_OUTPUT_LIMIT("zeppelin.interpreter.output.limit", 1024 * 100),
+    ZEPPELIN_INTERPRETER_OUTPUT_WORKER_COUNT("zeppelin.interpreter.output.worker.count", 4),
+    ZEPPELIN_INTERPRETER_OUTPUT_EVENTS_PER_BATCH(
+        "zeppelin.interpreter.output.events.per.batch", 1000),
     ZEPPELIN_INTERPRETER_INCLUDES("zeppelin.interpreter.include", ""),
     ZEPPELIN_INTERPRETER_EXCLUDES("zeppelin.interpreter.exclude", ""),
 

zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java

@@ -17,6 +17,10 @@
 
 package org.apache.zeppelin.interpreter;
 
+import static org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars
+    .ZEPPELIN_INTERPRETER_OUTPUT_EVENTS_PER_BATCH;
+import static org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars
+    .ZEPPELIN_INTERPRETER_OUTPUT_WORKER_COUNT;
 import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
 import org.apache.commons.io.FileUtils;
@@ -29,6 +33,7 @@
 import org.apache.zeppelin.helium.ApplicationEventListener;
 import org.apache.zeppelin.interpreter.remote.AppendOutputRunner;
 import org.apache.zeppelin.interpreter.remote.InvokeResourceMethodEventMessage;
+import org.apache.zeppelin.interpreter.remote.ParagraphOutputDispatcher;
 import org.apache.zeppelin.interpreter.remote.RemoteAngularObject;
 import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
 import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcessListener;
@@ -69,10 +74,13 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 
 public class RemoteInterpreterEventServer implements RemoteInterpreterEventService.Iface {
 
@@ -88,7 +96,7 @@ public class RemoteInterpreterEventServer implements RemoteInterpreterEventServi
   private final ScheduledExecutorService appendService =
       Executors.newSingleThreadScheduledExecutor();
   private ScheduledFuture<?> appendFuture;
-  private AppendOutputRunner runner;
+  private final ParagraphOutputDispatcher outputDispatcher;
   private final RemoteInterpreterProcessListener listener;
   private final ApplicationEventListener appListener;
 
@@ -99,6 +107,9 @@ public RemoteInterpreterEventServer(ZeppelinConfiguration zConf,
     this.interpreterSettingManager = interpreterSettingManager;
     this.listener = interpreterSettingManager.getRemoteInterpreterProcessListener();
     this.appListener = interpreterSettingManager.getAppEventListener();
+    this.outputDispatcher = new ParagraphOutputDispatcher(listener,
+        zConf.getInt(ZEPPELIN_INTERPRETER_OUTPUT_WORKER_COUNT),
+        zConf.getInt(ZEPPELIN_INTERPRETER_OUTPUT_EVENTS_PER_BATCH));
   }
 
   public void start() throws IOException {
@@ -140,9 +151,8 @@ public void run() {
     }
     LOGGER.info("RemoteInterpreterEventServer is started");
 
-    runner = new AppendOutputRunner(listener);
     appendFuture = appendService.scheduleWithFixedDelay(
-        runner, 0, AppendOutputRunner.BUFFER_TIME_MS, TimeUnit.MILLISECONDS);
+        outputDispatcher::flush, 0, AppendOutputRunner.BUFFER_TIME_MS, TimeUnit.MILLISECONDS);
   }
 
   public void stop() {
@@ -153,6 +163,7 @@ public void stop() {
       appendFuture.cancel(true);
     }
     appendService.shutdownNow();
+    outputDispatcher.close();
     LOGGER.info("RemoteInterpreterEventServer is stopped");
   }
 
@@ -218,8 +229,13 @@ 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.getExecutionOwner(), event.getData());
+      try {
+        outputDispatcher.appendOutput(
+            event.getNoteId(), event.getParagraphId(), event.getIndex(),
+            event.getExecutionOwner(), event.getData());
+      } catch (IllegalStateException e) {
+        throw new InterpreterRPCException(e.toString());
+      }
     } else {
       appListener.onOutputAppend(event.getNoteId(), event.getParagraphId(), event.getIndex(),
           event.getAppId(), event.getData());
@@ -229,11 +245,10 @@ public void appendOutput(OutputAppendEvent event) throws InterpreterRPCException
   @Override
   public void updateOutput(OutputUpdateEvent event) throws InterpreterRPCException, TException {
     if (event.getAppId() == null) {
-      runner.updateBuffer(event.getNoteId(), event.getParagraphId(), event.getIndex(),
+      awaitOutput(event.getNoteId(), () -> outputDispatcher.updateOutput(
+          event.getNoteId(), event.getParagraphId(), event.getIndex(),
           event.getExecutionOwner(), InterpreterResult.Type.valueOf(event.getType()),
-          event.getData());
-      // Complete replacements before the interpreter can publish its terminal result.
-      runner.run();
+          event.getData()));
     } else {
       appListener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), event.getIndex(),
           event.getAppId(), InterpreterResult.Type.valueOf(event.getType()), event.getData());
@@ -242,18 +257,13 @@ public void updateOutput(OutputUpdateEvent event) throws InterpreterRPCException
 
   @Override
   public void updateAllOutput(OutputUpdateAllEvent event) throws InterpreterRPCException, TException {
-    synchronized (runner) {
-      // Finish earlier output before the clear; keep replacements ahead of the next drain.
-      runner.run();
-      listener.onParagraphOutputClear(
-          event.getNoteId(), event.getParagraphId(), event.getExecutionOwner());
-      for (int i = 0; i < event.getMsg().size(); i++) {
-        RemoteInterpreterResultMessage msg = event.getMsg().get(i);
-        listener.onParagraphOutputUpdated(event.getNoteId(), event.getParagraphId(), i,
-            event.getExecutionOwner(), InterpreterResult.Type.valueOf(msg.getType()),
-            msg.getData());
-      }
+    List<InterpreterResultMessage> messages = new ArrayList<>();
+    for (RemoteInterpreterResultMessage message : event.getMsg()) {
+      messages.add(new InterpreterResultMessage(
+          InterpreterResult.Type.valueOf(message.getType()), message.getData()));
     }
+    awaitOutput(event.getNoteId(), () -> outputDispatcher.updateAllOutput(
+        event.getNoteId(), event.getParagraphId(), event.getExecutionOwner(), messages));
   }
 
   @Override
@@ -275,10 +285,23 @@ public void updateAppStatus(AppStatusUpdateEvent event) throws InterpreterRPCExc
 
   @Override
   public void checkpointOutput(String noteId, String paragraphId) throws InterpreterRPCException, TException {
-    // Drain replacements before checkpointing.
-    // Keep storage callbacks outside the runner lock to avoid blocking output delivery.
-    runner.run();
-    listener.checkpointOutput(noteId, paragraphId);
+    awaitOutput(noteId, () -> outputDispatcher.checkpointOutput(noteId, paragraphId));
+  }
+
+  private void awaitOutput(String noteId, Supplier<Future<Void>> submission)
+      throws InterpreterRPCException {
+    try {
+      submission.get().get();
+    } catch (InterruptedException e) {
+      // Interrupting this RPC wait does not cancel output already accepted by the dispatcher.
+      Thread.currentThread().interrupt();
+      throw new InterpreterRPCException("Interrupted while waiting for output: " + noteId);
+    } catch (ExecutionException e) {
+      throw new InterpreterRPCException("Failed to process output for note " + noteId
+          + ": " + e.getCause());
+    } catch (IllegalStateException e) {
+      throw new InterpreterRPCException(e.toString());
+    }
   }
 
   @Override