Discard a failed interpreter launch so the next run relaunches

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;
   }

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 {