[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 server workers delivering standard paragraph output. Each note is
+  processed serially; independent notes may run concurrently. Tune for listener 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,12 @@ 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.getData());
+      try {
+        outputDispatcher.appendOutput(
+            event.getNoteId(), event.getParagraphId(), event.getIndex(), event.getData());
+      } catch (IllegalStateException e) {
+        throw new InterpreterRPCException(e.toString());
+      }
     } else {
       appListener.onOutputAppend(event.getNoteId(), event.getParagraphId(), event.getIndex(),
           event.getAppId(), event.getData());
@@ -229,10 +244,9 @@ 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(),
-          InterpreterResult.Type.valueOf(event.getType()), event.getData());
-      // Complete replacements before the interpreter can publish its terminal result.
-      runner.run();
+      awaitOutput(event.getNoteId(), () -> outputDispatcher.updateOutput(
+          event.getNoteId(), event.getParagraphId(), event.getIndex(),
+          InterpreterResult.Type.valueOf(event.getType()), event.getData()));
     } else {
       appListener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), event.getIndex(),
           event.getAppId(), InterpreterResult.Type.valueOf(event.getType()), event.getData());
@@ -241,16 +255,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.onOutputClear(event.getNoteId(), event.getParagraphId());
-      for (int i = 0; i < event.getMsg().size(); i++) {
-        RemoteInterpreterResultMessage msg = event.getMsg().get(i);
-        listener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), i,
-            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(), messages));
   }
 
   @Override
@@ -272,10 +283,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

zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java

@@ -17,118 +17,117 @@
 
 package org.apache.zeppelin.interpreter.remote;
 
-import org.apache.zeppelin.interpreter.InterpreterResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
-import java.util.HashMap;
-import java.util.LinkedList;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.Objects;
+import java.util.function.BooleanSupplier;
 
 /**
- * Sends paragraph output periodically. Adjacent append events are batched, while update events
- * share the same queue so that they cannot overtake earlier appends.
+ * Synchronously delivers an append batch, merging events for each paragraph output.
  */
