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 <[email protected]>
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 <[email protected]>
---
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);
+ }
+ }
+}
