This is an automated email from the ASF dual-hosted git repository.
ParkGyeongTae 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 79a6530ec0 [ZEPPELIN-6539] Fix DockerClient / container resource leaks
on interpreter failure paths
79a6530ec0 is described below
commit 79a6530ec0529f007f99b8f20f473dde3ca94704
Author: HyeonUk Kang <[email protected]>
AuthorDate: Thu Jul 16 23:52:02 2026 +0900
[ZEPPELIN-6539] Fix DockerClient / container resource leaks on interpreter
failure paths
### What is this PR for?
The Docker interpreter launcher (DockerInterpreterProcess) can leak
resources on error paths when interpreters are started/stopped repeatedly by
notebook workloads.
1. stop() may leak the DockerClient. docker.close() was called after the
try/catch that kills/removes the container. If killContainer/removeContainer
throws an unexpected (unchecked) exception, it is not caught and docker.close()
is skipped, leaking the underlying HTTP socket / file descriptors.
2. start() may leave an orphaned container. start() runs pull → create →
start → copy files → exec → wait-for-register. If preparation fails after the
container is started (e.g. copyRunFileToContainer / execInContainer), the
running container is not cleaned up
### What type of PR is it?
Bug Fix
### Todos
- [x] Fix stop() to always close the DockerClient
- [x] Fix start() to clean up a started container on preparation failure
- [x] Add unit tests (Mockito interaction tests)
### What is the Jira issue?
* [[ZEPPELIN-6539]](https://issues.apache.org/jira/browse/ZEPPELIN-6539)
### How should this be tested?
Automated unit tests added in DockerInterpreterProcessTest (DockerClient is
mocked):
- stop_alwaysClosesDockerClient_evenWhenKillContainerFails
- start_removesContainer_whenContainerPreparationFails
### Screenshots (if appropriate)
### Questions:
* Does the license files need to update? - No
* Is there breaking changes for older versions? - No
* Does this needs documentation? - No
Closes #5302 from hyunw9/ZEPPELIN-6539.
Signed-off-by: ParkGyeongTae <[email protected]>
---
.../launcher/DockerInterpreterProcess.java | 50 +++++++++++-
.../launcher/DockerInterpreterProcessTest.java | 90 ++++++++++++++++++++++
2 files changed, 136 insertions(+), 4 deletions(-)
diff --git
a/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java
b/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java
index 643afb2061..3004ae13f7 100644
---
a/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java
+++
b/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java
@@ -79,7 +79,8 @@ public class DockerInterpreterProcess extends
RemoteInterpreterProcess {
private AtomicBoolean dockerStarted = new AtomicBoolean(false);
- private DockerClient docker = null;
+ @VisibleForTesting
+ DockerClient docker;
private final String containerName;
private String containerHost = "";
private int containerPort = 0;
@@ -154,9 +155,15 @@ public class DockerInterpreterProcess extends
RemoteInterpreterProcess {
return interpreterSettingName;
}
+ // allows a mock DockerClient to be injected in unit tests.
+ @VisibleForTesting
+ DockerClient createDockerClient(String dockerHost) {
+ return DefaultDockerClient.builder().uri(URI.create(dockerHost)).build();
+ }
+
@Override
public void start(String userName) throws IOException {
- docker = DefaultDockerClient.builder().uri(URI.create(dockerHost)).build();
+ docker = createDockerClient(dockerHost);
removeExistContainer(containerName);
@@ -220,7 +227,17 @@ public class DockerInterpreterProcess extends
RemoteInterpreterProcess {
}
}
});
+ } catch (DockerException e) {
+ throw new IOException(e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Docker preparations were interrupted.", e);
+ }
+ // Create, start and prepare the container. If anything fails after the
+ // container has been created/started, roll it back so we don't leak an
+ // orphaned container holding resources until the next launch reuses the
name.
+ try {
final ContainerCreation containerCreation
= docker.createContainer(containerConfig, containerName);
String containerId = containerCreation.id();
@@ -232,10 +249,15 @@ public class DockerInterpreterProcess extends
RemoteInterpreterProcess {
execInContainer(containerId, dockerCommand, false);
} catch (DockerException e) {
+ cleanupContainerQuietly();
throw new IOException(e);
+ } catch (IOException e) {
+ cleanupContainerQuietly();
+ throw e;
} catch (InterruptedException e) {
// Restore interrupted state...
Thread.currentThread().interrupt();
+ cleanupContainerQuietly();
throw new IOException("Docker preparations were interrupted.", e);
}
@@ -363,10 +385,30 @@ public class DockerInterpreterProcess extends
RemoteInterpreterProcess {
Thread.currentThread().interrupt();
} catch (DockerException e) {
LOGGER.error(e.getMessage(), e);
+ } finally {
+ docker.close();
}
+ }
- // Close the docker client
- docker.close();
+ // Best-effort removal of a container that was (partially) created during
start().
+ private void cleanupContainerQuietly() {
+ try {
+ docker.killContainer(containerName);
+ } catch (InterruptedException e) {
+ LOGGER.warn("Interrupted while killing container {} during cleanup",
containerName, e);
+ Thread.currentThread().interrupt();
+ } catch (DockerException e) {
+ LOGGER.warn("Failed to kill container {} during cleanup", containerName,
e);
+ }
+
+ try {
+ docker.removeContainer(containerName);
+ } catch (InterruptedException e) {
+ LOGGER.warn("Interrupted while removing container {} during cleanup",
containerName, e);
+ Thread.currentThread().interrupt();
+ } catch (DockerException e) {
+ LOGGER.warn("Failed to remove container {} during cleanup",
containerName, e);
+ }
}
// Because docker can't create a container with the same name, it will cause
the creation to fail.
diff --git
a/zeppelin-plugins/launcher/docker/src/test/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcessTest.java
b/zeppelin-plugins/launcher/docker/src/test/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcessTest.java
index 1e9ad7d0bd..ea5b5bd84a 100644
---
a/zeppelin-plugins/launcher/docker/src/test/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcessTest.java
+++
b/zeppelin-plugins/launcher/docker/src/test/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcessTest.java
@@ -16,11 +16,16 @@
*/
package org.apache.zeppelin.interpreter.launcher;
+import com.spotify.docker.client.DockerClient;
+import com.spotify.docker.client.exceptions.DockerException;
+import com.spotify.docker.client.messages.ContainerConfig;
+import com.spotify.docker.client.messages.ContainerCreation;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
import org.apache.zeppelin.interpreter.InterpreterOption;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -30,14 +35,99 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class DockerInterpreterProcessTest {
protected static ZeppelinConfiguration zConf =
spy(ZeppelinConfiguration.load());
+ private DockerInterpreterProcess newProcess() {
+ ZeppelinConfiguration conf = spy(ZeppelinConfiguration.load());
+ Properties properties = new Properties();
+ properties.setProperty(
+ ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT.getVarName(), "5000");
+ return new DockerInterpreterProcess(
+ conf,
+ "interpreter-container:1.0",
+ "test_intp_group",
+ "sh",
+ "shell",
+ properties,
+ new HashMap<>(),
+ "zeppelin.server.hostname",
+ 12320,
+ 5000, 10);
+ }
+
+ // stop() must always close the DockerClient, even when killing the container
+ // fails, so the underlying HTTP socket / file descriptors are never leaked.
+ @Test
+ void stop_alwaysClosesDockerClient_evenWhenKillContainerFails() throws
Exception {
+ DockerInterpreterProcess intp = newProcess();
+ DockerClient mockDocker = mock(DockerClient.class);
+ intp.docker = mockDocker;
+ doThrow(new
RuntimeException("unexpected")).when(mockDocker).killContainer(anyString());
+
+ assertThrows(RuntimeException.class, intp::stop);
+
+ verify(mockDocker, times(1)).close();
+ }
+
+ // When start() fails after the container has been started (e.g. file copy /
exec
+ // fails), the container must be cleaned up instead of being left orphaned.
+ @Test
+ void start_removesContainer_whenContainerPreparationFails() throws Exception
{
+ DockerInterpreterProcess intp = spy(newProcess());
+ DockerClient mockDocker = mock(DockerClient.class);
+ doReturn(mockDocker).when(intp).createDockerClient(anyString());
+
+ // No pre-existing container to remove.
+ when(mockDocker.listContainers(any())).thenReturn(Collections.emptyList());
+ // Container is created and started successfully...
+ when(mockDocker.createContainer(any(ContainerConfig.class), anyString()))
+
.thenReturn(ContainerCreation.builder().id("test-container-id").build());
+ // ...but preparing it (the first exec inside the container) fails.
+ doThrow(new DockerException("exec failed"))
+ .when(mockDocker).execCreate(anyString(), any(String[].class), any());
+
+ assertThrows(IOException.class, () -> intp.start("user1"));
+
+ // The container was started, so start() must roll it back before
returning.
+ verify(mockDocker).startContainer("test-container-id");
+ verify(mockDocker).killContainer(anyString());
+ verify(mockDocker).removeContainer(anyString());
+ }
+
+ @Test
+ void start_removesContainer_evenWhenKillFailsDuringCleanup() throws
Exception {
+ DockerInterpreterProcess intp = spy(newProcess());
+ DockerClient mockDocker = mock(DockerClient.class);
+ doReturn(mockDocker).when(intp).createDockerClient(anyString());
+
+ when(mockDocker.listContainers(any())).thenReturn(Collections.emptyList());
+ // Container is created...
+ when(mockDocker.createContainer(any(ContainerConfig.class), anyString()))
+
.thenReturn(ContainerCreation.builder().id("test-container-id").build());
+ // ...but fails to start, so it is created-but-not-running.
+ doThrow(new DockerException("start
failed")).when(mockDocker).startContainer(anyString());
+ // Killing a non-running container fails, but removeContainer must still
fire.
+ doThrow(new DockerException("not
running")).when(mockDocker).killContainer(anyString());
+
+ assertThrows(IOException.class, () -> intp.start("user1"));
+
+ verify(mockDocker).removeContainer(anyString());
+ }
+
@Test
void testCreateIntpProcess() throws IOException {
DockerInterpreterLauncher launcher