-public class AppendOutputRunner implements Runnable {
-
+public class AppendOutputRunner {
   private static final Logger LOGGER = LoggerFactory.getLogger(AppendOutputRunner.class);
   public static final Long BUFFER_TIME_MS = Long.valueOf(100);
   private static final Long SAFE_PROCESSING_TIME = Long.valueOf(10);
   private static final Long SAFE_PROCESSING_STRING_SIZE = Long.valueOf(100000);
 
-  private final BlockingQueue<AppendOutputBuffer> queue = new LinkedBlockingQueue<>();
   private final RemoteInterpreterProcessListener listener;
 
   public AppendOutputRunner(RemoteInterpreterProcessListener listener) {
     this.listener = listener;
   }
 
-  // Serialize scheduled and RPC drains to preserve callback order.
-  // Empty drains must return immediately to RPC callers.
-  @Override
-  public synchronized void run() {
-
-    Map<String, StringBuilder> stringBufferMap = new HashMap<>();
-    List<AppendOutputBuffer> list = new LinkedList<>();
+  public void run(List<AppendOutputBuffer> batch) {
+    run(batch, () -> true);
+  }
 
-    queue.drainTo(list);
-    if (list.isEmpty()) {
+  /** Stops between append groups when delivery is disallowed; an active callback may finish. */
+  void run(List<AppendOutputBuffer> batch, BooleanSupplier mayDeliver) {
+    if (batch.isEmpty()) {
       return;
     }
-    Long processingStartTime = System.currentTimeMillis();
-
-    Long sizeProcessed = Long.valueOf(0);
-    for (AppendOutputBuffer buffer : list) {
-      if (buffer instanceof UpdateOutputBuffer) {
-        sizeProcessed += flushAppendBuffers(stringBufferMap);
-        UpdateOutputBuffer update = (UpdateOutputBuffer) buffer;
-        try {
-          listener.onOutputUpdated(update.getNoteId(), update.getParagraphId(), update.getIndex(),
-              update.getType(), update.getData());
-        } catch (RuntimeException e) {
-          // A stale callback must not abort another paragraph's synchronous drain.
-          LOGGER.warn("Failed to update output for note {} paragraph {}",
-              update.getNoteId(), update.getParagraphId(), e);
-        }
-        continue;
+    long start = System.currentTimeMillis();
+    long size = 0;
+    Map<ParagraphOutputKey, StringBuilder> groups = new LinkedHashMap<>();
+    ParagraphOutputKey currentKey = null;
+    StringBuilder currentData = null;
+    for (AppendOutputBuffer append : batch) {
+      if (currentKey == null || !currentKey.matches(append)) {
+        currentKey = new ParagraphOutputKey(append);
+        currentData = groups.computeIfAbsent(currentKey, key -> new StringBuilder());
       }
-
-      String noteId = buffer.getNoteId();
-      String paragraphId = buffer.getParagraphId();
-      int index = buffer.getIndex();
-      String stringBufferKey = noteId + ":" + paragraphId + ":" + index;
-
-      StringBuilder builder = stringBufferMap.containsKey(stringBufferKey) ?
-          stringBufferMap.get(stringBufferKey) : new StringBuilder();
-
-      builder.append(buffer.getData());
-      stringBufferMap.put(stringBufferKey, builder);
+      currentData.append(append.getData());
     }
-    sizeProcessed += flushAppendBuffers(stringBufferMap);
-    Long processingTime = System.currentTimeMillis() - processingStartTime;
-
-    if (processingTime > SAFE_PROCESSING_TIME) {
-      LOGGER.warn("Processing time for buffered append-output is high: {} milliseconds.", processingTime);
+    for (Map.Entry<ParagraphOutputKey, StringBuilder> group : groups.entrySet()) {
+      if (!mayDeliver.getAsBoolean()) {
+        return;
+      }
+      size += flush(group.getKey(), group.getValue());
+    }
+    long time = System.currentTimeMillis() - start;
+    if (time > SAFE_PROCESSING_TIME) {
+      LOGGER.warn("Processing time for buffered append-output is high: {} milliseconds.", time);
     } else {
-      LOGGER.debug("Processing time for append-output took {} milliseconds", processingTime);
+      LOGGER.debug("Processing time for append-output took {} milliseconds", time);
     }
-
-    if (sizeProcessed > SAFE_PROCESSING_STRING_SIZE) {
-      LOGGER.warn("Processing size for buffered append-output is high: {} characters.", sizeProcessed);
+    if (size > SAFE_PROCESSING_STRING_SIZE) {
+      LOGGER.warn("Processing size for buffered append-output is high: {} characters.", size);
     } else {
-      LOGGER.debug("Processing size for append-output is {} characters", sizeProcessed);
+      LOGGER.debug("Processing size for append-output is {} characters", size);
     }
   }
 
-  private long flushAppendBuffers(Map<String, StringBuilder> stringBufferMap) {
-    long sizeProcessed = 0;
-    for (Entry<String, StringBuilder> stringBufferMapEntry : stringBufferMap.entrySet()) {
-      String stringBufferKey = stringBufferMapEntry.getKey();
-      StringBuilder buffer = stringBufferMapEntry.getValue();
-      sizeProcessed += buffer.length();
-      try {
-        String[] keys = stringBufferKey.split(":");
-        listener.onOutputAppend(keys[0], keys[1], Integer.parseInt(keys[2]), buffer.toString());
-      } catch (RuntimeException e) {
-        // One stale append must not abort another paragraph's synchronous drain.
-        LOGGER.warn("Failed to append output for {}", stringBufferKey, e);
-      }
+  private long flush(ParagraphOutputKey key, StringBuilder data) {
+    long size = data.length();
+    try {
+      listener.onOutputAppend(key.noteId, key.paragraphId, key.index, data.toString());
+    } catch (RuntimeException e) {
+      // A stale paragraph must not abort delivery of later output in this drain.
+      LOGGER.warn("Failed to append output for note {} paragraph {}",
+          key.noteId, key.paragraphId, e);
     }
-    stringBufferMap.clear();
-    return sizeProcessed;
+    data.setLength(0);
+    return size;
   }
 
-  public void appendBuffer(String noteId, String paragraphId, int index, String outputToAppend) {
-    queue.offer(new AppendOutputBuffer(noteId, paragraphId, index, outputToAppend));
-  }
+  private static final class ParagraphOutputKey {
+    private final String noteId;
+    private final String paragraphId;
+    private final int index;
+
+    private ParagraphOutputKey(AppendOutputBuffer append) {
+      noteId = append.getNoteId();
+      paragraphId = append.getParagraphId();
+      index = append.getIndex();
+    }
+
+    private boolean matches(AppendOutputBuffer append) {
+      return index == append.getIndex()
+          && Objects.equals(noteId, append.getNoteId())
+          && Objects.equals(paragraphId, append.getParagraphId());
+    }
 
-  /** Enqueues a replacement; callers needing completion must also invoke run(). */
-  public void updateBuffer(String noteId, String paragraphId, int index,
-                           InterpreterResult.Type type, String output) {
-    queue.offer(new UpdateOutputBuffer(noteId, paragraphId, index, type, output));
+    @Override
+    public boolean equals(Object other) {
+      if (this == other) {
+        return true;
+      }
+      if (!(other instanceof ParagraphOutputKey)) {
+        return false;
+      }
+      ParagraphOutputKey key = (ParagraphOutputKey) other;
+      return index == key.index && Objects.equals(noteId, key.noteId)
+          && Objects.equals(paragraphId, key.paragraphId);
+    }
+
+    @Override
+    public int hashCode() {
+      int hash = Objects.hashCode(noteId);
+      hash = 31 * hash + Objects.hashCode(paragraphId);
+      return 31 * hash + index;
+    }
   }
 }