Address review feedback: fix double-apply, log levels, and Initializing→Stopped

- Fix transition() to use get()+updateAndGet() instead of re-applying the
  transition function to derive the new state for logging
- Log all transitions at INFO, Failed transitions at WARN
- Support Initializing→Stopped transition so VCs are not orphaned if
  shutdown occurs during startup
- Assert failure cause on Stopped state in startup failure test

Assisted-by: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Sam Barker <sam@quadrocket.co.uk>

Author: SamBarker

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/KafkaProxy.java

@@ -427,6 +427,9 @@ private void transitionAllToDraining() {
             else if (m.getState() instanceof VirtualClusterLifecycleState.Failed) {
                 m.stop();
             }
+            else if (m.getState() instanceof VirtualClusterLifecycleState.Initializing) {
+                m.stop();
+            }
         });
     }
 

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/VirtualClusterLifecycleManager.java

@@ -86,13 +86,16 @@ public void drainComplete() {
     }
 
     /**
-     * Transitions from {@link Failed} to {@link Stopped}.
+     * Transitions to {@link Stopped} from {@link Failed} or {@link Initializing}.
      */
     public void stop() {
         transition(current -> {
             if (current instanceof Failed s) {
                 return s.toStopped();
             }
+            if (current instanceof Initializing s) {
+                return s.toStopped();
+            }
             throw unexpectedState(current, "stop");
         });
     }
@@ -106,9 +109,11 @@ public String getClusterName() {
     }
 
     private void transition(UnaryOperator<VirtualClusterLifecycleState> transitionFn) {
-        VirtualClusterLifecycleState previous = state.getAndUpdate(transitionFn);
-        VirtualClusterLifecycleState current = transitionFn.apply(previous);
-        var logBuilder = (current instanceof Initializing || current instanceof Stopped) ? LOGGER.atInfo() : LOGGER.atDebug();
+        // Use get() then updateAndGet() rather than getAndUpdate() so we capture
+        // both the previous and new state without re-applying the transition function.
+        VirtualClusterLifecycleState previous = state.get();
+        VirtualClusterLifecycleState current = state.updateAndGet(transitionFn);
+        var logBuilder = (current instanceof Failed) ? LOGGER.atWarn() : LOGGER.atInfo();
         logBuilder
                 .addKeyValue("virtualCluster", clusterName)
                 .addKeyValue("from", previous.getClass().getSimpleName())

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/VirtualClusterLifecycleState.java

@@ -40,6 +40,14 @@ public Serving toServing() {
         public Failed toFailed(Throwable cause) {
             return new Failed(cause);
         }
+
+        /**
+         * Cluster removed before initialization completed (e.g. proxy shutdown during startup).
+         * @return the new stopped state
+         */
+        public Stopped toStopped() {
+            return new Stopped(null);
+        }
     }
 
     /** The proxy has completed setup and is serving traffic for this cluster. */

kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/KafkaProxyLifecycleTest.java

@@ -142,7 +142,10 @@ void shouldTransitionToStoppedOnStartupFailure() throws Exception {
             // then
             var manager = proxy.lifecycleManagerFor("demo1");
             assertThat(manager).isNotNull();
-            assertThat(manager.getState()).isInstanceOf(Stopped.class);
+            assertThat(manager.getState())
+                    .asInstanceOf(org.assertj.core.api.InstanceOfAssertFactories.type(Stopped.class))
+                    .extracting(Stopped::priorFailureCause)
+                    .isInstanceOf(PluginConfigurationException.class);
         }
     }
 }

kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/VirtualClusterLifecycleManagerTest.java

@@ -116,6 +116,16 @@ void shouldRetainFailureCauseAfterStop() {
                 .isInstanceOfSatisfying(Stopped.class, stopped -> assertThat(stopped.priorFailureCause()).isSameAs(cause));
     }
 
+    @Test
+    void shouldTransitionFromInitializingToStoppedOnShutdown() {
+        // when
+        manager.stop();
+
+        // then
+        assertThat(manager.getState())
+                .isInstanceOfSatisfying(Stopped.class, stopped -> assertThat(stopped.priorFailureCause()).isNull());
+    }
+
     @Test
     void shouldHaveNoPriorFailureCauseWhenStoppedFromDraining() {
         // given