You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zeppelin.apache.org by jo...@apache.org on 2026/08/10 00:21:24 UTC
(zeppelin) branch master updated: [ZEPPELIN-6575] Reclaim idle interpreters on the server with a per-setting timeout
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 4d02c15e51 [ZEPPELIN-6575] Reclaim idle interpreters on the server with a per-setting timeout
4d02c15e51 is described below
commit 4d02c15e510d8e9874f40284e0008994d56fede1
Author: dae won <99...@users.noreply.github.com>
AuthorDate: Mon Aug 10 09:21:18 2026 +0900
[ZEPPELIN-6575] Reclaim idle interpreters on the server with a per-setting timeout
### What is this PR for?
An interpreter setting cannot have its own idle timeout today, and worse, trying to give it one fails silently.
Idle reclaim is decided inside the interpreter process by `TimeoutLifecycleManager`, which reads its threshold from the configuration map the server pushes over Thrift when the process starts. That map is built by `ZeppelinConfiguration#getCompleteConfiguration()`:
```java
for (ConfVars c : ConfVars.values()) {
if (getString(c) != null) {
completeConfiguration.put(c.getVarName(), getString(c));
}
}
```
Its key set is closed over the `ConfVars` enum, so an interpreter setting property has no slot to travel in. An operator can put the threshold property on a single interpreter's settings, it is stored, it is shown again when the form is reopened - and the process still starts with the global value. No error, no warning.
So the threshold is effectively all-or-nothing across every interpreter, which does not match how they differ in cost. A Spark interpreter holding tens of gigabytes of cluster memory is worth reclaiming aggressively; a JDBC interpreter that only keeps a few connections open is usually worth keeping. Today an operator who enables idle reclaim gets one number for both.
This moves the decision to the server, which already knows the per-setting value:
- `ManagedInterpreterGroup` records when the group was last used. The three hooks mirror the ones `RemoteInterpreterServer` already calls in-process (`interpret`, `getProgress`, `getStatus`), so no new notion of activity is introduced. `getStatus` matters most: `RemoteScheduler.JobStatusPoller` calls it while a paragraph runs, which is what keeps a long-running paragraph from having its interpreter pulled out from under it.
- `IdleInterpreterReclaimer` walks `getAllInterpreterGroup()` on a timer, resolves the threshold from the owning interpreter setting (falling back to the global property), and closes the group by reusing `ManagedInterpreterGroup.close()`. It reads in-memory state only and never calls `isAlive()`/`isRunning()`, whose cost depends on the launcher - a socket connect with a 1s timeout for docker, an unbounded kube-apiserver round trip for k8s - and this runs over every group on a timer.
- A group whose process is still launching is skipped. The handle field is assigned before `start()`, while the idle clock has been running since the group was created, so without this a launch slower than the threshold gets killed while coming up. Spark on YARN takes minutes to launch, so this is not hypothetical.
**No new configuration property, and nothing changes unless asked for.** This follows the existing lifecycle manager class property, which already expresses whether idle reclaim is wanted at all. Its `NullLifecycleManager` default means an existing deployment sees no change whatsoever; `TimeoutLifecycleManager` now also enables server-driven reclaim. Any other implementation is left untouched. `ZeppelinConfiguration#getLifecycleManagerClass()` had no caller before this change.
An interpreter setting overrides the threshold by carrying the same property name, where `0` or below means never reclaimed. This is opt-in per interpreter: a setting that carries nothing keeps following the global threshold exactly as before, so the JDBC interpreter above is still reclaimed on the global schedule until an operator marks it as exempt. What the change adds is the ability to say it at all.
**`TimeoutLifecycleManager` is kept, not replaced.** It still runs in the interpreter process, because that is what shuts a process down if the Zeppelin server itself exits unexpectedly and can no longer reclaim anything. It is handed the same threshold resolved here rather than the global one, so the two sides agree instead of one of them shutting a process down on the wrong schedule. Closing a group twice cannot happen either: if the process shuts itself down it unregisters, which r [...]
That fallback is intentionally given up for one case only - a setting the operator marked as never reclaimed (`0`), whose process receives `Long.MAX_VALUE` and therefore will not self-terminate either. `TimeoutLifecycleManager` has no way to express "never" and would read a threshold of `0` as "shut down at the next check".
### What type of PR is it?
Feature
### Todos
* [x] - Track last-used time per interpreter group on the server
* [x] - Close groups idle beyond the threshold, reusing `ManagedInterpreterGroup.close()`
* [x] - Resolve the threshold per interpreter setting, falling back to the global property
* [x] - Keep the in-process fallback consistent by pushing the resolved threshold to the process
* [x] - Skip groups whose process is still launching
* [x] - Unit tests, including guards against probing and against reclaiming a launching group
* [x] - Document the per-interpreter threshold in `docs/usage/interpreter/overview.md` and `conf/zeppelin-site.xml.template`
### What is the Jira issue?
* [ZEPPELIN-6575](https://issues.apache.org/jira/browse/ZEPPELIN-6575), a sub-task of [ZEPPELIN-6568](https://issues.apache.org/jira/browse/ZEPPELIN-6568)
### How should this be tested?
`IdleInterpreterReclaimerTest` (9 tests) covers both directions of the override, the launching guard, the no-probe guard, threshold resolution, and that the default lifecycle manager changes nothing.
```bash
export JAVA_HOME=$(/usr/libexec/java_home -v 11)
./mvnw package -pl zeppelin-server --am \
-Dtest=IdleInterpreterReclaimerTest,TimeoutLifecycleManagerTest -DfailIfNoTests=false
```
The two override tests fail before the change, because the per-setting value never reaches the process:
```
perSettingThresholdReclaimsEarlierThanTheGlobalOne
the group should be reclaimed after the per setting threshold of 10s
==> expected: <0> but was: <1>
perSettingThresholdCanOptOutOfAShortGlobalThreshold
the setting opted out of reclaim, so the short global threshold must not apply
==> expected: <1> but was: <0>
```
`aGroupBeingLaunchedIsNotReclaimed` fails without the launching guard, and `scanNeverProbesTheInterpreterProcess` fails if the scan is written with `isAlive()`/`isRunning()`.
Local runs, all passing:
| Scope | Result |
|---|---|
| `org.apache.zeppelin.interpreter.**` (19 classes, includes recovery and launcher tests) | pass |
| `notebook`, `rest`, `service`, `socket`, `server`, `notebook.repo` (39 classes, 318 tests) | pass |
| `TimeoutLifecycleManagerTest` (existing idle reclaim behaviour) | pass, no regression |
| `./mvnw clean org.apache.rat:apache-rat-plugin:check -Prat` | `Unapproved: 0` |
Manual steps on a running server. This uses two of the lightweight built-in interpreters, `md` (markdown) and `sh` (shell), so that two interpreters run side by side under one server and one global threshold:
1. In `zeppelin-site.xml` set the lifecycle manager class to `TimeoutLifecycleManager`, the global threshold to `10s`, and the check interval to `5s`.
2. On the `md` interpreter setting only, add the threshold property with the value `0` to mark it as never reclaimed. Leave the `sh` setting untouched so that it follows the global 10s.
3. Run one `%md` paragraph and one `%sh` paragraph, then leave the note idle.
Each interpreter setting gets its own process, and each process logs the threshold it was handed:
```
logs/zeppelin-interpreter-md-shared_process-*.log
TimeoutLifecycleManager is started with checkInterval: 5000, timeoutThreshold: 9223372036854775807
logs/zeppelin-interpreter-sh-shared_process-*.log
TimeoutLifecycleManager is started with checkInterval: 5000, timeoutThreshold: 10000
```
Before this change both processes were handed the global `10000` and both were shut down after 10s of idle time; the `0` on the `md` setting had no effect at all. Now only `sh` is reclaimed, and the server log records why:
```
logs/zeppelin-*.log
Reclaiming interpreter group sh-shared_process of interpreter setting sh:
idle for 11603ms which exceeds its threshold of 10000ms
```
`ps` confirmed the `sh` process was gone about 15s after its last use, while the `md` process was still running after 33s of idle time. Both processes kept the lifecycle manager class the operator configured; only the threshold they received differed.
Not verified locally, left to CI and to deployments that have the runtimes: the docker, k8s and yarn launchers. The reclaim path does not call into a launcher - it neither probes nor launches, only closes - so the exposure is limited to `close()`, which the existing restart endpoint already uses.
### Screenshots (if appropriate)
N/A
### Questions:
* Does the license files need to update? No. The two new files carry the ASF header and `apache-rat-plugin:check` reports no unapproved files.
* Is there breaking changes for older versions? No. With the default `NullLifecycleManager` nothing is scheduled and no configuration is overridden for the interpreter process, so an untouched deployment behaves exactly as before.
* Does this needs documentation? Yes, and it is included. `docs/usage/interpreter/overview.md` gains a "Per interpreter idle threshold" section, and the descriptions in `conf/zeppelin-site.xml.template` are extended. No values in the template were changed.
Closes #5358 from big-cir/ZEPPELIN-6575.
Signed-off-by: Jongyoul Lee <jo...@gmail.com>
---
conf/zeppelin-site.xml.template | 8 +-
docs/usage/interpreter/overview.md | 23 ++
.../zeppelin/conf/ZeppelinConfiguration.java | 48 +++
.../interpreter/InterpreterSettingManager.java | 11 +
.../interpreter/ManagedInterpreterGroup.java | 56 +++-
.../lifecycle/IdleInterpreterReclaimer.java | 204 +++++++++++++
.../interpreter/remote/RemoteInterpreter.java | 16 +
.../remote/RemoteInterpreterProcess.java | 17 +-
.../lifecycle/IdleInterpreterReclaimerTest.java | 337 +++++++++++++++++++++
9 files changed, 708 insertions(+), 12 deletions(-)
diff --git a/conf/zeppelin-site.xml.template b/conf/zeppelin-site.xml.template
index d5e54b91f1..d04aee833e 100755
--- a/conf/zeppelin-site.xml.template
+++ b/conf/zeppelin-site.xml.template
@@ -576,7 +576,9 @@
<name>zeppelin.interpreter.lifecyclemanager.class</name>
<value>org.apache.zeppelin.interpreter.lifecycle.TimeoutLifecycleManager</value>
<description>LifecycleManager class for managing the lifecycle of interpreters, by default interpreter will
- be closed after timeout</description>
+ be closed after timeout. With TimeoutLifecycleManager, Zeppelin server tracks the last use of each
+ interpreter group and closes the idle ones itself, so the threshold below can be overridden per
+ interpreter setting</description>
</property>
<property>
@@ -588,7 +590,9 @@
<property>
<name>zeppelin.interpreter.lifecyclemanager.timeout.threshold</name>
<value>1h</value>
- <description>Interpreter timeout threshold, by default it is 1 hour</description>
+ <description>Interpreter timeout threshold, by default it is 1 hour. Set the same property on an
+ individual interpreter setting to override it for that interpreter only, or set it to 0 there to
+ keep that interpreter from ever being reclaimed</description>
</property>
-->
diff --git a/docs/usage/interpreter/overview.md b/docs/usage/interpreter/overview.md
index c664d7ac24..862fb69074 100644
--- a/docs/usage/interpreter/overview.md
+++ b/docs/usage/interpreter/overview.md
@@ -115,6 +115,29 @@ Before 0.8.0, Zeppelin doesn't have lifecycle management for interpreters. Users
`NullLifecycleManager` will do nothing, i.e., the user needs to control the lifecycle of interpreter by themselves as before. `TimeoutLifecycleManager` will shut down interpreters after an interpreter remains idle for a while. By default, the idle threshold is 1 hour.
Users can change this threshold via the `zeppelin.interpreter.lifecyclemanager.timeout.threshold` setting. `NullLifecycleManager` is the default lifecycle manager, and users can change it via `zeppelin.interpreter.lifecyclemanager.class`.
+### Per interpreter idle threshold
+
+One global threshold is not always enough: a Spark interpreter holding tens of gigabytes of cluster
+memory is worth reclaiming quickly, while a JDBC interpreter that only keeps a few connections open
+is usually worth keeping. With `TimeoutLifecycleManager` configured, Zeppelin server keeps track of
+when each interpreter group was last used and closes the idle ones itself, which lets an individual
+interpreter setting override the global value.
+
+To do so, add `zeppelin.interpreter.lifecyclemanager.timeout.threshold` as a property of that
+interpreter on the interpreter setting page, or in `interpreter.json`. The value accepts the same
+formats as the global one, i.e. a plain number of milliseconds or a unit suffix such as `10m`:
+
+| Value on an interpreter setting | Effect on that interpreter |
+|---|---|
+| not set | the global `zeppelin.interpreter.lifecyclemanager.timeout.threshold` applies |
+| `10m` | it is shut down after 10 minutes of idle time, whatever the global value is |
+| `0` | it is never shut down for being idle |
+
+A paragraph that is still running keeps its interpreter alive regardless of the threshold. The check
+runs every `zeppelin.interpreter.lifecyclemanager.timeout.checkinterval`, which is shared with the
+global behaviour, so an interpreter can be shut down up to one interval after its threshold has
+passed.
+
## Inline Generic Configuration
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 179dcce6e8..b2e15160b5 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
@@ -780,6 +780,38 @@ public class ZeppelinConfiguration {
return getString(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS);
}
+ /**
+ * Shared with {@code TimeoutLifecycleManager} so that both ways of reclaiming an idle
+ * interpreter check at the same cadence.
+ *
+ * @return interval in milliseconds between two idle checks
+ */
+ public long getInterpreterIdleCheckInterval() {
+ return getTimeMillis(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_CHECK_INTERVAL);
+ }
+
+ /**
+ * Global idle threshold, which an interpreter setting can override with its own
+ * {@code zeppelin.interpreter.lifecyclemanager.timeout.threshold} property.
+ *
+ * @return threshold in milliseconds
+ */
+ public long getInterpreterIdleTimeoutThreshold() {
+ return getTimeMillis(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD);
+ }
+
+ /**
+ * Reads a time valued property. {@link #getString(ConfVars)} returns null for a ConfVars
+ * declared with a numeric default, so the declared default is used when nothing is configured.
+ */
+ private long getTimeMillis(ConfVars c) {
+ String value = getString(c);
+ if (StringUtils.isBlank(value)) {
+ return c.getLongValue();
+ }
+ return parseTimeMillis(value);
+ }
+
public boolean getZeppelinImpersonateSparkProxyUser() {
return getBoolean(ConfVars.ZEPPELIN_IMPERSONATE_SPARK_PROXY_USER);
}
@@ -1239,4 +1271,20 @@ public class ZeppelinConfiguration {
return Duration.parse("PT" + timeStrWithUnit).toMillis();
}
+ /**
+ * Parses a time value that is either a plain millisecond number or carries a unit suffix,
+ * e.g. {@code 600000}, {@code 10m} or {@code 500ms}.
+ *
+ * @throws NumberFormatException if the value carries no unit and is not a number
+ * @throws java.time.format.DateTimeParseException if the unit suffix is not understood
+ */
+ public static long parseTimeMillis(String timeStr) {
+ String trimmed = timeStr.trim();
+ try {
+ return Long.parseLong(trimmed);
+ } catch (NumberFormatException e) {
+ return timeUnitToMill(trimmed);
+ }
+ }
+
}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
index 08d11629ba..8b4c2fe55b 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
@@ -47,6 +47,7 @@ import org.apache.zeppelin.display.AngularObjectRegistry;
import org.apache.zeppelin.display.AngularObjectRegistryListener;
import org.apache.zeppelin.helium.ApplicationEventListener;
import org.apache.zeppelin.interpreter.Interpreter.RegisteredInterpreter;
+import org.apache.zeppelin.interpreter.lifecycle.IdleInterpreterReclaimer;
import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
import org.apache.zeppelin.interpreter.remote.RemoteAngularObjectRegistry;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
@@ -146,6 +147,7 @@ public class InterpreterSettingManager implements NoteEventListener {
private Map<String, String> jupyterKernelLanguageMap = new HashMap<>();
private List<String> includesInterpreters;
private List<String> excludesInterpreters;
+ private final IdleInterpreterReclaimer idleInterpreterReclaimer;
@Inject
public InterpreterSettingManager(ZeppelinConfiguration zConf,
@@ -206,6 +208,14 @@ public class InterpreterSettingManager implements NoteEventListener {
this.configStorage = configStorage;
init();
+
+ this.idleInterpreterReclaimer = new IdleInterpreterReclaimer(zConf, this);
+ this.idleInterpreterReclaimer.start();
+ }
+
+ @VisibleForTesting
+ public IdleInterpreterReclaimer getIdleInterpreterReclaimer() {
+ return idleInterpreterReclaimer;
}
public RemoteInterpreterEventServer getInterpreterEventServer() {
@@ -1121,6 +1131,7 @@ public class InterpreterSettingManager implements NoteEventListener {
}
public void close() {
+ idleInterpreterReclaimer.stop();
List<Thread> closeThreads = interpreterSettings.values().stream()
.map(intpSetting-> new Thread(intpSetting::close, intpSetting.getId() + "-close"))
.peek(t -> t.setUncaughtExceptionHandler((th, e) ->
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 f3f5441319..3a2f78af89 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
@@ -19,6 +19,7 @@
package org.apache.zeppelin.interpreter;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.lifecycle.IdleInterpreterReclaimer;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
import org.apache.zeppelin.scheduler.Job;
import org.apache.zeppelin.scheduler.Scheduler;
@@ -43,6 +44,8 @@ public class ManagedInterpreterGroup extends InterpreterGroup {
private RemoteInterpreterProcess remoteInterpreterProcess; // attached remote interpreter process
private Object interpreterProcessCreationLock = new Object();
private final ZeppelinConfiguration zConf;
+ private volatile long lastUsedTimeInMillis = System.currentTimeMillis();
+ private volatile boolean launchingInterpreterProcess;
/**
* Create InterpreterGroup with given id and interpreterSetting, used in ZeppelinServer
@@ -64,19 +67,54 @@ public class ManagedInterpreterGroup extends InterpreterGroup {
Properties properties)
throws IOException {
synchronized (interpreterProcessCreationLock) {
- if (remoteInterpreterProcess == null) {
- LOGGER.info("Create InterpreterProcess for InterpreterGroup: {}", getId());
- remoteInterpreterProcess = interpreterSetting.createInterpreterProcess(id, userName,
- properties);
- remoteInterpreterProcess.start(userName);
- remoteInterpreterProcess.init(zConf);
- getInterpreterSetting().getRecoveryStorage()
- .onInterpreterClientStart(remoteInterpreterProcess);
+ try {
+ if (remoteInterpreterProcess == null) {
+ LOGGER.info("Create InterpreterProcess for InterpreterGroup: {}", getId());
+ launchingInterpreterProcess = true;
+ remoteInterpreterProcess = interpreterSetting.createInterpreterProcess(id, userName,
+ properties);
+ remoteInterpreterProcess.start(userName);
+ remoteInterpreterProcess.init(zConf,
+ IdleInterpreterReclaimer.processConfigurationOverrides(zConf,
+ interpreterSetting));
+ getInterpreterSetting().getRecoveryStorage()
+ .onInterpreterClientStart(remoteInterpreterProcess);
+ }
+ return remoteInterpreterProcess;
+ } finally {
+ // Reset the idle clock before dropping the flag, so that this group is never momentarily
+ // visible as idle with a timestamp from before the launch.
+ onInterpreterUse();
+ launchingInterpreterProcess = false;
}
- return remoteInterpreterProcess;
}
}
+ /**
+ * A launch takes a while - minutes for Spark on YARN - and counts as activity rather than as
+ * idle time.
+ *
+ * @return whether a process is currently being launched for this group
+ */
+ public boolean isLaunchingInterpreterProcess() {
+ return launchingInterpreterProcess;
+ }
+
+ /**
+ * Records that this group has just been used, so that server side idle reclaim does not consider
+ * it idle. A single volatile write on purpose: while a paragraph runs this is called on every
+ * status poll of that paragraph.
+ *
+ * @see org.apache.zeppelin.interpreter.lifecycle.IdleInterpreterReclaimer
+ */
+ public void onInterpreterUse() {
+ lastUsedTimeInMillis = System.currentTimeMillis();
+ }
+
+ public long getLastUsedTimeInMillis() {
+ return lastUsedTimeInMillis;
+ }
+
public RemoteInterpreterProcess getInterpreterProcess() {
return remoteInterpreterProcess;
}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/lifecycle/IdleInterpreterReclaimer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/lifecycle/IdleInterpreterReclaimer.java
new file mode 100644
index 0000000000..ccb48366b4
--- /dev/null
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/lifecycle/IdleInterpreterReclaimer.java
@@ -0,0 +1,204 @@
+/*
+ * 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.lifecycle;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
+import org.apache.zeppelin.interpreter.InterpreterSetting;
+import org.apache.zeppelin.interpreter.InterpreterSettingManager;
+import org.apache.zeppelin.interpreter.ManagedInterpreterGroup;
+import org.apache.zeppelin.scheduler.ExecutorFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+/**
+ * Closes interpreter groups that have been idle for longer than a threshold, driven by Zeppelin
+ * server rather than by the interpreter process itself.
+ *
+ * <p>{@link TimeoutLifecycleManager} does the same thing from inside the interpreter process, where
+ * the threshold can only arrive through the configuration map pushed over Thrift at startup. That
+ * map holds {@link ConfVars} entries only, so an interpreter setting property never reaches it and
+ * every process gets the same global threshold. Deciding here means the threshold of the owning
+ * interpreter setting can just be read.
+ *
+ * <p>Follows {@code zeppelin.interpreter.lifecyclemanager.class}, which already says whether idle
+ * reclaim is wanted: its {@link NullLifecycleManager} default leaves a deployment untouched, and
+ * {@link TimeoutLifecycleManager} enables this. Any other implementation is left alone. The
+ * in-process manager stays as a fallback for a server that went away and is given the same resolved
+ * threshold by {@link #processConfigurationOverrides}.
+ */
+public class IdleInterpreterReclaimer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(IdleInterpreterReclaimer.class);
+
+ private static final String SCHEDULER_NAME = "IdleInterpreterReclaimer";
+
+ /**
+ * Threshold property. On an interpreter setting, {@code 0} or below means never reclaimed.
+ */
+ public static final String IDLE_TIMEOUT_THRESHOLD_PROPERTY =
+ ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD.getVarName();
+
+ private final ZeppelinConfiguration zConf;
+ private final InterpreterSettingManager interpreterSettingManager;
+
+ private ScheduledExecutorService checkScheduler;
+
+ public IdleInterpreterReclaimer(ZeppelinConfiguration zConf,
+ InterpreterSettingManager interpreterSettingManager) {
+ this.zConf = zConf;
+ this.interpreterSettingManager = interpreterSettingManager;
+ }
+
+ private static boolean isEnabled(ZeppelinConfiguration zConf) {
+ return TimeoutLifecycleManager.class.getName().equals(zConf.getLifecycleManagerClass());
+ }
+
+ public void start() {
+ if (!isEnabled(zConf)) {
+ LOGGER.debug("Server driven idle interpreter reclaim is off, {} is {}",
+ ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS.getVarName(),
+ zConf.getLifecycleManagerClass());
+ return;
+ }
+ long checkInterval = zConf.getInterpreterIdleCheckInterval();
+ if (checkInterval <= 0) {
+ LOGGER.warn("Not starting idle interpreter reclaim: {} must be positive but is {}",
+ ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_CHECK_INTERVAL.getVarName(),
+ checkInterval);
+ return;
+ }
+ checkScheduler = ExecutorFactory.singleton().createOrGetScheduled(SCHEDULER_NAME, 1);
+ // Fixed delay rather than fixed rate, so that a slow close does not queue up further checks.
+ checkScheduler.scheduleWithFixedDelay(this::reclaimIdleInterpreterGroups,
+ checkInterval, checkInterval, MILLISECONDS);
+ LOGGER.info("Server driven idle interpreter reclaim started with checkInterval: {}ms, "
+ + "default threshold: {}ms", checkInterval, zConf.getInterpreterIdleTimeoutThreshold());
+ }
+
+ public void stop() {
+ if (checkScheduler != null) {
+ ExecutorFactory.singleton().shutdown(SCHEDULER_NAME);
+ checkScheduler = null;
+ LOGGER.info("Server driven idle interpreter reclaim stopped");
+ }
+ }
+
+ /**
+ * Closes every interpreter group idle for longer than the threshold of its interpreter setting.
+ * Uses in-memory state only: {@code isAlive()} and {@code isRunning()} cost a socket connect for
+ * docker and a kube-apiserver round trip for k8s, and this walks every group on a timer.
+ */
+ @VisibleForTesting
+ void reclaimIdleInterpreterGroups() {
+ long now = System.currentTimeMillis();
+ for (ManagedInterpreterGroup interpreterGroup :
+ interpreterSettingManager.getAllInterpreterGroup()) {
+ try {
+ reclaimIfIdle(interpreterGroup, now);
+ } catch (Exception e) {
+ LOGGER.error("Fail to reclaim idle interpreter group: {}", interpreterGroup.getId(), e);
+ }
+ }
+ }
+
+ private void reclaimIfIdle(ManagedInterpreterGroup interpreterGroup, long now) {
+ if (interpreterGroup.isLaunchingInterpreterProcess()) {
+ // The handle is published before the process is ready, and a launch can outlast the
+ // threshold, so this would close a process that is starting rather than an idle one.
+ return;
+ }
+ if (interpreterGroup.getInterpreterProcess() == null) {
+ // Like TimeoutLifecycleManager, only manage a group once its process has started.
+ return;
+ }
+ if (interpreterGroup.isEmpty()) {
+ // No session left: the group is already on its way out through close().
+ return;
+ }
+
+ InterpreterSetting interpreterSetting = interpreterGroup.getInterpreterSetting();
+ long threshold = getIdleTimeoutThreshold(zConf, interpreterSetting);
+ if (threshold <= 0) {
+ LOGGER.debug("Interpreter group {} is never reclaimed, its threshold is {}ms",
+ interpreterGroup.getId(), threshold);
+ return;
+ }
+
+ long idleTimeInMillis = now - interpreterGroup.getLastUsedTimeInMillis();
+ if (idleTimeInMillis <= threshold) {
+ return;
+ }
+
+ LOGGER.info("Reclaiming interpreter group {} of interpreter setting {}: idle for {}ms which "
+ + "exceeds its threshold of {}ms", interpreterGroup.getId(),
+ interpreterSetting == null ? "?" : interpreterSetting.getName(),
+ idleTimeInMillis, threshold);
+ interpreterGroup.close();
+ }
+
+ /**
+ * @return idle threshold in milliseconds for the given interpreter setting, taking its own
+ * {@link #IDLE_TIMEOUT_THRESHOLD_PROPERTY} property over the global configuration
+ */
+ @VisibleForTesting
+ static long getIdleTimeoutThreshold(ZeppelinConfiguration zConf,
+ InterpreterSetting interpreterSetting) {
+ if (interpreterSetting != null) {
+ String override =
+ interpreterSetting.getJavaProperties().getProperty(IDLE_TIMEOUT_THRESHOLD_PROPERTY);
+ if (StringUtils.isNotBlank(override)) {
+ try {
+ return ZeppelinConfiguration.parseTimeMillis(override);
+ } catch (RuntimeException e) {
+ LOGGER.warn("Ignoring unparsable {} of interpreter setting {}: {}",
+ IDLE_TIMEOUT_THRESHOLD_PROPERTY, interpreterSetting.getName(), override, e);
+ }
+ }
+ }
+ return zConf.getInterpreterIdleTimeoutThreshold();
+ }
+
+ /**
+ * Gives the in-process {@link TimeoutLifecycleManager} fallback the threshold resolved here
+ * instead of the global one. A setting that opted out gets {@link Long#MAX_VALUE} rather than its
+ * own {@code 0}, which {@link TimeoutLifecycleManager} would read as "shut down at the next
+ * check" since it has no way to express "never".
+ *
+ * @return entries to put on top of {@link ZeppelinConfiguration#getCompleteConfiguration()},
+ * empty when server driven reclaim is off
+ */
+ public static Map<String, String> processConfigurationOverrides(
+ ZeppelinConfiguration zConf, InterpreterSetting interpreterSetting) {
+ if (!isEnabled(zConf)) {
+ return Collections.emptyMap();
+ }
+ long threshold = getIdleTimeoutThreshold(zConf, interpreterSetting);
+ return Collections.singletonMap(IDLE_TIMEOUT_THRESHOLD_PROPERTY,
+ String.valueOf(threshold <= 0 ? Long.MAX_VALUE : threshold));
+ }
+}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
index 6bd2e20232..efa9ea99a0 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
@@ -111,6 +111,19 @@ public class RemoteInterpreter extends Interpreter {
return (ManagedInterpreterGroup) super.getInterpreterGroup();
}
+ /**
+ * Mirrors the {@code onInterpreterUse} hooks that {@code RemoteInterpreterServer} calls inside
+ * the interpreter process, so that server driven idle reclaim sees the same activity signal.
+ * Needed explicitly because {@link #getOrCreateInterpreterProcess()} returns the cached handle
+ * without going through the interpreter group once the process exists.
+ */
+ private void markInterpreterGroupUsed() {
+ ManagedInterpreterGroup intpGroup = getInterpreterGroup();
+ if (intpGroup != null) {
+ intpGroup.onInterpreterUse();
+ }
+ }
+
@Override
public void open() throws InterpreterException {
synchronized (this) {
@@ -194,6 +207,7 @@ public class RemoteInterpreter extends Interpreter {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("st:\n{}", st);
}
+ markInterpreterGroupUsed();
final FormType form = getFormType();
RemoteInterpreterProcess interpreterProcess = null;
@@ -292,6 +306,7 @@ public class RemoteInterpreter extends Interpreter {
LOGGER.warn("getProgress is called when RemoterInterpreter is not opened for {}", className);
return 0;
}
+ markInterpreterGroupUsed();
RemoteInterpreterProcess interpreterProcess = null;
try {
interpreterProcess = getOrCreateInterpreterProcess();
@@ -325,6 +340,7 @@ public class RemoteInterpreter extends Interpreter {
LOGGER.warn("getStatus is called when RemoteInterpreter is not opened for {}", className);
return Job.Status.UNKNOWN.name();
}
+ markInterpreterGroupUsed();
RemoteInterpreterProcess interpreterProcess = null;
try {
interpreterProcess = getOrCreateInterpreterProcess();
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java
index 95802a64fe..e994439c89 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java
@@ -29,7 +29,10 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.text.SimpleDateFormat;
+import java.util.Collections;
import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
/**
* Abstract class for interpreter process
@@ -101,8 +104,20 @@ public abstract class RemoteInterpreterProcess implements InterpreterClient, Aut
}
public void init(ZeppelinConfiguration zConf) {
+ init(zConf, Collections.emptyMap());
+ }
+
+ /**
+ * Pushes the server configuration into the interpreter process.
+ *
+ * @param overrides entries to put on top of the global configuration, for settings that are
+ * resolved per interpreter setting rather than globally
+ */
+ public void init(ZeppelinConfiguration zConf, Map<String, String> overrides) {
+ Map<String, String> properties = new HashMap<>(zConf.getCompleteConfiguration());
+ properties.putAll(overrides);
callRemoteFunction(client -> {
- client.init(zConf.getCompleteConfiguration());
+ client.init(properties);
return null;
});
}
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/lifecycle/IdleInterpreterReclaimerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/lifecycle/IdleInterpreterReclaimerTest.java
new file mode 100644
index 0000000000..877b834b04
--- /dev/null
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/lifecycle/IdleInterpreterReclaimerTest.java
@@ -0,0 +1,337 @@
+/*
+ * 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.lifecycle;
+
+import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
+import org.apache.zeppelin.interpreter.AbstractInterpreterTest;
+import org.apache.zeppelin.interpreter.ExecutionContext;
+import org.apache.zeppelin.interpreter.InterpreterSetting;
+import org.apache.zeppelin.interpreter.InterpreterSettingManager;
+import org.apache.zeppelin.interpreter.ManagedInterpreterGroup;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreter;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
+import org.apache.zeppelin.scheduler.Job;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests server driven idle reclaim, above all that an interpreter setting can override the global
+ * threshold in either direction. That override is what the interpreter process side
+ * {@link TimeoutLifecycleManager} cannot offer, because its threshold only reaches the process
+ * through the global configuration map.
+ */
+class IdleInterpreterReclaimerTest extends AbstractInterpreterTest {
+
+ private static final String THRESHOLD_PROPERTY =
+ IdleInterpreterReclaimer.IDLE_TIMEOUT_THRESHOLD_PROPERTY;
+
+ @Override
+ @BeforeEach
+ public void setUp() throws Exception {
+ super.setUp();
+ zConf.setProperty(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS.getVarName(),
+ TimeoutLifecycleManager.class.getName());
+ zConf.setProperty(
+ ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_CHECK_INTERVAL.getVarName(),
+ "1000");
+ // The reclaimer picks these up when it starts, and that already happened while
+ // super.setUp() built the InterpreterSettingManager, so restart it.
+ interpreterSettingManager.getIdleInterpreterReclaimer().stop();
+ interpreterSettingManager.getIdleInterpreterReclaimer().start();
+ }
+
+ /**
+ * A setting may ask to be reclaimed sooner than the global threshold allows. The global
+ * threshold stays at its 1h default here, so only the per setting value of 10s can close it.
+ */
+ @Test
+ void perSettingThresholdReclaimsEarlierThanTheGlobalOne() throws Exception {
+ InterpreterSetting interpreterSetting =
+ interpreterSettingManager.getInterpreterSettingByName("test");
+ interpreterSetting.setProperty(THRESHOLD_PROPERTY, "10s");
+
+ startEchoInterpreter();
+ assertEquals(1, interpreterSetting.getAllInterpreterGroups().size());
+
+ waitForInterpreterGroups(interpreterSetting, 0, 40);
+ assertEquals(0, interpreterSetting.getAllInterpreterGroups().size(),
+ "the group should be reclaimed after the per setting threshold of 10s");
+ }
+
+ /**
+ * The other direction: a non positive per setting threshold means keep it, whatever the short
+ * global threshold says.
+ */
+ @Test
+ void perSettingThresholdCanOptOutOfAShortGlobalThreshold() throws Exception {
+ zConf.setProperty(
+ ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD.getVarName(), "5s");
+
+ InterpreterSetting interpreterSetting =
+ interpreterSettingManager.getInterpreterSettingByName("test");
+ interpreterSetting.setProperty(THRESHOLD_PROPERTY, "0");
+
+ startEchoInterpreter();
+ assertEquals(1, interpreterSetting.getAllInterpreterGroups().size());
+
+ Thread.sleep(20 * 1000);
+ assertEquals(1, interpreterSetting.getAllInterpreterGroups().size(),
+ "the setting opted out of reclaim, so the short global threshold must not apply");
+ }
+
+ @Test
+ void globalThresholdAppliesWhenTheSettingDoesNotOverrideIt() throws Exception {
+ zConf.setProperty(
+ ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD.getVarName(), "10s");
+
+ InterpreterSetting interpreterSetting =
+ interpreterSettingManager.getInterpreterSettingByName("test");
+
+ startEchoInterpreter();
+ assertEquals(1, interpreterSetting.getAllInterpreterGroups().size());
+
+ waitForInterpreterGroups(interpreterSetting, 0, 40);
+ assertEquals(0, interpreterSetting.getAllInterpreterGroups().size());
+ }
+
+ /**
+ * A paragraph running for longer than the threshold must not have its interpreter pulled out
+ * from under it. While a job runs the server polls its status, which counts as use.
+ */
+ @Test
+ void aRunningParagraphKeepsItsInterpreterAlive() throws Exception {
+ zConf.setProperty(
+ ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD.getVarName(), "5s");
+
+ InterpreterSetting interpreterSetting =
+ interpreterSettingManager.getInterpreterSettingByName("test");
+ final RemoteInterpreter sleepInterpreter =
+ (RemoteInterpreter) interpreterFactory.getInterpreter("test.sleep",
+ new ExecutionContext("user1", "note1", "test"));
+
+ // Submit through the scheduler the way Zeppelin submits a paragraph, so that the job status
+ // poller runs.
+ sleepInterpreter.getScheduler().submit(new Job<Object>("test-job", null) {
+ @Override
+ public Object getReturn() {
+ return null;
+ }
+
+ @Override
+ public int progress() {
+ return 0;
+ }
+
+ @Override
+ public Map<String, Object> info() {
+ return null;
+ }
+
+ @Override
+ protected Object jobRun() throws Throwable {
+ return sleepInterpreter.interpret("30000", createDummyInterpreterContext());
+ }
+
+ @Override
+ protected boolean jobAbort() {
+ return false;
+ }
+
+ @Override
+ public void setResult(Object results) {
+ }
+ });
+
+ long deadline = System.currentTimeMillis() + 30 * 1000;
+ while (!sleepInterpreter.isOpened() && System.currentTimeMillis() < deadline) {
+ Thread.sleep(500);
+ }
+ assertTrue(sleepInterpreter.isOpened(), "interpreter did not start");
+ assertEquals(1, interpreterSetting.getAllInterpreterGroups().size());
+
+ Thread.sleep(20 * 1000);
+ assertEquals(1, interpreterSetting.getAllInterpreterGroups().size(),
+ "a running paragraph must keep its interpreter group alive");
+ }
+
+ /**
+ * A probe is cheap for the local launcher but not for docker or k8s, and this scan walks every
+ * group on a timer.
+ */
+ @Test
+ void scanNeverProbesTheInterpreterProcess() {
+ RemoteInterpreterProcess process = mock(RemoteInterpreterProcess.class);
+
+ InterpreterSetting interpreterSetting = mock(InterpreterSetting.class);
+ when(interpreterSetting.getName()).thenReturn("probe-guard");
+ when(interpreterSetting.getJavaProperties()).thenReturn(new Properties());
+
+ ManagedInterpreterGroup interpreterGroup = mock(ManagedInterpreterGroup.class);
+ when(interpreterGroup.getId()).thenReturn("probe-guard-shared_process");
+ when(interpreterGroup.getInterpreterProcess()).thenReturn(process);
+ when(interpreterGroup.getInterpreterSetting()).thenReturn(interpreterSetting);
+ when(interpreterGroup.isEmpty()).thenReturn(false);
+ // Idle since the epoch, so it is well past any threshold and does get closed.
+ when(interpreterGroup.getLastUsedTimeInMillis()).thenReturn(0L);
+
+ InterpreterSettingManager settingManager = mock(InterpreterSettingManager.class);
+ when(settingManager.getAllInterpreterGroup())
+ .thenReturn(Collections.singletonList(interpreterGroup));
+
+ new IdleInterpreterReclaimer(zConf, settingManager).reclaimIdleInterpreterGroups();
+
+ verify(interpreterGroup).close();
+ verify(process, never()).isAlive();
+ verify(process, never()).isRunning();
+ }
+
+ /**
+ * The handle is published before the process is ready and the group has been idle since it was
+ * created, so without the launching check the scan closes a process that is starting up.
+ */
+ @Test
+ void aGroupBeingLaunchedIsNotReclaimed() {
+ ManagedInterpreterGroup interpreterGroup = mock(ManagedInterpreterGroup.class);
+ when(interpreterGroup.getId()).thenReturn("launching-shared_process");
+ when(interpreterGroup.isLaunchingInterpreterProcess()).thenReturn(true);
+ when(interpreterGroup.getInterpreterProcess())
+ .thenReturn(mock(RemoteInterpreterProcess.class));
+ when(interpreterGroup.isEmpty()).thenReturn(false);
+ when(interpreterGroup.getLastUsedTimeInMillis()).thenReturn(0L);
+
+ InterpreterSettingManager settingManager = mock(InterpreterSettingManager.class);
+ when(settingManager.getAllInterpreterGroup())
+ .thenReturn(Collections.singletonList(interpreterGroup));
+
+ new IdleInterpreterReclaimer(zConf, settingManager).reclaimIdleInterpreterGroups();
+
+ verify(interpreterGroup, never()).close();
+ }
+
+ @Test
+ void thresholdResolutionPrefersTheSettingAndFallsBackOnGarbage() {
+ zConf.setProperty(
+ ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD.getVarName(), "1h");
+
+ assertEquals(3600000L, IdleInterpreterReclaimer.getIdleTimeoutThreshold(zConf, null),
+ "no setting at all means the global threshold");
+
+ InterpreterSetting interpreterSetting = mock(InterpreterSetting.class);
+ when(interpreterSetting.getName()).thenReturn("threshold-resolution");
+
+ when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties(null));
+ assertEquals(3600000L,
+ IdleInterpreterReclaimer.getIdleTimeoutThreshold(zConf, interpreterSetting),
+ "no override means the global threshold");
+
+ when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("10s"));
+ assertEquals(10000L,
+ IdleInterpreterReclaimer.getIdleTimeoutThreshold(zConf, interpreterSetting));
+
+ when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("600000"));
+ assertEquals(600000L,
+ IdleInterpreterReclaimer.getIdleTimeoutThreshold(zConf, interpreterSetting),
+ "a plain number is milliseconds");
+
+ when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("0"));
+ assertEquals(0L,
+ IdleInterpreterReclaimer.getIdleTimeoutThreshold(zConf, interpreterSetting),
+ "zero opts the setting out of reclaim");
+
+ when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("not-a-duration"));
+ assertEquals(3600000L,
+ IdleInterpreterReclaimer.getIdleTimeoutThreshold(zConf, interpreterSetting),
+ "an unparsable override must fall back to the global threshold");
+ }
+
+ /**
+ * A setting that opted out must not be shut down by the in-process fallback either. Its own
+ * {@code 0} would mean "shut down at the next check" there, so it never reaches the process.
+ */
+ @Test
+ void optingOutDisablesTheInProcessFallbackToo() {
+ InterpreterSetting interpreterSetting = mock(InterpreterSetting.class);
+ when(interpreterSetting.getName()).thenReturn("opt-out");
+ when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("0"));
+
+ Map<String, String> overrides =
+ IdleInterpreterReclaimer.processConfigurationOverrides(zConf, interpreterSetting);
+ assertEquals(String.valueOf(Long.MAX_VALUE), overrides.get(THRESHOLD_PROPERTY));
+ assertNull(overrides.get(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS.getVarName()),
+ "the lifecycle manager the operator configured must be left alone");
+
+ when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("10s"));
+ overrides = IdleInterpreterReclaimer.processConfigurationOverrides(zConf, interpreterSetting);
+ assertEquals("10000", overrides.get(THRESHOLD_PROPERTY),
+ "the process gets the resolved threshold, not the global one");
+ }
+
+ /**
+ * With the default lifecycle manager nothing is reclaimed and nothing is overridden, so an
+ * existing deployment is untouched.
+ */
+ @Test
+ void defaultLifecycleManagerLeavesEverythingAlone() {
+ zConf.setProperty(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS.getVarName(),
+ NullLifecycleManager.class.getName());
+
+ InterpreterSetting interpreterSetting = mock(InterpreterSetting.class);
+ when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("10s"));
+
+ assertTrue(IdleInterpreterReclaimer.processConfigurationOverrides(zConf, interpreterSetting)
+ .isEmpty());
+ }
+
+ private Properties thresholdProperties(String threshold) {
+ Properties properties = new Properties();
+ if (threshold != null) {
+ properties.setProperty(THRESHOLD_PROPERTY, threshold);
+ }
+ return properties;
+ }
+
+ private void startEchoInterpreter() throws Exception {
+ RemoteInterpreter echoInterpreter =
+ (RemoteInterpreter) interpreterFactory.getInterpreter("test.echo",
+ new ExecutionContext("user1", "note1", "test"));
+ echoInterpreter.interpret("hello", createDummyInterpreterContext());
+ assertTrue(echoInterpreter.isOpened());
+ }
+
+ private void waitForInterpreterGroups(InterpreterSetting interpreterSetting,
+ int expectedSize,
+ int maxSeconds) throws Exception {
+ long deadline = System.currentTimeMillis() + maxSeconds * 1000L;
+ while (interpreterSetting.getAllInterpreterGroups().size() != expectedSize
+ && System.currentTimeMillis() < deadline) {
+ Thread.sleep(1000);
+ }
+ }
+}