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 3a8b14ee81e..52b1e9f68e5 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 @@ -73,10 +73,15 @@ public RemoteInterpreterProcess getOrCreateInterpreterProcess(String userName, launchingInterpreterProcess = true; remoteInterpreterProcess = interpreterSetting.createInterpreterProcess(id, userName, properties); - remoteInterpreterProcess.start(userName); - remoteInterpreterProcess.init(zConf, - IdleInterpreterReclaimer.processConfigurationOverrides(zConf, - interpreterSetting)); + try { + remoteInterpreterProcess.start(userName); + remoteInterpreterProcess.init(zConf, + IdleInterpreterReclaimer.processConfigurationOverrides(zConf, + interpreterSetting)); + } catch (IOException | RuntimeException e) { + discardFailedLaunch(e); + throw e; + } getInterpreterSetting().getRecoveryStorage() .onInterpreterClientStart(remoteInterpreterProcess); } @@ -119,6 +124,23 @@ public RemoteInterpreterProcess getInterpreterProcess() { return remoteInterpreterProcess; } + /** + * Without this, a failed launch (e.g. an interpreter pod rejected by a namespace quota) keeps a + * handle that never runs, and every later paragraph fails with "Interpreter process is not + * running" until the interpreter is restarted manually. + */ + private void discardFailedLaunch(Exception cause) { + RemoteInterpreterProcess failedProcess = remoteInterpreterProcess; + remoteInterpreterProcess = null; + LOGGER.warn("Fail to launch interpreter process for InterpreterGroup: {}, discarding it so " + + "that the next run launches a new one", id, cause); + try { + failedProcess.stop(); + } catch (RuntimeException e) { + LOGGER.warn("Fail to clean up interpreter process of InterpreterGroup: {}", id, e); + } + } + public RemoteInterpreterProcess getRemoteInterpreterProcess() { return remoteInterpreterProcess; } diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroupTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroupTest.java index c1bff6e264b..df1a6955d97 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroupTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroupTest.java @@ -19,6 +19,8 @@ import org.junit.jupiter.api.BeforeEach; import org.apache.zeppelin.conf.ZeppelinConfiguration; +import org.apache.zeppelin.interpreter.recovery.RecoveryStorage; +import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess; import org.eclipse.aether.RepositoryException; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; @@ -33,8 +35,18 @@ import java.util.concurrent.TimeUnit; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; +import static org.mockito.ArgumentMatchers.any; +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.verify; class ManagedInterpreterGroupTest { @@ -97,6 +109,27 @@ void testInterpreterGroup() { assertEquals(0, interpreterGroup.getSessionNum()); } + @Test + void failedLaunchIsDiscardedSoTheNextRunLaunchesAgain() throws Exception { + InterpreterSetting setting = spy(interpreterSetting); + doReturn(mock(RecoveryStorage.class)).when(setting).getRecoveryStorage(); + RemoteInterpreterProcess rejected = mock(RemoteInterpreterProcess.class); + // fabric8's KubernetesClientException, as thrown for a quota rejected pod, is unchecked. + doThrow(new IllegalStateException("pods is forbidden: exceeded quota")) + .when(rejected).start(any()); + RemoteInterpreterProcess relaunched = mock(RemoteInterpreterProcess.class); + doReturn(rejected, relaunched).when(setting).createInterpreterProcess(any(), any(), any()); + ManagedInterpreterGroup group = new ManagedInterpreterGroup("g1", setting, zConf); + + assertThrows(IllegalStateException.class, + () -> group.getOrCreateInterpreterProcess("u", new Properties())); + assertNull(group.getInterpreterProcess(), "a failed launch must not be reused"); + assertFalse(group.isLaunchingInterpreterProcess()); + verify(rejected).stop(); + + assertSame(relaunched, group.getOrCreateInterpreterProcess("u", new Properties())); + } + @Test @Timeout(30) void close_doesNotDeadlockWithConcurrentOpen() throws Exception {