Skip to content

[ZEPPELIN-6129] Restore parallel paragraph execution for concurrent interpreters - #5332

Open
HwangRock wants to merge 1 commit into
apache:masterfrom
HwangRock:ZEPPELIN-6129
Open

[ZEPPELIN-6129] Restore parallel paragraph execution for concurrent interpreters#5332
HwangRock wants to merge 1 commit into
apache:masterfrom
HwangRock:ZEPPELIN-6129

Conversation

@HwangRock

Copy link
Copy Markdown
Contributor

What is this PR for?

Paragraphs of the same interpreter cannot run in parallel, even when the interpreter is configured for concurrent execution (e.g. a presto JDBC interpreter with zeppelin.jdbc.concurrent.use=true).

This is a regression from [ZEPPELIN-5900] (#4582). That PR switched RemoteScheduler's executor to a per-scheduler newSingleThreadExecutor so that RemoteSchedulerTest.testAbortOnPending became deterministic. Before it, RemoteScheduler was constructed with the shared multi-threaded pool from SchedulerFactory.getExecutor(), and every interpreter could submit paragraphs to the remote side concurrently.

Scheduling is two-tier: the server-side RemoteScheduler is a proxy, and each interpreter's remote-side getScheduler() (FIFO vs Parallel) decides the real concurrency. Once the server side became single-threaded, it gated every interpreter behind a single in-flight submission, so a remote ParallelScheduler never received a second job to run. The JDBC concurrency flag was read on the remote side but the server side never let a second paragraph through.

The fix restores an interpreter-neutral pool on the server side and delegates parallelism back to the remote scheduler:

  • RemoteScheduler: paragraph mode now builds a bounded newFixedThreadPool sized from zeppelin.interpreter.connection.poolsize (default 100, matching the RPC connection pool). Note mode keeps newSingleThreadExecutor to preserve in-note paragraph ordering. The JDBC-specific zeppelin.jdbc.concurrent.* gate is removed — it only helped JDBC and left Shell / Markdown / MongoDB / Neo4j / Cassandra / Flink SQL and other always-parallel interpreters serialized on the server side.
  • The paragraph/note blocking-wait in runJobInScheduler is untouched. It is what keeps FIFO interpreters (Python, Spark, ...) serial and note mode ordered: a FIFO interpreter's second job never reaches RUNNING remotely, so the wait still serializes it.
  • Cancellation hardening for the now-multithreaded pool: Job.aborted is volatile, and AbstractScheduler.runJob's abort gate runs under synchronized(runningJob), so a cancel arriving right before run reliably skips execution instead of racing.
  • JDBCUserConfigurations: paragraphIdStatementMap is a ConcurrentHashMap and cancelStatement null-checks the statement, so a cancel arriving before the statement is registered is a no-op rather than an NPE.

Net effect: interpreters whose remote getScheduler() returns a ParallelScheduler (Shell, Markdown, MongoDB, Neo4j, Cassandra, Flink SQL, JDBC with concurrency on, ...) regain parallel paragraph execution; FIFO interpreters and note mode stay serial.

What type of PR is it?

Bug Fix

What is the Jira issue?

ZEPPELIN-6129

How should this be tested?

Unit / integration:

./mvnw test -pl zeppelin-interpreter,jdbc,zeppelin-server -am \
  -Dtest=RemoteSchedulerTest,AbstractSchedulerAbortRaceTest,JobTest,JDBCUserConfigurationsTest,JDBCInterpreterTest
  • RemoteSchedulerTest#testParallelExecution_bothJobsRunConcurrently — two jobs reach RUNNING simultaneously through RemoteScheduler; fails against the single-thread executor, passes with the neutral pool.
  • RemoteSchedulerTest#testAbortOnPending_noteModeSerial — note mode still serializes and aborts a queued job before it runs.
  • AbstractSchedulerAbortRaceTest — the abort gate and cancel share the job monitor (latch-driven, deterministic).
  • JDBCUserConfigurationsTest — cancel-before-register is a no-op, not an NPE.

End-to-end (verified locally against a real server; the IT itself is kept out of this PR to keep CI light): two %sh paragraphs each running sleep 5, triggered via the async REST endpoint POST /api/notebook/job/{noteId}/{paragraphId}, with no concurrency property set (Shell is a ParallelScheduler by default). Status was polled and the two paragraphs were observed RUNNING at the same time:

both paragraphs observed RUNNING simultaneously at +3425ms
total elapsed until both FINISHED: 8287ms

Serial execution would be ~10000ms (2 x sleep 5), and a single-thread pool can never reach simultaneous RUNNING because the second paragraph stays PENDING until the first finishes. The overlap at +3425ms and the ~8.3s wall-clock confirm the two paragraphs ran in parallel.

Questions:

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

…nterpreters

RemoteScheduler serialized every remote submission through a per-scheduler
single-thread executor, so paragraphs of the same interpreter ran
sequentially even when the interpreter's remote scheduler is a
ParallelScheduler. This regressed from [ZEPPELIN-5900] (apache#4582), which
switched the shared multi-threaded pool to newSingleThreadExecutor to make
testAbortOnPending deterministic.

Replace the JDBC-only concurrency gate with an interpreter-neutral bounded
pool for paragraph mode (note mode stays single-threaded); actual
parallelism is delegated to each interpreter's remote-side scheduler, so
FIFO interpreters stay serial and concurrent ones run in parallel.

Harden cancellation for the now-multithreaded pool: make Job.aborted
volatile, run the runJob abort gate under synchronized(runningJob), and
guard JDBCUserConfigurations with ConcurrentHashMap + a null check in
cancelStatement.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant