Persist supervisor spec only after successful start (#14150)

* Persist spec after successful start

* Fix checkstyle.

* checkstyle after mvn install

.github/workflows/static-checks.yml

@@ -81,6 +81,10 @@ jobs:
           ${MVN} clean install -q -ff -pl '!distribution,!:druid-it-image,!:druid-it-cases' ${MAVEN_SKIP} ${MAVEN_SKIP_TESTS} -T1C &&
           ${MVN} install -q -ff -pl 'distribution' ${MAVEN_SKIP} ${MAVEN_SKIP_TESTS}
 
+      - name: checkstyle
+        if: ${{ matrix.java == 'jdk8' }}
+        run: ${MVN} checkstyle:checkstyle --fail-at-end
+
       - name: license checks
         if: ${{ matrix.java == 'jdk8' }}
         run: ./.github/scripts/license_checks_script.sh
@@ -101,10 +105,6 @@ jobs:
         if: ${{ matrix.java == 'jdk8' }}
         run: ${MVN} animal-sniffer:check --fail-at-end
 
-      - name: checkstyle
-        if: ${{ matrix.java == 'jdk8' }}
-        run: ${MVN} checkstyle:checkstyle --fail-at-end
-
       - name: enforcer checks
         if: ${{ matrix.java == 'jdk8' }}
         run: ${MVN} enforcer:enforce --fail-at-end

indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java

@@ -132,7 +132,7 @@ public class SupervisorManager
             createAndStartSupervisorInternal(spec, false);
           }
           catch (Exception ex) {
-            log.error(ex, "Failed to start supervisor: [%s]", spec.getId());
+            log.error(ex, "Failed to start supervisor: id [%s]", spec.getId());
           }
         }
       }
@@ -313,10 +313,6 @@ public class SupervisorManager
       return false;
     }
 
-    if (persistSpec) {
-      metadataSupervisorManager.insert(id, spec);
-    }
-
     Supervisor supervisor;
     SupervisorTaskAutoScaler autoscaler;
     try {
@@ -326,18 +322,22 @@ public class SupervisorManager
       supervisor.start();
       if (autoscaler != null) {
         autoscaler.start();
-        autoscalers.put(id, autoscaler);
       }
     }
     catch (Exception e) {
-      // Supervisor creation or start failed write tombstone only when trying to start a new supervisor
+      log.error("Failed to create and start supervisor: [%s]", spec.getId());
-      if (persistSpec) {
-        metadataSupervisorManager.insert(id, new NoopSupervisorSpec(null, spec.getDataSources()));
-      }
       throw new RuntimeException(e);
     }
 
+    if (persistSpec) {
+      metadataSupervisorManager.insert(id, spec);
+    }
+
     supervisors.put(id, Pair.of(supervisor, spec));
+    if (autoscaler != null) {
+      autoscalers.put(id, autoscaler);
+    }
+
     return true;
   }
 }

indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java

@@ -38,6 +38,7 @@ import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -260,6 +261,35 @@ public class SupervisorManagerTest extends EasyMockSupport
     // if we get here, we are properly insulated from exploding supervisors
   }
 
+  @Test
+  public void testNoPersistOnFailedStart()
+  {
+    exception.expect(RuntimeException.class);
+
+    Capture<TestSupervisorSpec> capturedInsert = Capture.newInstance();
+
+    EasyMock.expect(metadataSupervisorManager.getLatest()).andReturn(Collections.emptyMap());
+    metadataSupervisorManager.insert(EasyMock.eq("id1"), EasyMock.capture(capturedInsert));
+    supervisor1.start();
+    supervisor1.stop(true);
+    supervisor2.start();
+    EasyMock.expectLastCall().andThrow(new RuntimeException("supervisor failed to start"));
+    replayAll();
+
+    final SupervisorSpec testSpecOld = new TestSupervisorSpec("id1", supervisor1);
+    final SupervisorSpec testSpecNew = new TestSupervisorSpec("id1", supervisor2);
+
+    manager.start();
+    try {
+      manager.createOrUpdateAndStartSupervisor(testSpecOld);
+      manager.createOrUpdateAndStartSupervisor(testSpecNew);
+    }
+    catch (Exception e) {
+      Assert.assertEquals(testSpecOld, capturedInsert.getValue());
+      throw e;
+    }
+  }
+
   @Test
   public void testStopThrowsException()
   {