From 425018b5e60335fa6051b0c313baa214b9712956 Mon Sep 17 00:00:00 2001 From: miinhho Date: Wed, 16 Sep 2026 03:38:19 +0900 Subject: [PATCH 1/4] [ZEPPELIN-6709] Dispatch paragraph output concurrently across notes --- conf/zeppelin-site.xml.template | 15 + .../zeppelin/conf/ZeppelinConfiguration.java | 3 + .../RemoteInterpreterEventServer.java | 68 ++- .../remote/AppendOutputRunner.java | 133 ++--- .../remote/ParagraphOutputDispatcher.java | 324 +++++++++++ .../remote/UpdateOutputBuffer.java | 41 -- .../RemoteInterpreterEventServerTest.java | 408 +++++++++++-- .../remote/AppendOutputRunnerTest.java | 377 ++++-------- .../remote/ParagraphOutputDispatcherTest.java | 536 ++++++++++++++++++ 9 files changed, 1447 insertions(+), 458 deletions(-) create mode 100644 zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcher.java delete mode 100644 zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java create mode 100644 zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcherTest.java diff --git a/conf/zeppelin-site.xml.template b/conf/zeppelin-site.xml.template index 4107a6799cf..188adb26acd 100755 --- a/conf/zeppelin-site.xml.template +++ b/conf/zeppelin-site.xml.template @@ -439,6 +439,21 @@ Output message from interpreter exceeding the limit will be truncated + + zeppelin.interpreter.output.worker.count + 4 + 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. + + + + zeppelin.interpreter.output.events.per.batch + 1000 + Maximum output events processed for one note before yielding to other ready notes. + This controls scheduling granularity, not queue capacity or output truncation. + + zeppelin.ssl false diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java b/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java index 01ad388ebac..dca1482a8a6 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java +++ b/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", ""), diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java index c8f8886e6a9..38c362aafec 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java +++ b/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 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> 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 diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java index 8bc063af106..3007172dca9 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java @@ -17,118 +17,87 @@ 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.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 adjacent events for the same 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 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 stringBufferMap = new HashMap<>(); - List list = new LinkedList<>(); + public void run(List 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 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); + long start = System.currentTimeMillis(); + long size = 0; + AppendOutputBuffer previous = null; + StringBuilder data = new StringBuilder(); + for (AppendOutputBuffer append : batch) { + if (!mayDeliver.getAsBoolean()) { + return; + } + if (previous != null && !sameOutput(previous, append)) { + size += flush(previous, data); + if (!mayDeliver.getAsBoolean()) { + return; } - continue; } - - 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); + previous = append; + data.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); - } else { - LOGGER.debug("Processing time for append-output took {} milliseconds", processingTime); + if (!mayDeliver.getAsBoolean()) { + return; } - - if (sizeProcessed > SAFE_PROCESSING_STRING_SIZE) { - LOGGER.warn("Processing size for buffered append-output is high: {} characters.", sizeProcessed); + size += flush(previous, data); + 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 size for append-output is {} characters", sizeProcessed); + LOGGER.debug("Processing time for append-output took {} milliseconds", time); } - } - - private long flushAppendBuffers(Map stringBufferMap) { - long sizeProcessed = 0; - for (Entry 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); - } + 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", size); } - stringBufferMap.clear(); - return sizeProcessed; } - public void appendBuffer(String noteId, String paragraphId, int index, String outputToAppend) { - queue.offer(new AppendOutputBuffer(noteId, paragraphId, index, outputToAppend)); + private boolean sameOutput(AppendOutputBuffer first, AppendOutputBuffer second) { + return first.getIndex() == second.getIndex() + && Objects.equals(first.getNoteId(), second.getNoteId()) + && Objects.equals(first.getParagraphId(), second.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)); + private long flush(AppendOutputBuffer append, StringBuilder data) { + long size = data.length(); + try { + listener.onOutputAppend(append.getNoteId(), append.getParagraphId(), append.getIndex(), + 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 {}", + append.getNoteId(), append.getParagraphId(), e); + } + data.setLength(0); + return size; } } diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcher.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcher.java new file mode 100644 index 00000000000..70535b88bd0 --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcher.java @@ -0,0 +1,324 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.zeppelin.interpreter.remote; + +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 org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import org.apache.zeppelin.interpreter.InterpreterResult; +import org.apache.zeppelin.interpreter.InterpreterResultMessage; + +/** + * Buffers output in per-note FIFO queues, consumed by a fixed pool of workers. Only one worker + * owns a note at a time. Callers periodically invoke {@link #flush()} to deliver buffered appends; + * output boundaries request immediate delivery and expose callback completion to the caller. + * Queues are retired as soon as their pending and in-flight output has been processed. + * + *

Accepted boundaries cannot be cancelled. Their completion acknowledges callback delivery, + * not note idleness. RPC callers wait outside the queue monitor. + */ +public class ParagraphOutputDispatcher implements AutoCloseable { + + private static final Logger LOGGER = LoggerFactory.getLogger(ParagraphOutputDispatcher.class); + private static final long SAFE_PROCESSING_TIME_MS = 10; + + // Guarded by this: notes, mutable NoteQueue state, and worker startup. + private final Map notes = new HashMap<>(); + private final BlockingQueue ready = new LinkedBlockingQueue<>(); + private final int eventsPerBatch; + private final ExecutorService workers; + private final int workerCount; + private final RemoteInterpreterProcessListener listener; + private final AppendOutputRunner appendRunner; + private volatile boolean closed; + private boolean workersStarted; + + public ParagraphOutputDispatcher(RemoteInterpreterProcessListener listener) { + this(listener, ZEPPELIN_INTERPRETER_OUTPUT_WORKER_COUNT.getIntValue(), + ZEPPELIN_INTERPRETER_OUTPUT_EVENTS_PER_BATCH.getIntValue()); + } + + ParagraphOutputDispatcher(RemoteInterpreterProcessListener listener, int workerCount) { + this(listener, workerCount, ZEPPELIN_INTERPRETER_OUTPUT_EVENTS_PER_BATCH.getIntValue()); + } + + public ParagraphOutputDispatcher(RemoteInterpreterProcessListener listener, int workerCount, + int eventsPerBatch) { + if (workerCount < 1) { + throw new IllegalArgumentException("Output worker count must be positive"); + } + if (eventsPerBatch < 1) { + throw new IllegalArgumentException("Output events per batch must be positive"); + } + this.eventsPerBatch = eventsPerBatch; + this.listener = listener; + appendRunner = new AppendOutputRunner(listener); + workers = Executors.newFixedThreadPool(workerCount, runnable -> { + Thread thread = new Thread(runnable, "zeppelin-output-worker"); + thread.setDaemon(true); + return thread; + }); + this.workerCount = workerCount; + } + + /** Makes pending notes ready without waiting for any listener callback. */ + public synchronized void flush() { + if (!closed) { + for (NoteQueue note : notes.values()) { + makeReady(note); + } + } + } + + public void appendOutput(String noteId, String paragraphId, int index, String output) { + enqueue(noteId, new OutputEvent(new AppendOutputBuffer(noteId, paragraphId, index, output)), + false); + } + + public Future updateOutput(String noteId, String paragraphId, int index, + InterpreterResult.Type type, String output) { + return enqueueBoundary(noteId, + () -> listener.onOutputUpdated(noteId, paragraphId, index, type, output)); + } + + public Future updateAllOutput(String noteId, String paragraphId, + List messages) { + // The caller may change its list before this queued operation gets a worker. + List replacements = new ArrayList<>(messages); + return enqueueBoundary(noteId, () -> { + // Clear and replacements must stay together; later appends belong to the replaced output. + listener.onOutputClear(noteId, paragraphId); + for (int i = 0; i < replacements.size(); i++) { + InterpreterResultMessage message = replacements.get(i); + listener.onOutputUpdated(noteId, paragraphId, i, message.getType(), message.getData()); + } + }); + } + + public Future checkpointOutput(String noteId, String paragraphId) { + return enqueueBoundary(noteId, () -> listener.checkpointOutput(noteId, paragraphId)); + } + + private Future enqueueBoundary(String noteId, Runnable callback) { + OutputEvent event = new OutputEvent(new Boundary(() -> { + long start = System.nanoTime(); + try { + callback.run(); + } catch (RuntimeException e) { + LOGGER.warn("Failed to process output boundary for note {}", noteId, e); + throw e; + } finally { + long time = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + if (time > SAFE_PROCESSING_TIME_MS) { + LOGGER.warn("Processing output boundary for note {} took {} milliseconds", noteId, time); + } else { + LOGGER.debug("Processing output boundary for note {} took {} milliseconds", noteId, time); + } + } + })); + enqueue(noteId, event, true); + return event.boundary; + } + + private synchronized void enqueue(String noteId, OutputEvent event, boolean immediate) { + if (closed) { + throw new IllegalStateException("Output dispatcher is stopped"); + } + if (!workersStarted) { + workersStarted = true; + for (int i = 0; i < workerCount; i++) { + workers.execute(this::consume); + } + } + NoteQueue note = notes.computeIfAbsent(noteId, NoteQueue::new); + note.events.addLast(event); + if (immediate) { + makeReady(note); + } + } + + // Caller must hold this monitor: the scheduled check and ready insertion must be atomic. + private void makeReady(NoteQueue note) { + note.flushRequested = true; + if (!note.scheduled) { + note.scheduled = true; + ready.offer(note); + } + } + + private void consume() { + while (!closed && !Thread.currentThread().isInterrupted()) { + NoteQueue note; + try { + note = ready.take(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + List batch = new ArrayList<>(); + synchronized (this) { + if (closed) { + return; + } + note.flushRequested = false; + while (!note.events.isEmpty() && batch.size() < eventsPerBatch) { + batch.add(note.events.removeFirst()); + } + note.inFlight = batch; + } + try { + deliver(batch); + } finally { + synchronized (this) { + note.inFlight = null; + note.scheduled = false; + if (!closed) { + if (note.events.isEmpty()) { + notes.remove(note.noteId); + } else if (note.flushRequested || batch.size() == eventsPerBatch) { + makeReady(note); + } + } + } + } + } + } + + private void deliver(List batch) { + List appends = new ArrayList<>(); + for (OutputEvent event : batch) { + if (closed) { + return; + } + if (event.append != null) { + appends.add(event.append); + } else { + appendRunner.run(appends, () -> !closed); + appends.clear(); + if (closed) { + return; + } + event.boundary.run(); + } + } + if (!closed) { + appendRunner.run(appends, () -> !closed); + } + } + + /** + * Stops accepting output and fails unfinished boundaries. Remaining output is discarded when + * workers observe shutdown; in-flight listener calls may finish. Does not wait for worker exit. + */ + @Override + public void close() { + synchronized (this) { + if (closed) { + return; + } + closed = true; + IllegalStateException stopped = new IllegalStateException("Output dispatcher is stopped"); + for (NoteQueue note : notes.values()) { + failBoundaries(note.events, stopped); + if (note.inFlight != null) { + failBoundaries(note.inFlight, stopped); + } + note.events.clear(); + } + notes.clear(); + ready.clear(); + } + workers.shutdownNow(); + } + + private void failBoundaries(Iterable events, IllegalStateException stopped) { + for (OutputEvent event : events) { + if (event.boundary != null) { + event.boundary.fail(stopped); + } + } + } + + // Retained note queues, not a worker termination signal. + synchronized int pendingNoteCount() { + return notes.size(); + } + + private static class NoteQueue { + private final String noteId; + private final ArrayDeque events = new ArrayDeque<>(); + // Shutdown must release RPCs waiting on boundaries already drained from events. + private List inFlight; + // Covers ready and running: releasing ownership before delivery ends would allow two writers. + private boolean scheduled; + // A request arriving during delivery must survive until the current owner releases the note. + private boolean flushRequested; + + private NoteQueue(String noteId) { + this.noteId = noteId; + } + } + + // Exactly one of append and boundary is set. + private static class OutputEvent { + private final AppendOutputBuffer append; + private final Boundary boundary; + + private OutputEvent(AppendOutputBuffer append) { + this.append = append; + boundary = null; + } + + private OutputEvent(Boundary boundary) { + append = null; + this.boundary = boundary; + } + } + + // FutureTask wakes get() waiters without running externally supplied completion handlers. + private static final class Boundary extends FutureTask { + private Boundary(Runnable callback) { + super(callback, null); + } + + private void fail(Throwable failure) { + setException(failure); + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + // Accepted output must stay in the FIFO even if its RPC caller stops waiting. + return false; + } + } +} diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java deleted file mode 100644 index 15de2d5092b..00000000000 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java +++ /dev/null @@ -1,41 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.zeppelin.interpreter.remote; - -import org.apache.zeppelin.interpreter.InterpreterResult; - -/** - * This element stores the buffered update-data of paragraph's output. It shares the - * append-data queue so that an update, which replaces a result, can never be sent - * ahead of the appends that preceded it. - */ -public class UpdateOutputBuffer extends AppendOutputBuffer { - - private final InterpreterResult.Type type; - - public UpdateOutputBuffer(String noteId, String paragraphId, int index, - InterpreterResult.Type type, String data) { - super(noteId, paragraphId, index, data); - this.type = type; - } - - public InterpreterResult.Type getType() { - return type; - } - -} diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java index db73b10a427..035f06137e5 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java @@ -16,32 +16,42 @@ */ package org.apache.zeppelin.interpreter; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; - +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.InOrder; import java.io.IOException; import java.io.ObjectOutputStream; import java.io.Serializable; import java.lang.reflect.Field; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Collections; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; - +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars; import org.apache.zeppelin.conf.ZeppelinConfiguration; -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.RemoteInterpreterProcess; import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcessListener; import org.apache.zeppelin.interpreter.thrift.InterpreterRPCException; @@ -51,8 +61,6 @@ import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterResultMessage; import org.apache.zeppelin.resource.Resource; import org.apache.zeppelin.resource.ResourceId; -import org.junit.jupiter.api.Test; -import org.mockito.InOrder; public class RemoteInterpreterEventServerTest { @@ -60,12 +68,20 @@ public class RemoteInterpreterEventServerTest { @Test void updateOutputCompletesBeforeReturning() throws Exception { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); - RemoteInterpreterEventServer server = - serverWithRunner(listener, new AppendOutputRunner(listener)); + RemoteInterpreterEventServer server = serverWithListener(listener); try { + server.appendOutput(new OutputAppendEvent("note", "para", 0, "before", null)); server.updateOutput(new OutputUpdateEvent("note", "para", 0, "TEXT", "final", null)); - // A caller may publish terminal status as soon as the RPC returns. verify(listener).onOutputUpdated("note", "para", 0, InterpreterResult.Type.TEXT, "final"); + server.appendOutput(new OutputAppendEvent("note", "para", 0, "after", null)); + server.checkpointOutput("note", "para"); + InOrder order = inOrder(listener); + order.verify(listener).onOutputAppend("note", "para", 0, "before"); + order.verify(listener).onOutputUpdated( + "note", "para", 0, InterpreterResult.Type.TEXT, "final"); + order.verify(listener).onOutputAppend("note", "para", 0, "after"); + order.verify(listener).checkpointOutput("note", "para"); + order.verifyNoMoreInteractions(); } finally { server.stop(); } @@ -74,8 +90,7 @@ void updateOutputCompletesBeforeReturning() throws Exception { @Test void checkpointDrainsPendingOutputBeforeSaving() throws Exception { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); - RemoteInterpreterEventServer server = - serverWithRunner(listener, new AppendOutputRunner(listener)); + RemoteInterpreterEventServer server = serverWithListener(listener); try { server.appendOutput(new OutputAppendEvent("note", "para", 0, "pending", null)); server.checkpointOutput("note", "para"); @@ -90,85 +105,373 @@ void checkpointDrainsPendingOutputBeforeSaving() throws Exception { @Test void updateAllIsAnOrderedClearAndReplacement() throws Exception { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); - AppendOutputRunner runner = new AppendOutputRunner(listener); - RemoteInterpreterEventServer server = serverWithRunner(listener, runner); + RemoteInterpreterEventServer server = serverWithListener(listener); try { - runner.appendBuffer("note", "para", 0, "old"); - server.updateAllOutput(new OutputUpdateAllEvent("note", "para", Collections.singletonList( - new RemoteInterpreterResultMessage("HTML", "replacement")))); - verify(listener).onOutputUpdated("note", "para", 0, - InterpreterResult.Type.HTML, "replacement"); - runner.appendBuffer("note", "para", 0, "new"); - runner.run(); + server.appendOutput(new OutputAppendEvent("note", "para", 0, "old", null)); + server.updateAllOutput(new OutputUpdateAllEvent("note", "para", Arrays.asList( + new RemoteInterpreterResultMessage("HTML", "replacement"), + new RemoteInterpreterResultMessage("TEXT", "second")))); + verify(listener).onOutputUpdated("note", "para", 1, InterpreterResult.Type.TEXT, "second"); + server.appendOutput(new OutputAppendEvent("note", "para", 1, "new", null)); + server.checkpointOutput("note", "para"); InOrder order = inOrder(listener); order.verify(listener).onOutputAppend("note", "para", 0, "old"); order.verify(listener).onOutputClear("note", "para"); order.verify(listener).onOutputUpdated("note", "para", 0, InterpreterResult.Type.HTML, "replacement"); - order.verify(listener).onOutputAppend("note", "para", 0, "new"); + order.verify(listener).onOutputUpdated("note", "para", 1, + InterpreterResult.Type.TEXT, "second"); + order.verify(listener).onOutputAppend("note", "para", 1, "new"); + order.verify(listener).checkpointOutput("note", "para"); + order.verifyNoMoreInteractions(); } finally { server.stop(); } } @Test - void updateAllWaitsForInFlightAppendAndCompletesBeforeReturning() throws Exception { + void emptyUpdateAllStillClearsPendingOutput() throws Exception { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); - AppendOutputRunner runner = new AppendOutputRunner(listener); - RemoteInterpreterEventServer server = serverWithRunner(listener, runner); + RemoteInterpreterEventServer server = serverWithListener(listener); + try { + server.appendOutput(new OutputAppendEvent("note", "para", 0, "old", null)); + server.updateAllOutput(new OutputUpdateAllEvent("note", "para", Collections.emptyList())); + InOrder order = inOrder(listener); + order.verify(listener).onOutputAppend("note", "para", 0, "old"); + order.verify(listener).onOutputClear("note", "para"); + order.verifyNoMoreInteractions(); + } finally { + server.stop(); + } + } + + @ParameterizedTest + @ValueSource(strings = {"UPDATE", "UPDATE_ALL", "CHECKPOINT"}) + void sameNoteBoundaryWaitsForInFlightAppend(String operation) throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + RemoteInterpreterEventServer server = serverWithListener(listener); CountDownLatch entered = new CountDownLatch(1); CountDownLatch release = new CountDownLatch(1); - CountDownLatch updateStarted = new CountDownLatch(1); + CountDownLatch started = new CountDownLatch(1); doAnswer(invocation -> { entered.countDown(); assertTrue(release.await(5, TimeUnit.SECONDS)); return null; - }).when(listener).onOutputAppend("note", "para", 0, "old"); - ExecutorService executor = Executors.newFixedThreadPool(2); + }).when(listener).onOutputAppend("note", "first", 0, "old"); + ExecutorService callers = Executors.newSingleThreadExecutor(); try { - runner.appendBuffer("note", "para", 0, "old"); - Future first = executor.submit(runner); + server.appendOutput(new OutputAppendEvent("note", "first", 0, "old", null)); + dispatcherOf(server).flush(); assertTrue(entered.await(5, TimeUnit.SECONDS)); - Future update = executor.submit(() -> { - updateStarted.countDown(); - server.updateAllOutput(new OutputUpdateAllEvent("note", "para", Collections.singletonList( - new RemoteInterpreterResultMessage("HTML", "replacement")))); - verify(listener).onOutputUpdated("note", "para", 0, - InterpreterResult.Type.HTML, "replacement"); + Future boundary = callers.submit(() -> { + started.countDown(); + callBoundary(server, "note", "second", operation); return null; }); - assertTrue(updateStarted.await(5, TimeUnit.SECONDS)); - assertThrows(TimeoutException.class, () -> update.get(100, TimeUnit.MILLISECONDS)); + assertTrue(started.await(5, TimeUnit.SECONDS)); + assertThrows(TimeoutException.class, () -> boundary.get(100, TimeUnit.MILLISECONDS)); release.countDown(); - first.get(5, TimeUnit.SECONDS); - update.get(5, TimeUnit.SECONDS); + boundary.get(5, TimeUnit.SECONDS); InOrder order = inOrder(listener); - order.verify(listener).onOutputAppend("note", "para", 0, "old"); + order.verify(listener).onOutputAppend("note", "first", 0, "old"); + if ("UPDATE".equals(operation)) { + order.verify(listener).onOutputUpdated( + "note", "second", 0, InterpreterResult.Type.TEXT, "replacement"); + } else if ("UPDATE_ALL".equals(operation)) { + order.verify(listener).onOutputClear("note", "second"); + order.verify(listener).onOutputUpdated( + "note", "second", 0, InterpreterResult.Type.TEXT, "replacement"); + } else { + order.verify(listener).checkpointOutput("note", "second"); + } + } finally { + release.countDown(); + callers.shutdownNow(); + server.stop(); + } + } + + @Test + void independentNoteMakesProgressWhileAnotherNoteAppendIsBlocked() throws Exception { + // These note IDs have identical hash codes, so concurrency cannot depend on hash lanes. + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + RemoteInterpreterEventServer server = serverWithListener(listener); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + CountDownLatch otherAppend = new CountDownLatch(1); + doAnswer(call -> { + entered.countDown(); + assertTrue(release.await(5, TimeUnit.SECONDS)); + return null; + }).when(listener).onOutputAppend("Aa", "para", 0, "blocked"); + doAnswer(call -> { + otherAppend.countDown(); + return null; + }).when(listener).onOutputAppend("BB", "para", 0, "periodic"); + ExecutorService callers = Executors.newSingleThreadExecutor(); + try { + server.appendOutput(new OutputAppendEvent("Aa", "para", 0, "blocked", null)); + dispatcherOf(server).flush(); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + server.appendOutput(new OutputAppendEvent("BB", "para", 0, "periodic", null)); + dispatcherOf(server).flush(); + assertTrue(otherAppend.await(5, TimeUnit.SECONDS)); + Future independent = callers.submit(() -> { + callBoundary(server, "BB", "para", "UPDATE"); + callBoundary(server, "BB", "para", "UPDATE_ALL"); + callBoundary(server, "BB", "para", "CHECKPOINT"); + return null; + }); + independent.get(5, TimeUnit.SECONDS); + verify(listener).onOutputClear("BB", "para"); + verify(listener).checkpointOutput("BB", "para"); + } finally { + release.countDown(); + callers.shutdownNow(); + server.stop(); + } + } + + @ParameterizedTest + @ValueSource(strings = {"UPDATE", "UPDATE_ALL", "CHECKPOINT"}) + void laterAppendCannotOvertakeBoundaryCallback(String operation) throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + RemoteInterpreterEventServer server = serverWithListener(listener); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicBoolean boundaryCompleted = new AtomicBoolean(); + AtomicBoolean appendOverlappedBoundary = new AtomicBoolean(); + doAnswer(call -> { + entered.countDown(); + assertTrue(release.await(5, TimeUnit.SECONDS)); + return null; + }).when(listener).onOutputClear("note", "para"); + doAnswer(call -> { + entered.countDown(); + assertTrue(release.await(5, TimeUnit.SECONDS)); + boundaryCompleted.set(true); + return null; + }).when(listener).checkpointOutput("note", "para"); + doAnswer(call -> { + entered.countDown(); + assertTrue(release.await(5, TimeUnit.SECONDS)); + boundaryCompleted.set(true); + return null; + }).when(listener).onOutputUpdated( + "note", "para", 0, InterpreterResult.Type.TEXT, "replacement"); + doAnswer(call -> { + if (!boundaryCompleted.get()) { + appendOverlappedBoundary.set(true); + } + return null; + }).when(listener).onOutputAppend("note", "para", 0, "later"); + ExecutorService callers = Executors.newSingleThreadExecutor(); + try { + Future boundary = callers.submit(() -> { + callBoundary(server, "note", "para", operation); + return null; + }); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + server.appendOutput(new OutputAppendEvent("note", "para", 0, "later", null)); + dispatcherOf(server).flush(); + verify(listener, never()).onOutputAppend("note", "para", 0, "later"); + assertThrows(TimeoutException.class, () -> boundary.get(100, TimeUnit.MILLISECONDS)); + release.countDown(); + boundary.get(5, TimeUnit.SECONDS); + server.checkpointOutput("note", "drained"); + assertTrue(boundaryCompleted.get()); + assertFalse(appendOverlappedBoundary.get(), "Append must wait for the entire boundary"); + InOrder order = inOrder(listener); + if ("UPDATE".equals(operation)) { + order.verify(listener).onOutputUpdated( + "note", "para", 0, InterpreterResult.Type.TEXT, "replacement"); + } else if ("UPDATE_ALL".equals(operation)) { + order.verify(listener).onOutputClear("note", "para"); + order.verify(listener).onOutputUpdated( + "note", "para", 0, InterpreterResult.Type.TEXT, "replacement"); + } else { + order.verify(listener).checkpointOutput("note", "para"); + } + order.verify(listener).onOutputAppend("note", "para", 0, "later"); + } finally { + release.countDown(); + callers.shutdownNow(); + server.stop(); + } + } + + @ParameterizedTest + @ValueSource(strings = {"UPDATE", "UPDATE_ALL", "CHECKPOINT"}) + void boundaryCallbackFailureBecomesRpcExceptionAndLaterOutputStillWorks(String operation) + throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + IllegalStateException removed = new IllegalStateException("paragraph removed"); + doThrow(removed).when(listener).onOutputUpdated( + "note", "gone", 0, InterpreterResult.Type.TEXT, "replacement"); + doThrow(removed).when(listener).onOutputClear("note", "gone"); + doThrow(removed).when(listener).checkpointOutput("note", "gone"); + RemoteInterpreterEventServer server = serverWithListener(listener); + try { + InterpreterRPCException failure = assertThrows(InterpreterRPCException.class, + () -> callBoundary(server, "note", "gone", operation)); + assertTrue(failure.getErrorMessage().contains("paragraph removed")); + server.appendOutput(new OutputAppendEvent("note", "present", 0, "good", null)); + server.checkpointOutput("note", "present"); + verify(listener).onOutputAppend("note", "present", 0, "good"); + } finally { + server.stop(); + } + } + + @Test + void largeParagraphOutputIsDeliveredWithoutAnAdditionalDispatcherLimit() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + RemoteInterpreterEventServer server = serverWithListener(listener); + String output = "a".repeat(4 * 1024 * 1024 + 1); + try { + server.appendOutput(new OutputAppendEvent("note", "para", 0, output, null)); + server.updateOutput(new OutputUpdateEvent("note", "para", 0, "TEXT", output, null)); + server.updateAllOutput(new OutputUpdateAllEvent("note", "para", List.of( + new RemoteInterpreterResultMessage("HTML", output)))); + server.checkpointOutput("note", "para"); + InOrder order = inOrder(listener); + order.verify(listener).onOutputAppend("note", "para", 0, output); + order.verify(listener).onOutputUpdated( + "note", "para", 0, InterpreterResult.Type.TEXT, output); order.verify(listener).onOutputClear("note", "para"); - order.verify(listener).onOutputUpdated("note", "para", 0, - InterpreterResult.Type.HTML, "replacement"); + order.verify(listener).onOutputUpdated( + "note", "para", 0, InterpreterResult.Type.HTML, output); + order.verify(listener).checkpointOutput("note", "para"); + order.verifyNoMoreInteractions(); + } finally { + server.stop(); + } + } + + @Test + void serverAppliesConfiguredOutputBatchSize() throws Exception { + ZeppelinConfiguration zConf = ZeppelinConfiguration.load("zeppelin-test-site.xml"); + zConf.setProperty(ConfVars.ZEPPELIN_INTERPRETER_OUTPUT_WORKER_COUNT.getVarName(), "1"); + zConf.setProperty(ConfVars.ZEPPELIN_INTERPRETER_OUTPUT_EVENTS_PER_BATCH.getVarName(), "2"); + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + RemoteInterpreterEventServer server = serverWithListener(listener, zConf); + try { + for (String output : List.of("a", "b", "c", "d")) { + server.appendOutput(new OutputAppendEvent("note", "para", 0, output, null)); + } + server.checkpointOutput("note", "para"); + InOrder order = inOrder(listener); + order.verify(listener).onOutputAppend("note", "para", 0, "ab"); + order.verify(listener).onOutputAppend("note", "para", 0, "cd"); + order.verify(listener).checkpointOutput("note", "para"); + order.verifyNoMoreInteractions(); } finally { + server.stop(); + } + } + + @Test + void stoppedServerRejectsOutputWithRpcExceptions() { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + RemoteInterpreterEventServer server = serverWithListener(listener); + server.stop(); + assertThrows(InterpreterRPCException.class, () -> + server.appendOutput(new OutputAppendEvent("note", "para", 0, "late", null))); + for (String operation : Arrays.asList("UPDATE", "UPDATE_ALL", "CHECKPOINT")) { + assertThrows(InterpreterRPCException.class, + () -> callBoundary(server, "note", "para", operation)); + } + verify(listener, never()).onOutputAppend("note", "para", 0, "late"); + } + + @Test + void interruptedRpcWaitPreservesInterruptAndDoesNotCancelAcceptedOutput() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + doAnswer(call -> { + entered.countDown(); + assertTrue(release.await(5, TimeUnit.SECONDS)); + return null; + }).when(listener).checkpointOutput("note", "para"); + RemoteInterpreterEventServer server = serverWithListener(listener); + AtomicBoolean interruptPreserved = new AtomicBoolean(); + AtomicReference failure = new AtomicReference<>(); + Thread caller = new Thread(() -> { + try { + server.checkpointOutput("note", "para"); + } catch (Throwable e) { + failure.set(e); + interruptPreserved.set(Thread.currentThread().isInterrupted()); + } + }); + try { + caller.start(); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + caller.interrupt(); + caller.join(5000); + assertFalse(caller.isAlive()); + assertTrue(failure.get() instanceof InterpreterRPCException); + assertTrue(interruptPreserved.get()); release.countDown(); - executor.shutdownNow(); + server.appendOutput(new OutputAppendEvent("note", "para", 0, "after", null)); + server.checkpointOutput("note", "done"); + InOrder order = inOrder(listener); + order.verify(listener).checkpointOutput("note", "para"); + order.verify(listener).onOutputAppend("note", "para", 0, "after"); + order.verify(listener).checkpointOutput("note", "done"); + } finally { + release.countDown(); + caller.interrupt(); + caller.join(5000); server.stop(); } } - private RemoteInterpreterEventServer serverWithRunner( - RemoteInterpreterProcessListener listener, AppendOutputRunner runner) throws Exception { + private void callBoundary(RemoteInterpreterEventServer server, String noteId, + String paragraphId, String operation) throws Exception { + if ("UPDATE".equals(operation)) { + server.updateOutput(new OutputUpdateEvent( + noteId, paragraphId, 0, "TEXT", "replacement", null)); + } else if ("UPDATE_ALL".equals(operation)) { + server.updateAllOutput(new OutputUpdateAllEvent(noteId, paragraphId, + Collections.singletonList(new RemoteInterpreterResultMessage("TEXT", "replacement")))); + } else { + server.checkpointOutput(noteId, paragraphId); + } + } + + private RemoteInterpreterEventServer serverWithListener( + RemoteInterpreterProcessListener listener) { + return serverWithListener(listener, outputConfiguration()); + } + + private RemoteInterpreterEventServer serverWithListener( + RemoteInterpreterProcessListener listener, ZeppelinConfiguration zConf) { InterpreterSettingManager manager = mock(InterpreterSettingManager.class); when(manager.getRemoteInterpreterProcessListener()).thenReturn(listener); - RemoteInterpreterEventServer server = new RemoteInterpreterEventServer( - mock(ZeppelinConfiguration.class), manager); - Field field = RemoteInterpreterEventServer.class.getDeclaredField("runner"); + return new RemoteInterpreterEventServer(zConf, manager); + } + + private ZeppelinConfiguration outputConfiguration() { + ZeppelinConfiguration zConf = mock(ZeppelinConfiguration.class); + when(zConf.getInt(ConfVars.ZEPPELIN_INTERPRETER_OUTPUT_WORKER_COUNT)) + .thenReturn(ConfVars.ZEPPELIN_INTERPRETER_OUTPUT_WORKER_COUNT.getIntValue()); + when(zConf.getInt(ConfVars.ZEPPELIN_INTERPRETER_OUTPUT_EVENTS_PER_BATCH)) + .thenReturn(ConfVars.ZEPPELIN_INTERPRETER_OUTPUT_EVENTS_PER_BATCH.getIntValue()); + return zConf; + } + + private ParagraphOutputDispatcher dispatcherOf(RemoteInterpreterEventServer server) + throws Exception { + Field field = RemoteInterpreterEventServer.class.getDeclaredField("outputDispatcher"); field.setAccessible(true); - field.set(server, runner); - return server; + return (ParagraphOutputDispatcher) field.get(server); } @Test void invokeMethodThrowsRpcExceptionWhenSerializationFails() throws Exception { - ZeppelinConfiguration zConf = mock(ZeppelinConfiguration.class); + ZeppelinConfiguration zConf = outputConfiguration(); InterpreterSettingManager manager = mock(InterpreterSettingManager.class); RemoteInterpreterEventServer server = new RemoteInterpreterEventServer(zConf, manager); @@ -188,7 +491,8 @@ void invokeMethodThrowsRpcExceptionWhenSerializationFails() throws Exception { .callRemoteFunction(any()); ResourceId resourceId = ResourceId.fromJson( - "{\"resourcePoolId\":\"pool-id\",\"name\":\"resource-name\",\"noteId\":\"note-id\",\"paragraphId\":\"paragraph-id\"}" + "{\"resourcePoolId\":\"pool-id\",\"name\":\"resource-name\"," + + "\"noteId\":\"note-id\",\"paragraphId\":\"paragraph-id\"}" ); InvokeResourceMethodEventMessage message = new InvokeResourceMethodEventMessage( diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java index 2d6e08beba7..98b6f658b47 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java @@ -17,309 +17,164 @@ package org.apache.zeppelin.interpreter.remote; -import org.apache.zeppelin.interpreter.InterpreterResult; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import org.apache.log4j.AppenderSkeleton; import org.apache.log4j.Level; import org.apache.log4j.Logger; import org.apache.log4j.spi.LoggingEvent; -import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import org.mockito.InOrder; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; - import java.util.ArrayList; import java.util.List; -import java.time.Duration; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Future; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.TimeoutException; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; - -import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.junit.jupiter.api.Assertions.assertThrows; -import static org.mockito.Mockito.doThrow; -import static org.junit.jupiter.api.Assertions.fail; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyInt; -import static org.mockito.Mockito.atMost; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.inOrder; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; class AppendOutputRunnerTest { - - private static final int NUM_EVENTS = 10000; - private static final int NUM_CLUBBED_EVENTS = 100; - private static final ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(); - private static ScheduledFuture future = null; - /* It is being accessed by multiple threads. - * While loop for 'loopForBufferCompletion' could - * run for-ever. - */ - private volatile static int numInvocations = 0; - - @AfterEach - public void afterEach() { - if (future != null) { - future.cancel(true); - } - } - @Test - void testSingleEvent() throws InterruptedException { + void batchesAdjacentAppendsAndHandlesEmptyBatches() { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); - String[][] buffer = {{"note", "para", "data\n"}}; - - loopForCompletingEvents(listener, 1, buffer); - verify(listener, times(1)).onOutputAppend(any(String.class), any(String.class), anyInt(), any(String.class)); - verify(listener, times(1)).onOutputAppend("note", "para", 0, "data\n"); - } - - @Test - public void testMultipleEventsOfSameParagraph() throws InterruptedException { - RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); - String note1 = "note1"; - String para1 = "para1"; - String[][] buffer = { - {note1, para1, "data1\n"}, - {note1, para1, "data2\n"}, - {note1, para1, "data3\n"} - }; - - loopForCompletingEvents(listener, 1, buffer); - verify(listener, times(1)).onOutputAppend(any(String.class), any(String.class), anyInt(), any(String.class)); - verify(listener, times(1)).onOutputAppend(note1, para1, 0, "data1\ndata2\ndata3\n"); + AppendOutputRunner runner = new AppendOutputRunner(listener); + List batch = new ArrayList<>(); + batch.add(new AppendOutputBuffer("note", "para", 0, "a")); + batch.add(new AppendOutputBuffer("note", "para", 0, "b")); + verifyNoInteractions(listener); + runner.run(batch); + batch.clear(); + batch.add(new AppendOutputBuffer("note", "para", 0, "c")); + runner.run(batch); + batch.clear(); + runner.run(batch); + InOrder order = inOrder(listener); + order.verify(listener).onOutputAppend("note", "para", 0, "ab"); + order.verify(listener).onOutputAppend("note", "para", 0, "c"); + order.verifyNoMoreInteractions(); } @Test - void testUpdateDoesNotOvertakeQueuedAppend() { + void preservesOrderAndDoesNotMergeDifferentOutputKeys() { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); AppendOutputRunner runner = new AppendOutputRunner(listener); - runner.appendBuffer("note", "para", 0, "before-1\n"); - runner.appendBuffer("note", "para", 0, "before-2\n"); - runner.updateBuffer("note", "para", 0, InterpreterResult.Type.TEXT, "replacement\n"); - runner.appendBuffer("note", "para", 0, "after\n"); - - runner.run(); - + List batch = new ArrayList<>(); + batch.add(new AppendOutputBuffer("note:1", "p:1", 0, "first")); + batch.add(new AppendOutputBuffer("note:1", "p:2", 0, "second")); + batch.add(new AppendOutputBuffer("note:1", "p:1", 1, "third")); + batch.add(new AppendOutputBuffer("note:2", "p:1", 1, "fourth")); + batch.add(new AppendOutputBuffer("note:1", "p:1", 0, "fifth")); + runner.run(batch); InOrder order = inOrder(listener); - order.verify(listener).onOutputAppend("note", "para", 0, "before-1\nbefore-2\n"); - order.verify(listener).onOutputUpdated( - "note", "para", 0, InterpreterResult.Type.TEXT, "replacement\n"); - order.verify(listener).onOutputAppend("note", "para", 0, "after\n"); + order.verify(listener).onOutputAppend("note:1", "p:1", 0, "first"); + order.verify(listener).onOutputAppend("note:1", "p:2", 0, "second"); + order.verify(listener).onOutputAppend("note:1", "p:1", 1, "third"); + order.verify(listener).onOutputAppend("note:2", "p:1", 1, "fourth"); + order.verify(listener).onOutputAppend("note:1", "p:1", 0, "fifth"); + order.verifyNoMoreInteractions(); } @Test - void testMultipleEventsOfDifferentParagraphs() throws InterruptedException { + void staleAppendDoesNotDiscardLaterOutputOrLeakIntoItsBuffer() { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); - String note1 = "note1"; - String note2 = "note2"; - String para1 = "para1"; - String para2 = "para2"; - String[][] buffer = { - {note1, para1, "data1\n"}, - {note1, para2, "data2\n"}, - {note2, para1, "data3\n"}, - {note2, para2, "data4\n"} - }; - loopForCompletingEvents(listener, 4, buffer); - - verify(listener, times(4)).onOutputAppend(any(String.class), any(String.class), anyInt(), any(String.class)); - verify(listener, times(1)).onOutputAppend(note1, para1, 0, "data1\n"); - verify(listener, times(1)).onOutputAppend(note1, para2, 0, "data2\n"); - verify(listener, times(1)).onOutputAppend(note2, para1, 0, "data3\n"); - verify(listener, times(1)).onOutputAppend(note2, para2, 0, "data4\n"); + doThrow(new IllegalStateException("removed")).when(listener) + .onOutputAppend("note", "gone", 0, "bad"); + AppendOutputRunner runner = new AppendOutputRunner(listener); + List batch = new ArrayList<>(); + batch.add(new AppendOutputBuffer("note", "gone", 0, "bad")); + batch.add(new AppendOutputBuffer("note", "present", 0, "good")); + runner.run(batch); + InOrder order = inOrder(listener); + order.verify(listener).onOutputAppend("note", "gone", 0, "bad"); + order.verify(listener).onOutputAppend("note", "present", 0, "good"); + order.verifyNoMoreInteractions(); } @Test - void testClubbedData() throws InterruptedException { + void largeAppendStreamIsDeliveredAsOneBatchWithoutLosingData() { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + List received = new ArrayList<>(); + doAnswer(call -> { + received.add(call.getArgument(3)); + return null; + }).when(listener).onOutputAppend(anyString(), anyString(), anyInt(), anyString()); AppendOutputRunner runner = new AppendOutputRunner(listener); - future = service.scheduleWithFixedDelay(runner, 0, - AppendOutputRunner.BUFFER_TIME_MS, TimeUnit.MILLISECONDS); - Thread thread = new Thread(new BombardEvents(runner)); - thread.start(); - thread.join(); - Thread.sleep(1000); - - /* NUM_CLUBBED_EVENTS is a heuristic number. - * It has been observed that for 10,000 continuos event - * calls, 30-40 Web-socket calls are made. Keeping - * the unit-test to a pessimistic 100 web-socket calls. - */ - verify(listener, atMost(NUM_CLUBBED_EVENTS)).onOutputAppend(any(String.class), any(String.class), anyInt(), any(String.class)); + List batch = new ArrayList<>(); + StringBuilder expected = new StringBuilder(); + for (int i = 0; i < 10000; i++) { + String token = i + "\n"; + expected.append(token); + batch.add(new AppendOutputBuffer("note", "para", 0, token)); + } + runner.run(batch); + assertEquals(List.of(expected.toString()), received); } - @Test - void testWarnLoggerForLargeData() throws InterruptedException { + @ParameterizedTest + @ValueSource(ints = {100000, 100001}) + void warnsOnlyWhenBufferedOutputExceedsTheSizeThreshold(int size) { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); AppendOutputRunner runner = new AppendOutputRunner(listener); - String data = "data\n"; - int numEvents = 100000; + List batch = new ArrayList<>(); + String output = "a".repeat(size); + batch.add(new AppendOutputBuffer("note", "para", 0, output)); + List sizeWarnings = new ArrayList<>(); + AppenderSkeleton appender = new AppenderSkeleton() { + @Override + protected void append(LoggingEvent event) { + String message = event.getRenderedMessage(); + if (Level.WARN.equals(event.getLevel()) + && message.startsWith("Processing size for buffered append-output is high:")) { + sizeWarnings.add(message); + } + } - for (int i=0; i - Level.WARN.equals(event.getLevel()) && expected.equals(event.getMessage()))); + runner.run(batch); + assertEquals(size > 100000 ? List.of( + "Processing size for buffered append-output is high: " + size + " characters.") + : List.of(), sizeWarnings); + verify(listener).onOutputAppend("note", "para", 0, output); } finally { logger.removeAppender(appender); + logger.setLevel(previousLevel); + logger.setAdditivity(previousAdditivity); + appender.close(); } } @Test - void emptyDrainDoesNotBlock() { - AppendOutputRunner runner = - new AppendOutputRunner(mock(RemoteInterpreterProcessListener.class)); - assertTimeoutPreemptively(Duration.ofSeconds(1), runner::run); - } - - @Test - void updateFailureDoesNotDiscardOtherEventsOrLaterDrains() { + void disallowedDeliverySkipsTheBatchAndDoesNotAffectLaterBatches() { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); AppendOutputRunner runner = new AppendOutputRunner(listener); - doThrow(new IllegalStateException("removed")).when(listener) - .onOutputUpdated("note", "gone", 0, InterpreterResult.Type.TEXT, "bad"); - runner.appendBuffer("note", "gone", 0, "bad"); - runner.updateBuffer("note", "gone", 0, InterpreterResult.Type.TEXT, "bad"); - runner.appendBuffer("note", "present", 0, "good"); - runner.run(); - runner.appendBuffer("note", "present", 0, "later"); - runner.run(); - verify(listener).onOutputAppend("note", "present", 0, "good"); - verify(listener).onOutputAppend("note", "present", 0, "later"); - } - - @Test - void appendFailureDoesNotDiscardLaterUpdateOrDrain() { - RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); - AppendOutputRunner runner = new AppendOutputRunner(listener); - doThrow(new IllegalStateException("removed")).when(listener) - .onOutputAppend("note", "gone", 0, "bad"); - runner.appendBuffer("note", "gone", 0, "bad"); - runner.updateBuffer("note", "present", 0, InterpreterResult.Type.TEXT, "current"); - runner.run(); - runner.appendBuffer("note", "present", 0, "later"); - runner.run(); - verify(listener).onOutputUpdated("note", "present", 0, - InterpreterResult.Type.TEXT, "current"); - verify(listener).onOutputAppend("note", "present", 0, "later"); - } - - @Test - void concurrentDrainCannotOvertakeInFlightCallback() throws Exception { - RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); - AppendOutputRunner runner = new AppendOutputRunner(listener); - CountDownLatch entered = new CountDownLatch(1); - CountDownLatch release = new CountDownLatch(1); - doAnswer(invocation -> { - entered.countDown(); - assertTrue(release.await(5, TimeUnit.SECONDS)); - return null; - }).when(listener).onOutputAppend("note", "para", 0, "old"); - ExecutorService executor = Executors.newFixedThreadPool(2); - try { - runner.appendBuffer("note", "para", 0, "old"); - Future first = executor.submit(runner); - assertTrue(entered.await(5, TimeUnit.SECONDS)); - runner.updateBuffer("note", "para", 0, InterpreterResult.Type.TEXT, "new"); - Future second = executor.submit(runner); - assertThrows(TimeoutException.class, () -> second.get(100, TimeUnit.MILLISECONDS)); - release.countDown(); - first.get(5, TimeUnit.SECONDS); - second.get(5, TimeUnit.SECONDS); - InOrder order = inOrder(listener); - order.verify(listener).onOutputAppend("note", "para", 0, "old"); - order.verify(listener).onOutputUpdated("note", "para", 0, - InterpreterResult.Type.TEXT, "new"); - } finally { - release.countDown(); - executor.shutdownNow(); - } - } - - private class BombardEvents implements Runnable { - - private final AppendOutputRunner runner; - - private BombardEvents(AppendOutputRunner runner) { - this.runner = runner; - } - - @Override - public void run() { - String noteId = "noteId"; - String paraId = "paraId"; - for (int i=0; i log = new ArrayList<>(); - - @Override - public boolean requiresLayout() { - return false; - } - - @Override - protected void append(final LoggingEvent loggingEvent) { - log.add(loggingEvent); - } - - @Override - public void close() { - } - - public List getLog() { - return new ArrayList<>(log); - } - } - - private void prepareInvocationCounts(RemoteInterpreterProcessListener listener) { - doAnswer(new Answer() { - @Override - public Void answer(InvocationOnMock invocation) throws Throwable { - numInvocations += 1; - return null; - } - }).when(listener).onOutputAppend(any(String.class), any(String.class), anyInt(), any(String.class)); - } - - private void loopForCompletingEvents(RemoteInterpreterProcessListener listener, - int numTimes, String[][] buffer) { - numInvocations = 0; - prepareInvocationCounts(listener); - AppendOutputRunner runner = new AppendOutputRunner(listener); - for (String[] bufferElement: buffer) { - runner.appendBuffer(bufferElement[0], bufferElement[1], 0, bufferElement[2]); - } - future = service.scheduleWithFixedDelay(runner, 0, - AppendOutputRunner.BUFFER_TIME_MS, TimeUnit.MILLISECONDS); - long startTimeMs = System.currentTimeMillis(); - while(numInvocations != numTimes) { - if (System.currentTimeMillis() - startTimeMs > 2000) { - fail("Buffered events were not sent for 2 seconds"); - } - } + List batch = new ArrayList<>(); + batch.add(new AppendOutputBuffer("note", "para", 0, "discarded")); + runner.run(batch, () -> false); + batch.clear(); + verifyNoInteractions(listener); + batch.add(new AppendOutputBuffer("note", "para", 0, "new")); + runner.run(batch); + InOrder order = inOrder(listener); + order.verify(listener).onOutputAppend("note", "para", 0, "new"); + order.verifyNoMoreInteractions(); } } diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcherTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcherTest.java new file mode 100644 index 00000000000..58241dfc28b --- /dev/null +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcherTest.java @@ -0,0 +1,536 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.zeppelin.interpreter.remote; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.InOrder; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.zeppelin.interpreter.InterpreterResult; +import org.apache.zeppelin.interpreter.InterpreterResultMessage; + +class ParagraphOutputDispatcherTest { + @Test + void boundariesFlushEarlierAppendsAndKeepLaterAppendsInOrder() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener)) { + dispatcher.appendOutput("note", "para", 0, "a"); + dispatcher.appendOutput("note", "para", 0, "b"); + verifyNoInteractions(listener); + dispatcher.updateOutput("note", "para", 0, InterpreterResult.Type.TEXT, "replacement") + .get(5, TimeUnit.SECONDS); + dispatcher.appendOutput("note", "para", 0, "after"); + dispatcher.checkpointOutput("note", "para").get(5, TimeUnit.SECONDS); + InOrder order = inOrder(listener); + order.verify(listener).onOutputAppend("note", "para", 0, "ab"); + order.verify(listener).onOutputUpdated( + "note", "para", 0, InterpreterResult.Type.TEXT, "replacement"); + order.verify(listener).onOutputAppend("note", "para", 0, "after"); + order.verify(listener).checkpointOutput("note", "para"); + order.verifyNoMoreInteractions(); + } + } + + @Test + void scheduledFlushDeliversAppendsWithoutRpcBoundaries() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + CountDownLatch delivered = new CountDownLatch(1); + doAnswer(call -> { + delivered.countDown(); + return null; + }).when(listener).onOutputAppend("note", "para", 0, "periodic"); + ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener)) { + dispatcher.appendOutput("note", "para", 0, "periodic"); + timer.scheduleWithFixedDelay(dispatcher::flush, 0, + AppendOutputRunner.BUFFER_TIME_MS, TimeUnit.MILLISECONDS); + assertTrue(delivered.await(5, TimeUnit.SECONDS)); + dispatcher.checkpointOutput("note", "para").get(5, TimeUnit.SECONDS); + verify(listener).onOutputAppend("note", "para", 0, "periodic"); + } finally { + timer.shutdownNow(); + } + } + + @Test + void flushAndSubmissionDoNotWaitForCallbacksOrLoseAnInFlightFlushRequest() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + CountDownLatch later = new CountDownLatch(1); + doAnswer(call -> { + entered.countDown(); + assertTrue(release.await(5, TimeUnit.SECONDS)); + return null; + }).when(listener).onOutputAppend("note", "para", 0, "slow"); + doAnswer(call -> { + later.countDown(); + return null; + }).when(listener).onOutputAppend("note", "para", 0, "later"); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener)) { + dispatcher.appendOutput("note", "para", 0, "slow"); + dispatcher.flush(); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + assertTimeoutPreemptively(Duration.ofSeconds(1), () -> { + dispatcher.appendOutput("note", "para", 0, "later"); + dispatcher.flush(); + }); + release.countDown(); + // No further tick or boundary may rescue a lost flush request. + assertTrue(later.await(5, TimeUnit.SECONDS)); + } finally { + release.countDown(); + } + } + + @Test + void boundaryFailureIsReportedAndDoesNotDiscardLaterEvents() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + doThrow(new IllegalStateException("removed")).when(listener) + .onOutputUpdated("note", "gone", 0, InterpreterResult.Type.TEXT, "bad"); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener)) { + Future failed = dispatcher.updateOutput( + "note", "gone", 0, InterpreterResult.Type.TEXT, "bad"); + ExecutionException failure = assertThrows(ExecutionException.class, + () -> failed.get(5, TimeUnit.SECONDS)); + assertEquals("removed", failure.getCause().getMessage()); + dispatcher.appendOutput("note", "present", 0, "good"); + dispatcher.checkpointOutput("note", "present").get(5, TimeUnit.SECONDS); + verify(listener).onOutputAppend("note", "present", 0, "good"); + } + } + + @Test + void largeNoteYieldsAndKeepsAllOutputAcrossWorkerBatches() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + StringBuilder received = new StringBuilder(); + AtomicInteger callbacks = new AtomicInteger(); + AtomicInteger sizeWhenSmallRan = new AtomicInteger(); + doAnswer(call -> { + if ("large".equals(call.getArgument(0))) { + received.append((String) call.getArgument(3)); + if (callbacks.incrementAndGet() == 1) { + entered.countDown(); + awaitIgnoringInterrupt(release); + } + } else { + sizeWhenSmallRan.set(received.length()); + } + return null; + }).when(listener).onOutputAppend(anyString(), anyString(), anyInt(), anyString()); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener, 1, 2)) { + for (int i = 0; i < 5; i++) { + dispatcher.appendOutput("large", "para", 0, "x"); + } + Future largeDone = dispatcher.checkpointOutput("large", "para"); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + // Submit only after a large batch has been claimed, rather than ahead of its ready entry. + dispatcher.appendOutput("small", "para", 0, "small"); + Future smallDone = dispatcher.checkpointOutput("small", "para"); + release.countDown(); + smallDone.get(5, TimeUnit.SECONDS); + largeDone.get(5, TimeUnit.SECONDS); + assertTrue(sizeWhenSmallRan.get() > 0); + assertTrue(sizeWhenSmallRan.get() < 5, "Small note must run before all large output"); + InOrder order = inOrder(listener); + order.verify(listener).onOutputAppend("small", "para", 0, "small"); + order.verify(listener).checkpointOutput("large", "para"); + assertEquals("x".repeat(5), received.toString()); + assertTrue(callbacks.get() < 5, "Adjacent appends must still be batched"); + } finally { + release.countDown(); + } + } + + @Test + void concurrentProducersKeepOneWriterAndDeliverEveryEventOnceInProducerOrder() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + AtomicInteger active = new AtomicInteger(); + AtomicInteger maximum = new AtomicInteger(); + List received = Collections.synchronizedList(new ArrayList<>()); + doAnswer(call -> { + maximum.accumulateAndGet(active.incrementAndGet(), Math::max); + try { + String output = call.getArgument(3); + Collections.addAll(received, output.split("\n")); + } finally { + active.decrementAndGet(); + } + return null; + }).when(listener).onOutputAppend(anyString(), anyString(), anyInt(), anyString()); + ExecutorService producers = Executors.newFixedThreadPool(4); + ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener)) { + timer.scheduleWithFixedDelay(dispatcher::flush, 0, 1, TimeUnit.MILLISECONDS); + List> futures = new ArrayList<>(); + for (int producer = 0; producer < 4; producer++) { + final int id = producer; + futures.add(producers.submit(() -> { + for (int i = 0; i < 1000; i++) { + dispatcher.appendOutput("note", "para", 0, id + ":" + i + "\n"); + if (i % 100 == 0) { + dispatcher.checkpointOutput("note", "para").get(5, TimeUnit.SECONDS); + } + } + return null; + })); + } + for (Future future : futures) { + future.get(10, TimeUnit.SECONDS); + } + dispatcher.checkpointOutput("note", "para").get(5, TimeUnit.SECONDS); + assertEquals(1, maximum.get()); + assertEquals(4000, received.size()); + assertEquals(4000, new HashSet<>(received).size()); + int[] next = new int[4]; + for (String token : received) { + String[] parts = token.split(":"); + int producer = Integer.parseInt(parts[0]); + assertEquals(next[producer]++, Integer.parseInt(parts[1])); + } + } finally { + timer.shutdownNow(); + producers.shutdownNow(); + } + } + + @Test + void readyNotesWaitForWorkerCapacityAndReuseReleasedWorkers() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + CountDownLatch occupied = new CountDownLatch(2); + CountDownLatch release = new CountDownLatch(1); + AtomicInteger active = new AtomicInteger(); + AtomicInteger maximum = new AtomicInteger(); + doAnswer(call -> { + maximum.accumulateAndGet(active.incrementAndGet(), Math::max); + try { + if (!"C".equals(call.getArgument(0))) { + occupied.countDown(); + assertTrue(release.await(5, TimeUnit.SECONDS)); + } + } finally { + active.decrementAndGet(); + } + return null; + }).when(listener).checkpointOutput(anyString(), anyString()); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener, 2)) { + Future first = dispatcher.checkpointOutput("A", "para"); + Future second = dispatcher.checkpointOutput("B", "para"); + assertTrue(occupied.await(5, TimeUnit.SECONDS)); + Future third = dispatcher.checkpointOutput("C", "para"); + assertThrows(TimeoutException.class, () -> third.get(100, TimeUnit.MILLISECONDS)); + release.countDown(); + first.get(5, TimeUnit.SECONDS); + second.get(5, TimeUnit.SECONDS); + third.get(5, TimeUnit.SECONDS); + assertEquals(2, maximum.get()); + verify(listener).checkpointOutput("A", "para"); + verify(listener).checkpointOutput("B", "para"); + verify(listener).checkpointOutput("C", "para"); + } finally { + release.countDown(); + } + } + + @Test + void idleQueuesAreReclaimedAndTheSameNoteCanDeliverAgain() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener)) { + for (int i = 0; i < 200; i++) { + dispatcher.appendOutput("note" + i, "para", 0, "data"); + dispatcher.checkpointOutput("note" + i, "para").get(5, TimeUnit.SECONDS); + } + await().atMost(Duration.ofSeconds(5)).until(() -> dispatcher.pendingNoteCount() == 0); + dispatcher.appendOutput("note0", "para", 0, "again"); + dispatcher.checkpointOutput("note0", "para").get(5, TimeUnit.SECONDS); + verify(listener).onOutputAppend("note0", "para", 0, "data"); + verify(listener).onOutputAppend("note0", "para", 0, "again"); + await().atMost(Duration.ofSeconds(5)).until(() -> dispatcher.pendingNoteCount() == 0); + } + } + + @Test + void closeCancelsABoundaryAlreadyDrainedBehindABlockedAppend() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicReference callbackThread = new AtomicReference<>(); + doAnswer(call -> { + callbackThread.set(Thread.currentThread()); + entered.countDown(); + try { + release.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return null; + }).when(listener).onOutputAppend("note", "para", 0, "blocked"); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener, 1)) { + dispatcher.appendOutput("note", "para", 0, "blocked"); + Future boundary = dispatcher.checkpointOutput("note", "para"); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + dispatcher.close(); + assertStopped(boundary); + release.countDown(); + callbackThread.get().join(5000); + assertFalse(callbackThread.get().isAlive()); + verify(listener, never()).checkpointOutput("note", "para"); + } finally { + release.countDown(); + } + } + + @Test + void closeFailsQueuedAndInFlightBoundariesAndRejectsNewEvents() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + doAnswer(call -> { + entered.countDown(); + try { + release.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return null; + }).when(listener).checkpointOutput("A", "para"); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener, 1)) { + Future inFlight = dispatcher.checkpointOutput("A", "para"); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + Future queued = dispatcher.checkpointOutput("B", "para"); + dispatcher.close(); + assertStopped(inFlight); + assertStopped(queued); + assertEquals(0, dispatcher.pendingNoteCount()); + assertThrows(IllegalStateException.class, () -> + dispatcher.appendOutput("A", "para", 0, "late")); + assertThrows(IllegalStateException.class, () -> dispatcher.checkpointOutput("A", "para")); + verify(listener, never()).checkpointOutput("B", "para"); + } finally { + release.countDown(); + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void shutdownStopsLaterAppendGroupsEvenWhenActiveCallbackIgnoresInterrupt(boolean withBoundary) + throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicReference worker = new AtomicReference<>(); + doAnswer(call -> { + worker.set(Thread.currentThread()); + entered.countDown(); + awaitIgnoringInterrupt(release); + return null; + }).when(listener).onOutputAppend("note", "first", 0, "a"); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener, 1)) { + dispatcher.appendOutput("note", "first", 0, "a"); + dispatcher.appendOutput("note", "second", 0, "b"); + Future boundary = null; + if (withBoundary) { + boundary = dispatcher.checkpointOutput("note", "para"); + } else { + dispatcher.flush(); + } + assertTrue(entered.await(5, TimeUnit.SECONDS)); + dispatcher.close(); + if (boundary != null) { + assertStopped(boundary); + } + release.countDown(); + worker.get().join(5000); + assertFalse(worker.get().isAlive()); + verify(listener).onOutputAppend("note", "first", 0, "a"); + verify(listener, never()).onOutputAppend("note", "second", 0, "b"); + verify(listener, never()).checkpointOutput("note", "para"); + } finally { + release.countDown(); + } + } + + @Test + void closeReleasesAllBoundaryWaitersWithoutWaitingForAnActiveCallback() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + CountDownLatch interrupted = new CountDownLatch(1); + AtomicReference worker = new AtomicReference<>(); + doAnswer(call -> { + worker.set(Thread.currentThread()); + entered.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + interrupted.countDown(); + awaitIgnoringInterrupt(release); + } + return null; + }).when(listener).checkpointOutput("note", "running"); + ExecutorService callers = Executors.newFixedThreadPool(4); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener, 1)) { + List> boundaries = new ArrayList<>(); + boundaries.add(dispatcher.checkpointOutput("note", "running")); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + for (int i = 0; i < 20; i++) { + boundaries.add(dispatcher.checkpointOutput("note", "queued" + i)); + boundaries.add(dispatcher.checkpointOutput("other", "queued" + i)); + } + CountDownLatch waiting = new CountDownLatch(4); + List> waiters = new ArrayList<>(); + for (int i = 0; i < 4; i++) { + Future boundary = boundaries.get(i); + waiters.add(callers.submit(() -> { + waiting.countDown(); + assertStopped(boundary); + })); + } + assertTrue(waiting.await(5, TimeUnit.SECONDS)); + assertTimeoutPreemptively(Duration.ofSeconds(1), dispatcher::close); + assertTrue(interrupted.await(5, TimeUnit.SECONDS)); + for (Future boundary : boundaries) { + assertStopped(boundary); + } + for (Future waiter : waiters) { + waiter.get(5, TimeUnit.SECONDS); + } + assertEquals(1, release.getCount(), "Callback must still be blocked after waiters exit"); + assertThrows(IllegalStateException.class, + () -> dispatcher.appendOutput("other", "para", 0, "late")); + verify(listener, never()).checkpointOutput("note", "queued0"); + verify(listener, never()).checkpointOutput("other", "queued0"); + release.countDown(); + worker.get().join(5000); + assertFalse(worker.get().isAlive(), "Shutdown must survive a callback clearing interruption"); + } finally { + release.countDown(); + callers.shutdownNow(); + } + } + + @Test + void acceptedBoundaryCannotBeCancelledOrSkipped() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + doAnswer(call -> { + entered.countDown(); + awaitIgnoringInterrupt(release); + return null; + }).when(listener).onOutputAppend("note", "para", 0, "before"); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener, 1)) { + dispatcher.appendOutput("note", "para", 0, "before"); + Future boundary = dispatcher.checkpointOutput("note", "para"); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + assertFalse(boundary.cancel(true)); + assertFalse(boundary.isCancelled()); + release.countDown(); + boundary.get(5, TimeUnit.SECONDS); + verify(listener).checkpointOutput("note", "para"); + } finally { + release.countDown(); + } + } + + @Test + void queuedUpdateAllUsesASnapshotOfItsReplacementList() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + doAnswer(call -> { + entered.countDown(); + assertTrue(release.await(5, TimeUnit.SECONDS)); + return null; + }).when(listener).onOutputAppend("note", "para", 0, "old"); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener, 1)) { + dispatcher.appendOutput("note", "para", 0, "old"); + dispatcher.flush(); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + List replacements = new ArrayList<>(); + replacements.add(new InterpreterResultMessage(InterpreterResult.Type.TEXT, "new")); + Future updated = dispatcher.updateAllOutput("note", "para", replacements); + replacements.clear(); + release.countDown(); + updated.get(5, TimeUnit.SECONDS); + InOrder order = inOrder(listener); + order.verify(listener).onOutputAppend("note", "para", 0, "old"); + order.verify(listener).onOutputClear("note", "para"); + order.verify(listener).onOutputUpdated("note", "para", 0, InterpreterResult.Type.TEXT, "new"); + order.verifyNoMoreInteractions(); + } finally { + release.countDown(); + } + } + + private void assertStopped(Future completion) { + ExecutionException failure = assertThrows(ExecutionException.class, + () -> completion.get(5, TimeUnit.SECONDS)); + assertTrue(failure.getCause() instanceof IllegalStateException); + } + + private void awaitIgnoringInterrupt(CountDownLatch release) { + boolean interrupted = false; + try { + while (true) { + try { + release.await(); + return; + } catch (InterruptedException e) { + interrupted = true; + } + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } +} From 8646d29c4ce915db4eaa9f24f8daf02a6eb60aa5 Mon Sep 17 00:00:00 2001 From: miinhho Date: Wed, 16 Sep 2026 03:47:49 +0900 Subject: [PATCH 2/4] [ZEPPELIN-6709] Fix import order in output event server test --- .../zeppelin/interpreter/RemoteInterpreterEventServerTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java index 035f06137e5..2a60b0d9704 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java @@ -48,8 +48,8 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; -import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars; import org.apache.zeppelin.conf.ZeppelinConfiguration; +import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars; import org.apache.zeppelin.interpreter.remote.InvokeResourceMethodEventMessage; import org.apache.zeppelin.interpreter.remote.ParagraphOutputDispatcher; import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess; From 46a1351fbfac2125de440a4ad63173d8dc02c0e3 Mon Sep 17 00:00:00 2001 From: miinhho Date: Wed, 16 Sep 2026 04:04:06 +0900 Subject: [PATCH 3/4] [ZEPPELIN-6709] Log output boundary timing at debug level --- .../interpreter/remote/ParagraphOutputDispatcher.java | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcher.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcher.java index 70535b88bd0..8860bbf8efd 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcher.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcher.java @@ -50,7 +50,6 @@ public class ParagraphOutputDispatcher implements AutoCloseable { private static final Logger LOGGER = LoggerFactory.getLogger(ParagraphOutputDispatcher.class); - private static final long SAFE_PROCESSING_TIME_MS = 10; // Guarded by this: notes, mutable NoteQueue state, and worker startup. private final Map notes = new HashMap<>(); @@ -139,11 +138,7 @@ private Future enqueueBoundary(String noteId, Runnable callback) { throw e; } finally { long time = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); - if (time > SAFE_PROCESSING_TIME_MS) { - LOGGER.warn("Processing output boundary for note {} took {} milliseconds", noteId, time); - } else { - LOGGER.debug("Processing output boundary for note {} took {} milliseconds", noteId, time); - } + LOGGER.debug("Processing output boundary for note {} took {} milliseconds", noteId, time); } })); enqueue(noteId, event, true); From 00e8eace8a90f3f91f40262b1af9515252780465 Mon Sep 17 00:00:00 2001 From: miinhho Date: Tue, 22 Sep 2026 18:20:21 +0900 Subject: [PATCH 4/4] [ZEPPELIN-6709] Restore key-based append output coalescing --- .../remote/AppendOutputRunner.java | 80 +++++++++++++------ .../remote/AppendOutputRunnerTest.java | 5 +- .../remote/ParagraphOutputDispatcherTest.java | 47 +++++++++++ 3 files changed, 104 insertions(+), 28 deletions(-) diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java index 3007172dca9..dfddd5a15bb 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java @@ -19,12 +19,14 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.function.BooleanSupplier; /** - * Synchronously delivers an append batch, merging adjacent events for the same paragraph output. + * Synchronously delivers an append batch, merging events for each paragraph output. */ public class AppendOutputRunner { private static final Logger LOGGER = LoggerFactory.getLogger(AppendOutputRunner.class); @@ -49,25 +51,22 @@ void run(List batch, BooleanSupplier mayDeliver) { } long start = System.currentTimeMillis(); long size = 0; - AppendOutputBuffer previous = null; - StringBuilder data = new StringBuilder(); + Map 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()); + } + currentData.append(append.getData()); + } + for (Map.Entry group : groups.entrySet()) { if (!mayDeliver.getAsBoolean()) { return; } - if (previous != null && !sameOutput(previous, append)) { - size += flush(previous, data); - if (!mayDeliver.getAsBoolean()) { - return; - } - } - previous = append; - data.append(append.getData()); - } - if (!mayDeliver.getAsBoolean()) { - return; + size += flush(group.getKey(), group.getValue()); } - size += flush(previous, data); long time = System.currentTimeMillis() - start; if (time > SAFE_PROCESSING_TIME) { LOGGER.warn("Processing time for buffered append-output is high: {} milliseconds.", time); @@ -81,23 +80,54 @@ void run(List batch, BooleanSupplier mayDeliver) { } } - private boolean sameOutput(AppendOutputBuffer first, AppendOutputBuffer second) { - return first.getIndex() == second.getIndex() - && Objects.equals(first.getNoteId(), second.getNoteId()) - && Objects.equals(first.getParagraphId(), second.getParagraphId()); - } - - private long flush(AppendOutputBuffer append, StringBuilder data) { + private long flush(ParagraphOutputKey key, StringBuilder data) { long size = data.length(); try { - listener.onOutputAppend(append.getNoteId(), append.getParagraphId(), append.getIndex(), - data.toString()); + 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 {}", - append.getNoteId(), append.getParagraphId(), e); + key.noteId, key.paragraphId, e); } data.setLength(0); return size; } + + 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()); + } + + @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; + } + } } diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java index 98b6f658b47..548307f6d86 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java @@ -59,7 +59,7 @@ void batchesAdjacentAppendsAndHandlesEmptyBatches() { } @Test - void preservesOrderAndDoesNotMergeDifferentOutputKeys() { + void mergesEachOutputKeyInFirstAppearanceOrder() { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); AppendOutputRunner runner = new AppendOutputRunner(listener); List batch = new ArrayList<>(); @@ -70,11 +70,10 @@ void preservesOrderAndDoesNotMergeDifferentOutputKeys() { batch.add(new AppendOutputBuffer("note:1", "p:1", 0, "fifth")); runner.run(batch); InOrder order = inOrder(listener); - order.verify(listener).onOutputAppend("note:1", "p:1", 0, "first"); + order.verify(listener).onOutputAppend("note:1", "p:1", 0, "firstfifth"); order.verify(listener).onOutputAppend("note:1", "p:2", 0, "second"); order.verify(listener).onOutputAppend("note:1", "p:1", 1, "third"); order.verify(listener).onOutputAppend("note:2", "p:1", 1, "fourth"); - order.verify(listener).onOutputAppend("note:1", "p:1", 0, "fifth"); order.verifyNoMoreInteractions(); } diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcherTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcherTest.java index 58241dfc28b..50b2d68f1d8 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcherTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/ParagraphOutputDispatcherTest.java @@ -76,6 +76,53 @@ void boundariesFlushEarlierAppendsAndKeepLaterAppendsInOrder() throws Exception } } + @Test + void interleavedAppendsMergeWithinEachBoundary() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener)) { + dispatcher.appendOutput("note", "first", 0, "a"); + dispatcher.appendOutput("note", "second", 0, "b"); + dispatcher.appendOutput("note", "first", 0, "c"); + dispatcher.updateOutput("note", "first", 0, InterpreterResult.Type.TEXT, "replace") + .get(5, TimeUnit.SECONDS); + dispatcher.appendOutput("note", "first", 0, "after"); + dispatcher.checkpointOutput("note", "first").get(5, TimeUnit.SECONDS); + + InOrder order = inOrder(listener); + order.verify(listener).onOutputAppend("note", "first", 0, "ac"); + order.verify(listener).onOutputAppend("note", "second", 0, "b"); + order.verify(listener).onOutputUpdated( + "note", "first", 0, InterpreterResult.Type.TEXT, "replace"); + order.verify(listener).onOutputAppend("note", "first", 0, "after"); + order.verify(listener).checkpointOutput("note", "first"); + order.verifyNoMoreInteractions(); + } + } + + @Test + void heavilyInterleavedParagraphsUseOneCallbackPerOutputKey() throws Exception { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + StringBuilder expected = new StringBuilder(); + try (ParagraphOutputDispatcher dispatcher = new ParagraphOutputDispatcher(listener, 1, 1000)) { + for (int i = 0; i < 250; i++) { + String token = i + ";"; + expected.append(token); + for (int paragraph = 0; paragraph < 4; paragraph++) { + dispatcher.appendOutput("note", "p" + paragraph, 0, token); + } + } + dispatcher.checkpointOutput("note", "p0").get(5, TimeUnit.SECONDS); + + InOrder order = inOrder(listener); + for (int paragraph = 0; paragraph < 4; paragraph++) { + order.verify(listener).onOutputAppend("note", "p" + paragraph, 0, + expected.toString()); + } + order.verify(listener).checkpointOutput("note", "p0"); + order.verifyNoMoreInteractions(); + } + } + @Test void scheduledFlushDeliversAppendsWithoutRpcBoundaries() throws Exception { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class);