This is an automated email from the ASF dual-hosted git repository.
jongyoul pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
The following commit(s) were added to refs/heads/master by this push:
new 149f51fed7 [ZEPPELIN-6129] Restore parallel paragraph execution for
concurrent interpreters
149f51fed7 is described below
commit 149f51fed7d9d3fb4000cd08ade00b34ee09b42c
Author: HwangRock <[email protected]>
AuthorDate: Sun Aug 9 23:09:50 2026 +0900
[ZEPPELIN-6129] Restore parallel paragraph execution for concurrent
interpreters
### 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](https://issues.apache.org/jira/browse/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
Closes #5332 from HwangRock/ZEPPELIN-6129.
Signed-off-by: Jongyoul Lee <[email protected]>
---
.../zeppelin/jdbc/JDBCUserConfigurations.java | 18 ++-
.../zeppelin/jdbc/JDBCUserConfigurationsTest.java | 68 +++++++++
.../zeppelin/scheduler/AbstractScheduler.java | 30 ++--
.../java/org/apache/zeppelin/scheduler/Job.java | 2 +-
.../scheduler/AbstractSchedulerAbortRaceTest.java | 168 +++++++++++++++++++++
.../apache/zeppelin/scheduler/RemoteScheduler.java | 46 +++++-
.../zeppelin/scheduler/RemoteSchedulerTest.java | 139 ++++++++++++++++-
7 files changed, 447 insertions(+), 24 deletions(-)
diff --git
a/jdbc/src/main/java/org/apache/zeppelin/jdbc/JDBCUserConfigurations.java
b/jdbc/src/main/java/org/apache/zeppelin/jdbc/JDBCUserConfigurations.java
index 311fb0bad0..bcaea51e0c 100644
--- a/jdbc/src/main/java/org/apache/zeppelin/jdbc/JDBCUserConfigurations.java
+++ b/jdbc/src/main/java/org/apache/zeppelin/jdbc/JDBCUserConfigurations.java
@@ -19,9 +19,9 @@ import org.apache.zeppelin.user.UsernamePassword;
import java.sql.SQLException;
import java.sql.Statement;
-import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
/**
* UserConfigurations for JDBC impersonation.
@@ -33,7 +33,7 @@ public class JDBCUserConfigurations {
private Boolean isSuccessful;
public JDBCUserConfigurations() {
- paragraphIdStatementMap = new HashMap<>();
+ paragraphIdStatementMap = new ConcurrentHashMap<>();
}
public void initStatementMap() throws SQLException {
@@ -67,14 +67,26 @@ public class JDBCUserConfigurations {
}
public void saveStatement(String paragraphId, Statement statement) throws
SQLException {
+ if (paragraphId == null) {
+ return;
+ }
paragraphIdStatementMap.put(paragraphId, statement);
}
public void cancelStatement(String paragraphId) throws SQLException {
- paragraphIdStatementMap.get(paragraphId).cancel();
+ if (paragraphId == null) {
+ return;
+ }
+ Statement statement = paragraphIdStatementMap.get(paragraphId);
+ if (statement != null) {
+ statement.cancel();
+ }
}
public void removeStatement(String paragraphId) {
+ if (paragraphId == null) {
+ return;
+ }
paragraphIdStatementMap.remove(paragraphId);
}
diff --git
a/jdbc/src/test/java/org/apache/zeppelin/jdbc/JDBCUserConfigurationsTest.java
b/jdbc/src/test/java/org/apache/zeppelin/jdbc/JDBCUserConfigurationsTest.java
new file mode 100644
index 0000000000..40fa032a6e
--- /dev/null
+++
b/jdbc/src/test/java/org/apache/zeppelin/jdbc/JDBCUserConfigurationsTest.java
@@ -0,0 +1,68 @@
+/**
+ * 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.jdbc;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+class JDBCUserConfigurationsTest {
+
+ @Test
+ void cancelStatementBeforeSaveShouldNotThrowNPE() {
+ JDBCUserConfigurations jdbcUserConfigurations = new
JDBCUserConfigurations();
+
+ assertDoesNotThrow(() ->
jdbcUserConfigurations.cancelStatement("paragraph-not-registered"));
+ }
+
+ @Test
+ void cancelStatementAfterSaveShouldCallCancelOnStatement() throws
SQLException {
+ JDBCUserConfigurations jdbcUserConfigurations = new
JDBCUserConfigurations();
+ Statement statement = Mockito.mock(Statement.class);
+ jdbcUserConfigurations.saveStatement("paragraph-1", statement);
+
+ jdbcUserConfigurations.cancelStatement("paragraph-1");
+
+ verify(statement).cancel();
+ }
+
+ @Test
+ void cancelStatementAfterRemoveShouldNotThrowNPE() throws SQLException {
+ JDBCUserConfigurations jdbcUserConfigurations = new
JDBCUserConfigurations();
+ Statement statement = Mockito.mock(Statement.class);
+ jdbcUserConfigurations.saveStatement("paragraph-1", statement);
+ jdbcUserConfigurations.removeStatement("paragraph-1");
+
+ assertDoesNotThrow(() ->
jdbcUserConfigurations.cancelStatement("paragraph-1"));
+ verify(statement, never()).cancel();
+ }
+
+ @Test
+ void nullParagraphIdShouldBeNoOpAcrossAllMapOperations() throws SQLException
{
+ JDBCUserConfigurations jdbcUserConfigurations = new
JDBCUserConfigurations();
+ Statement statement = Mockito.mock(Statement.class);
+
+ assertDoesNotThrow(() -> jdbcUserConfigurations.saveStatement(null,
statement));
+ assertDoesNotThrow(() -> jdbcUserConfigurations.cancelStatement(null));
+ assertDoesNotThrow(() -> jdbcUserConfigurations.removeStatement(null));
+ verify(statement, never()).cancel();
+ }
+}
diff --git
a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/AbstractScheduler.java
b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/AbstractScheduler.java
index 7e99095b7f..5bb3c82e02 100644
---
a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/AbstractScheduler.java
+++
b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/AbstractScheduler.java
@@ -77,7 +77,11 @@ public abstract class AbstractScheduler implements Scheduler
{
@Override
public Job<?> cancel(String jobId) {
Job<?> job = jobs.remove(jobId);
- job.abort();
+ // Synchronize on the same monitor as runJob()'s abort gate so that a
cancellation
+ // happening right before the job is run is never missed (ZEPPELIN-6129).
+ synchronized (job) {
+ job.abort();
+ }
return job;
}
@@ -121,17 +125,21 @@ public abstract class AbstractScheduler implements
Scheduler {
* @param runningJob
*/
protected void runJob(Job<?> runningJob) {
- if (runningJob.isAborted()) {
- LOGGER.info("Job {} is aborted", runningJob.getId());
- runningJob.setStatus(Job.Status.ABORT);
- runningJob.aborted = false;
- return;
- }
+ // Synchronize the abort gate on the same monitor cancel() uses, so a
cancellation
+ // submitted right before the job runs is never missed (ZEPPELIN-6129).
+ synchronized (runningJob) {
+ if (runningJob.isAborted()) {
+ LOGGER.info("Job {} is aborted", runningJob.getId());
+ runningJob.setStatus(Job.Status.ABORT);
+ runningJob.aborted = false;
+ return;
+ }
- LOGGER.info("Job {} started by scheduler {}", runningJob.getId(), name);
- // Don't set RUNNING status when it is RemoteScheduler, update it via
JobStatusPoller
- if (!getClass().getSimpleName().equals("RemoteScheduler")) {
- runningJob.setStatus(Job.Status.RUNNING);
+ LOGGER.info("Job {} started by scheduler {}", runningJob.getId(), name);
+ // Don't set RUNNING status when it is RemoteScheduler, update it via
JobStatusPoller
+ if (!getClass().getSimpleName().equals("RemoteScheduler")) {
+ runningJob.setStatus(Job.Status.RUNNING);
+ }
}
runningJob.run();
Object jobResult = runningJob.getReturn();
diff --git
a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/Job.java
b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/Job.java
index b0ed600f45..d8b4a739b8 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/Job.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/Job.java
@@ -89,7 +89,7 @@ public abstract class Job<T> {
private Date dateFinished;
protected volatile Status status;
- transient boolean aborted = false;
+ transient volatile boolean aborted = false;
private volatile String errorMessage;
private transient volatile Throwable exception;
private transient JobListener listener;
diff --git
a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/scheduler/AbstractSchedulerAbortRaceTest.java
b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/scheduler/AbstractSchedulerAbortRaceTest.java
new file mode 100644
index 0000000000..e7f10d4451
--- /dev/null
+++
b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/scheduler/AbstractSchedulerAbortRaceTest.java
@@ -0,0 +1,168 @@
+/*
+ * 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.scheduler;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Covers the "abort right before run" race between a cancelling thread
+ * ({@link AbstractScheduler#cancel(String)}) and the scheduler thread that is
about to invoke
+ * {@link AbstractScheduler#runJob(Job)}.
+ *
+ * <p>Honest limitation: a pure memory-visibility race on a non-volatile field
cannot be
+ * reproduced deterministically without a specialized harness (e.g. jcstress)
because it depends
+ * on JVM safepoints/JIT reordering. This class therefore verifies the
observable behavior
+ * contract instead: (1) abort()/isAborted() agree, (2) a PENDING job aborted
before runJob() is
+ * invoked never has its run() executed and ends in ABORT, and (3) cancel()
and the runJob() gate
+ * are mutually exclusive on the same job monitor, which is a deterministic,
latch-driven proof
+ * that the race window described in ZEPPELIN-6129 Task 2 is closed.
+ */
+class AbstractSchedulerAbortRaceTest {
+
+ private FIFOScheduler scheduler;
+
+ @AfterEach
+ void tearDown() {
+ if (scheduler != null) {
+ scheduler.stop();
+ }
+ }
+
+ @Test
+ void testAbortSetsIsAbortedTrue() {
+ SleepingJob job = new SleepingJob("abortJob", null, 5000);
+
+ assertFalse(job.isAborted());
+ job.abort();
+
+ assertTrue(job.isAborted());
+ }
+
+ @Test
+ void testCancelBeforeRunJobBlocksExecutionThroughSchedulerCancelPath() {
+ scheduler = new FIFOScheduler("cancel-gate-test");
+ SleepingJob job = new SleepingJob("job1", null, 5000);
+ scheduler.submit(job);
+
+ scheduler.cancel(job.getId());
+ scheduler.runJob(job);
+
+ assertEquals(Job.Status.ABORT, job.getStatus());
+ assertNull(job.getReturn());
+ }
+
+ @Test
+ void testCancelAndRunJobGateAreMutuallyExclusiveOnJobMonitor() throws
Exception {
+ scheduler = new FIFOScheduler("mutex-test");
+ BlockingAbortJob job = new BlockingAbortJob("job1");
+ scheduler.submit(job);
+
+ Thread cancelThread = new Thread(() -> scheduler.cancel(job.getId()),
"cancel-thread");
+ cancelThread.start();
+
+ assertTrue(job.abortEntered.await(2, TimeUnit.SECONDS),
+ "cancel thread must reach jobAbort() and hold the job monitor");
+
+ Thread runJobThread = new Thread(() -> scheduler.runJob(job),
"runjob-thread");
+ runJobThread.start();
+
+ assertTrue(waitForState(runJobThread, Thread.State.BLOCKED, 2000),
+ "runJob() must block waiting for the same job monitor held by
cancel()");
+ assertFalse(job.runCalled, "job.run() must not start while cancel() still
holds the monitor");
+
+ job.releaseAbort.countDown();
+ cancelThread.join(2000);
+ runJobThread.join(2000);
+
+ assertFalse(job.runCalled, "aborted job must never invoke run()");
+ assertEquals(Job.Status.ABORT, job.getStatus());
+ }
+
+ private static boolean waitForState(Thread thread, Thread.State expected,
long timeoutMs)
+ throws InterruptedException {
+ long deadline = System.currentTimeMillis() + timeoutMs;
+ while (System.currentTimeMillis() < deadline) {
+ if (thread.getState() == expected) {
+ return true;
+ }
+ Thread.sleep(10);
+ }
+ return thread.getState() == expected;
+ }
+
+ /**
+ * Job whose {@code jobAbort()} blocks on a latch so the test can control
exactly how long the
+ * cancelling thread holds the job monitor.
+ */
+ private static class BlockingAbortJob extends Job<Object> {
+
+ private final CountDownLatch abortEntered = new CountDownLatch(1);
+ private final CountDownLatch releaseAbort = new CountDownLatch(1);
+ private volatile boolean runCalled = false;
+
+ BlockingAbortJob(String name) {
+ super(name, null);
+ }
+
+ @Override
+ protected Object jobRun() {
+ runCalled = true;
+ return null;
+ }
+
+ @Override
+ protected boolean jobAbort() {
+ abortEntered.countDown();
+ try {
+ releaseAbort.await(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return true;
+ }
+
+ @Override
+ public void setResult(Object result) {
+ }
+
+ @Override
+ public Object getReturn() {
+ return null;
+ }
+
+ @Override
+ public int progress() {
+ return 0;
+ }
+
+ @Override
+ public Map<String, Object> info() {
+ return Collections.emptyMap();
+ }
+ }
+}
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java
index e5807877f9..c47afa8094 100644
---
a/zeppelin-server/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java
@@ -17,6 +17,8 @@
package org.apache.zeppelin.scheduler;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreter;
import org.apache.zeppelin.scheduler.Job.Status;
import org.apache.zeppelin.util.ExecutorUtil;
@@ -36,17 +38,57 @@ import java.util.concurrent.atomic.AtomicBoolean;
public class RemoteScheduler extends AbstractScheduler {
private static final Logger LOGGER =
LoggerFactory.getLogger(RemoteScheduler.class);
+ private static final String PARAGRAPH_POOL_SIZE_KEY =
+ ConfVars.ZEPPELIN_INTERPRETER_CONNECTION_POOL_SIZE.getVarName();
+ private static final int DEFAULT_PARAGRAPH_POOL_SIZE =
+ ConfVars.ZEPPELIN_INTERPRETER_CONNECTION_POOL_SIZE.getIntValue();
+
private final RemoteInterpreter remoteInterpreter;
private final ExecutorService executor;
public RemoteScheduler(String name,
RemoteInterpreter remoteInterpreter) {
super(name);
- this.executor =
- Executors.newSingleThreadExecutor(new NamedThreadFactory("FIFO-" +
name));
+ this.executor = createExecutor(name, remoteInterpreter);
this.remoteInterpreter = remoteInterpreter;
}
+ /**
+ * Creates the server-side job submission pool. This pool only decides how
many jobs can be
+ * submitted to the remote interpreter process concurrently; actual
concurrency is still
+ * governed by the remote interpreter's own {@code Scheduler} (Parallel vs
FIFO), so this pool
+ * must stay interpreter-neutral.
+ *
+ * <p>"note" execution mode keeps a single-threaded pool because {@link
#runJobInScheduler}
+ * blocks until each job fully finishes before submitting the next one,
preserving in-note
+ * paragraph ordering. "paragraph" mode uses a bounded fixed pool sized from
+ * {@link ConfVars#ZEPPELIN_INTERPRETER_CONNECTION_POOL_SIZE} so any
interpreter whose remote
+ * scheduler is a ParallelScheduler can actually run jobs concurrently.
+ */
+ private static ExecutorService createExecutor(String name, RemoteInterpreter
remoteInterpreter) {
+ String executionMode = remoteInterpreter.getProperty(".execution.mode",
"paragraph");
+ if (!"paragraph".equals(executionMode)) {
+ return Executors.newSingleThreadExecutor(new NamedThreadFactory("FIFO-"
+ name));
+ }
+ int poolSize = resolveParagraphPoolSize(remoteInterpreter);
+ return Executors.newFixedThreadPool(poolSize, new
NamedThreadFactory("FIFO-" + name));
+ }
+
+ private static int resolveParagraphPoolSize(RemoteInterpreter
remoteInterpreter) {
+ String value = remoteInterpreter.getProperty(PARAGRAPH_POOL_SIZE_KEY);
+ if (StringUtils.isBlank(value)) {
+ return DEFAULT_PARAGRAPH_POOL_SIZE;
+ }
+ try {
+ int parsed = Integer.parseInt(value.trim());
+ return parsed > 0 ? parsed : DEFAULT_PARAGRAPH_POOL_SIZE;
+ } catch (NumberFormatException e) {
+ LOGGER.warn("Invalid {} value: {}, falling back to default {}",
+ PARAGRAPH_POOL_SIZE_KEY, value, DEFAULT_PARAGRAPH_POOL_SIZE);
+ return DEFAULT_PARAGRAPH_POOL_SIZE;
+ }
+ }
+
@Override
public void runJobInScheduler(Job<?> job) {
JobRunner jobRunner = new JobRunner(this, job);
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/scheduler/RemoteSchedulerTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/scheduler/RemoteSchedulerTest.java
index 2eb9afe976..14fbdd0970 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/scheduler/RemoteSchedulerTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/scheduler/RemoteSchedulerTest.java
@@ -32,6 +32,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -44,6 +46,8 @@ class RemoteSchedulerTest extends AbstractInterpreterTest {
private SchedulerFactory schedulerSvc;
private static final int TICK_WAIT = 100;
private static final int MAX_WAIT_CYCLES = 100;
+ private static final int CONCURRENT_JOB_SLEEP_MS = 3000;
+ private static final int OVERLAP_WAIT_CYCLES = 30;
private String note1Id;
@Override
@@ -132,8 +136,15 @@ class RemoteSchedulerTest extends AbstractInterpreterTest {
}
@Test
- void testAbortOnPending() throws Exception {
+ void testAbortOnPending_noteModeSerial() throws Exception {
final RemoteInterpreter intpA = (RemoteInterpreter)
interpreterSetting.getInterpreter("user1", note1Id, "mock");
+ // Force "note" execution mode: RemoteScheduler keeps a single-threaded
pool for it and its
+ // local dispatch gate (runJobInScheduler) blocks until job1 is fully
executed - not just
+ // submitted - before even attempting job2. So job2 is deterministically
still PENDING, and
+ // never dispatched, when it is aborted below, regardless of the
paragraph-mode pool now
+ // being multi-threaded for every interpreter (ZEPPELIN-6129).
+ intpA.setProperty(".execution.mode", "note");
+ intpA.setProperty(".noteId", note1Id);
intpA.open();
Scheduler scheduler = intpA.getScheduler();
@@ -237,23 +248,48 @@ class RemoteSchedulerTest extends AbstractInterpreterTest
{
scheduler.submit(job1);
scheduler.submit(job2);
+ CountDownLatch job1Running = new CountDownLatch(1);
+ Thread runningWatcher = new Thread(() -> {
+ int cycles = 0;
+ while (job1Running.getCount() > 0 && cycles < MAX_WAIT_CYCLES) {
+ if (job1.isRunning()) {
+ job1Running.countDown();
+ return;
+ }
+ try {
+ Thread.sleep(TICK_WAIT);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ cycles++;
+ }
+ });
+ runningWatcher.start();
+
+ assertTrue(job1Running.await(MAX_WAIT_CYCLES * TICK_WAIT,
TimeUnit.MILLISECONDS),
+ "job1 should reach RUNNING");
+ runningWatcher.join(TICK_WAIT);
- int cycles = 0;
- while (!job1.isRunning() && cycles < MAX_WAIT_CYCLES) {
- Thread.sleep(TICK_WAIT);
- cycles++;
- }
assertTrue(job1.isRunning());
assertEquals(Status.PENDING, job2.getStatus());
job2.abort();
- cycles = 0;
+ int cycles = 0;
while (!job1.isTerminated() && cycles < MAX_WAIT_CYCLES) {
Thread.sleep(TICK_WAIT);
cycles++;
}
+ // job1 terminating only unblocks the scheduler thread to dequeue and
abort job2; give it
+ // its own bounded wait instead of assuming it is already processed the
instant job1 is done.
+ cycles = 0;
+ while (!job2.isTerminated() && cycles < MAX_WAIT_CYCLES) {
+ Thread.sleep(TICK_WAIT);
+ cycles++;
+ }
+
assertNotNull(job1.getDateFinished());
assertTrue(job1.isTerminated());
assertEquals("1000", job1.getReturn());
@@ -265,4 +301,93 @@ class RemoteSchedulerTest extends AbstractInterpreterTest {
schedulerSvc.removeScheduler("test");
}
+ @Test
+ void testParallelExecution_bothJobsRunConcurrently() throws Exception {
+ final RemoteInterpreter intpA =
+ (RemoteInterpreter) interpreterSetting.getInterpreter("user1",
note1Id, "mock");
+ // enable parallel execution on the remote interpreter side so that the
two jobs
+ // are not serialized by the interpreter's own scheduler. RemoteScheduler
itself must
+ // stay interpreter-neutral: no JDBC-specific property is needed to unlock
concurrency.
+ intpA.setProperty("parallel", "true");
+ intpA.open();
+
+ Scheduler scheduler = intpA.getScheduler();
+
+ Job<Object> job1 = createSleepingJob("jobId1", intpA,
CONCURRENT_JOB_SLEEP_MS);
+ Job<Object> job2 = createSleepingJob("jobId2", intpA,
CONCURRENT_JOB_SLEEP_MS);
+
+ scheduler.submit(job1);
+ scheduler.submit(job2);
+
+ CountDownLatch overlapDetected = new CountDownLatch(1);
+ Thread overlapWatcher = new Thread(() -> {
+ int cycles = 0;
+ while (overlapDetected.getCount() > 0 && cycles < OVERLAP_WAIT_CYCLES) {
+ if (job1.isRunning() && job2.isRunning()) {
+ overlapDetected.countDown();
+ return;
+ }
+ try {
+ Thread.sleep(TICK_WAIT);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ cycles++;
+ }
+ });
+ overlapWatcher.start();
+
+ boolean bothRanConcurrently =
+ overlapDetected.await(OVERLAP_WAIT_CYCLES * TICK_WAIT,
TimeUnit.MILLISECONDS);
+ overlapWatcher.join(TICK_WAIT);
+
+ assertTrue(bothRanConcurrently, "job1 and job2 should both be RUNNING at
the same time");
+
+ intpA.close();
+ schedulerSvc.removeScheduler("test");
+ }
+
+ private Job<Object> createSleepingJob(String jobId, RemoteInterpreter intpA,
int sleepMillis) {
+ return new Job<Object>(jobId, jobId, null) {
+ Object results;
+ InterpreterContext context = InterpreterContext.builder()
+ .setNoteId("noteId")
+ .setParagraphId(jobId)
+ .setResourcePool(new LocalResourcePool("pool-" + jobId))
+ .build();
+
+ @Override
+ public Object getReturn() {
+ return results;
+ }
+
+ @Override
+ public int progress() {
+ return 0;
+ }
+
+ @Override
+ public Map<String, Object> info() {
+ return null;
+ }
+
+ @Override
+ protected Object jobRun() throws Throwable {
+ intpA.interpret(String.valueOf(sleepMillis), context);
+ return String.valueOf(sleepMillis);
+ }
+
+ @Override
+ protected boolean jobAbort() {
+ return false;
+ }
+
+ @Override
+ public void setResult(Object results) {
+ this.results = results;
+ }
+ };
+ }
+
}
