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 3d71dfef07 [ZEPPELIN-6555] Avoid deadlock in
ManagedInterpreterGroup.close() by not holding the group lock while closing
session interpreters
3d71dfef07 is described below
commit 3d71dfef07cbef158e5544c46783cdec0cb32290
Author: HyeonUk Kang <[email protected]>
AuthorDate: Sun Aug 9 22:58:29 2026 +0900
[ZEPPELIN-6555] Avoid deadlock in ManagedInterpreterGroup.close() by not
holding the group lock while closing session interpreters
### What is this PR for?
`ManagedInterpreterGroup.close()` can deadlock with a concurrent
`RemoteInterpreter.open()` because the two paths take the same two monitors in
the opposite order.
Two monitors are involved: the interpreter-group monitor and an individual
interpreter's monitor.
- `open()` takes the interpreter monitor first (`synchronized(this)`) and
then the group monitor (via `getOrCreateSession()` and the angular-registry
push). Order: interpreter → group.
- `close(String)` is a `synchronized` method, so it holds the group monitor
while it spawns the per-interpreter close threads and `join()`s them. Each
close thread runs `interpreter.close()`, which takes the interpreter monitor.
Order: group → interpreter.
So when an interpreter is opened while its session is concurrently closed
(for example, restarting or shutting down an interpreter while a paragraph on
that session is still starting up), the two orders form a circular wait.
### What type of PR is it?
Bug Fix
### Todos
* [x] Stop holding the group monitor while closing session interpreters in
`close(String)`
* [x] Keep only the session-map removal and last-session teardown under the
lock
* [x] Add regression test
`ManagedInterpreterGroupTest#close_doesNotDeadlockWithConcurrentOpen`
### What is the Jira issue?
[[ZEPPELIN-6555]](https://issues.apache.org/jira/browse/ZEPPELIN-6555)
### How should this be tested?
- Added a deterministic regression
test,`ManagedInterpreterGroupTest#close_doesNotDeadlockWithConcurrentOpen`.
### Screenshots (if appropriate)
### Questions:
* Does the license files need to update? - No
* Is there breaking changes for older versions? - No
* Does this needs documentation? - No
Closes #5329 from hyunw9/ZEPPELIN-6555.
Signed-off-by: Jongyoul Lee <[email protected]>
---
.../interpreter/ManagedInterpreterGroup.java | 35 ++++---
.../interpreter/ManagedInterpreterGroupTest.java | 109 +++++++++++++++++++++
2 files changed, 130 insertions(+), 14 deletions(-)
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
index 8f2c16c074..f3f5441319 100644
---
a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
@@ -101,22 +101,29 @@ public class ManagedInterpreterGroup extends
InterpreterGroup {
* Close all interpreter instances in this session
* @param sessionId
*/
- public synchronized void close(String sessionId) {
- LOGGER.info("Close Session: {} for interpreter setting: {}", sessionId,
interpreterSetting.getName());
- close(sessions.remove(sessionId));
+ public void close(String sessionId) {
+ LOGGER.info("Close Session: {} for interpreter setting: {}",
+ sessionId, interpreterSetting.getName());
+
+ Collection<Interpreter> interpreters = sessions.remove(sessionId);
+ close(interpreters);
+
//TODO(zjffdu) whether close InterpreterGroup if there's no session left
in Zeppelin Server
- if (sessions.isEmpty() && interpreterSetting != null) {
- LOGGER.info("Remove this InterpreterGroup: {} as all the sessions are
closed", id);
- interpreterSetting.removeInterpreterGroup(id);
- if (remoteInterpreterProcess != null) {
- LOGGER.info("Kill RemoteInterpreterProcess");
- remoteInterpreterProcess.stop();
- try {
-
interpreterSetting.getRecoveryStorage().onInterpreterClientStop(remoteInterpreterProcess);
- } catch (IOException e) {
- LOGGER.error("Fail to store recovery data", e);
+ synchronized (this) {
+ if (sessions.isEmpty() && interpreterSetting != null) {
+ LOGGER.info("Remove this InterpreterGroup: {} as all the sessions are
closed", id);
+ interpreterSetting.removeInterpreterGroup(id);
+ if (remoteInterpreterProcess != null) {
+ LOGGER.info("Kill RemoteInterpreterProcess");
+ remoteInterpreterProcess.stop();
+ try {
+ interpreterSetting.getRecoveryStorage()
+ .onInterpreterClientStop(remoteInterpreterProcess);
+ } catch (IOException e) {
+ LOGGER.error("Fail to store recovery data", e);
+ }
+ remoteInterpreterProcess = null;
}
- remoteInterpreterProcess = null;
}
}
}
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroupTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroupTest.java
index 09a8974672..c1bff6e264 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroupTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroupTest.java
@@ -21,13 +21,20 @@ import org.junit.jupiter.api.BeforeEach;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.eclipse.aether.RepositoryException;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import java.io.IOException;
+import java.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
+import java.util.Properties;
+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.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
class ManagedInterpreterGroupTest {
@@ -89,4 +96,106 @@ class ManagedInterpreterGroupTest {
interpreterGroup.close();
assertEquals(0, interpreterGroup.getSessionNum());
}
+
+ @Test
+ @Timeout(30)
+ void close_doesNotDeadlockWithConcurrentOpen() throws Exception {
+ ManagedInterpreterGroup group =
+ new ManagedInterpreterGroup("g1", interpreterSetting, zConf);
+
+ LockProbeInterpreter probe = new LockProbeInterpreter(new Properties());
+ probe.setInterpreterGroup(group);
+ List<Interpreter> s1 = new ArrayList<>();
+ s1.add(probe);
+ group.sessions.put("s1", s1);
+
+ CountDownLatch openerHasIntp = new CountDownLatch(1);
+
+ // "opener": mirrors open() lock ordering (interpreter -> group).
+ Thread opener = new Thread(() -> {
+ synchronized (probe) { // interpreter monitor
+ openerHasIntp.countDown();
+ try {
+ // wait until the close worker is blocked trying to take the
interpreter monitor, which
+ // means the closer is holding the group monitor inside
close(String).
+ probe.closeReached.await();
+ Thread w;
+ while ((w = probe.worker) == null || w.getState() !=
Thread.State.BLOCKED) {
+ Thread.onSpinWait();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ group.getOrCreateSession("u", "s2"); // needs the group monitor
+ }
+ }, "deadlock-opener");
+ opener.setDaemon(true);
+
+ // "closer": the real method under test.
+ Thread closer = new Thread(() -> group.close("s1"), "deadlock-closer");
+ closer.setDaemon(true);
+
+ opener.start();
+ assertTrue(openerHasIntp.await(5, TimeUnit.SECONDS),
+ "opener failed to acquire the interpreter monitor");
+ closer.start();
+
+ opener.join(TimeUnit.SECONDS.toMillis(10));
+ closer.join(TimeUnit.SECONDS.toMillis(10));
+
+ if (opener.isAlive() || closer.isAlive()) {
+ long[] deadlocked =
ManagementFactory.getThreadMXBean().findDeadlockedThreads();
+ fail("Deadlock: ManagedInterpreterGroup.close() holds the group monitor
while joining the "
+ + "close-worker thread, which needs the interpreter monitor held by
the concurrent "
+ + "open(). opener.alive=" + opener.isAlive() + ", closer.alive=" +
closer.isAlive()
+ + ", jvmDetectedMonitorDeadlock=" + (deadlocked != null));
+ }
+ }
+
+ /**
+ * Minimal interpreter whose close() takes its own monitor, like
RemoteInterpreter does via
+ * getOrCreateInterpreterProcess(). It signals when the close worker reaches
the monitor so the
+ * test can force the interleaving deterministically.
+ */
+ private static class LockProbeInterpreter extends Interpreter {
+
+ final CountDownLatch closeReached = new CountDownLatch(1);
+ volatile Thread worker;
+
+ LockProbeInterpreter(Properties properties) {
+ super(properties);
+ }
+
+ @Override
+ public void close() {
+ worker = Thread.currentThread();
+ closeReached.countDown();
+ synchronized (this) {
+ }
+ }
+
+ @Override
+ public void open() {
+ }
+
+ @Override
+ public InterpreterResult interpret(String st, InterpreterContext context) {
+ return null;
+ }
+
+ @Override
+ public void cancel(InterpreterContext context) {
+ }
+
+ @Override
+ public FormType getFormType() {
+ return FormType.NATIVE;
+ }
+
+ @Override
+ public int getProgress(InterpreterContext context) {
+ return 0;
+ }
+ }
}
