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 311fb0bad0c..bcaea51e0ca 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 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 void setUserProperty(UsernamePassword usernamePassword) { } 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 00000000000..40fa032a6e4 --- /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 7e99095b7ff..5bb3c82e020 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 void submit(Job job) { @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 void stop() { * @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 b0ed600f45a..d8b4a739b87 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 boolean isFailed() { 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 00000000000..e7f10d44514 --- /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)}. + * + *

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 { + + 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 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 e5807877f92..c47afa8094b 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 @@ 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. + * + *

"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 2eb9afe9768..14fbdd09701 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.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 @@ public void setResult(Object results) { } @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 @@ public void setResult(Object results) { 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 @@ public void setResult(Object results) { 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 job1 = createSleepingJob("jobId1", intpA, CONCURRENT_JOB_SLEEP_MS); + Job 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 createSleepingJob(String jobId, RemoteInterpreter intpA, int sleepMillis) { + return new Job(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 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; + } + }; + } + }