Address review feedback on lifecycle manager

- Remove beginShutdown/completeShutdown composite methods from
  VirtualClusterLifecycleManager; KafkaProxy now calls startDraining,
  drainComplete, and stop directly
- Call initializationFailed(cause) from KafkaProxy startup catch block
  so Failed state carries the real failure cause
- Fix race in transition(): use transitionFn.apply(previous) instead of
  state.get() to avoid observing another thread's update
- Log at DEBUG for most transitions, INFO only for initial and terminal
  states

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

@@ -59,6 +59,7 @@
 import io.kroxylicious.proxy.internal.MeterRegistries;
 import io.kroxylicious.proxy.internal.PortConflictDetector;
 import io.kroxylicious.proxy.internal.VirtualClusterLifecycleManager;
+import io.kroxylicious.proxy.internal.VirtualClusterLifecycleState;
 import io.kroxylicious.proxy.internal.admin.ManagementInitializer;
 import io.kroxylicious.proxy.internal.config.Features;
 import io.kroxylicious.proxy.internal.net.DefaultNetworkBindingOperationProcessor;
@@ -270,6 +271,7 @@ public KafkaProxy startup() {
             STARTUP_SHUTDOWN_LOGGER.atError()
                     .setCause(e)
                     .log("Exception during startup, shutting down");
+            lifecycleManagers.values().forEach(m -> m.initializationFailed(e));
             shutdown();
             throw new LifecycleException("Startup completed exceptionally", e);
         }
@@ -418,11 +420,22 @@ VirtualClusterLifecycleManager lifecycleManagerFor(String clusterName) {
     }
 
     private void transitionAllToDraining() {
-        lifecycleManagers.values().forEach(VirtualClusterLifecycleManager::beginShutdown);
+        lifecycleManagers.values().forEach(m -> {
+            if (m.getState() instanceof VirtualClusterLifecycleState.Serving) {
+                m.startDraining();
+            }
+            else if (m.getState() instanceof VirtualClusterLifecycleState.Failed) {
+                m.stop();
+            }
+        });
     }
 
     private void transitionAllToStopped() {
-        lifecycleManagers.values().forEach(VirtualClusterLifecycleManager::completeShutdown);
+        lifecycleManagers.values().forEach(m -> {
+            if (m.getState() instanceof VirtualClusterLifecycleState.Draining) {
+                m.drainComplete();
+            }
+        });
     }
 
     @Override

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

@@ -97,39 +97,6 @@ public void stop() {
         });
     }
 
-    /**
-     * Begins shutdown: transitions serving clusters to draining and
-     * failed/initializing clusters directly to stopped.
-     */
-    public void beginShutdown() {
-        transition(current -> {
-            if (current instanceof Serving s) {
-                return s.toDraining();
-            }
-            if (current instanceof Failed s) {
-                return s.toStopped();
-            }
-            if (current instanceof Initializing s) {
-                return s.toFailed(new IllegalStateException("Shutdown before initialization completed")).toStopped();
-            }
-            // Draining or Stopped — no-op
-            return current;
-        });
-    }
-
-    /**
-     * Completes shutdown: transitions draining clusters to stopped.
-     */
-    public void completeShutdown() {
-        transition(current -> {
-            if (current instanceof Draining s) {
-                return s.toStopped();
-            }
-            // already Stopped — no-op
-            return current;
-        });
-    }
-
     public VirtualClusterLifecycleState getState() {
         return state.get();
     }
@@ -140,8 +107,9 @@ public String getClusterName() {
 
     private void transition(UnaryOperator<VirtualClusterLifecycleState> transitionFn) {
         VirtualClusterLifecycleState previous = state.getAndUpdate(transitionFn);
-        VirtualClusterLifecycleState current = state.get();
-        LOGGER.atInfo()
+        VirtualClusterLifecycleState current = transitionFn.apply(previous);
+        var logBuilder = (current instanceof Initializing || current instanceof Stopped) ? LOGGER.atInfo() : LOGGER.atDebug();
+        logBuilder
                 .addKeyValue("virtualCluster", clusterName)
                 .addKeyValue("from", previous.getClass().getSimpleName())
                 .addKeyValue("to", current.getClass().getSimpleName())

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

@@ -130,78 +130,6 @@ void shouldHaveNoPriorFailureCauseWhenStoppedFromDraining() {
                 .isInstanceOfSatisfying(Stopped.class, stopped -> assertThat(stopped.priorFailureCause()).isNull());
     }
 
-    @Test
-    void shouldTransitionServingToDrainingOnBeginShutdown() {
-        // given
-        manager.initializationSucceeded();
-
-        // when
-        manager.beginShutdown();
-
-        // then
-        assertThat(manager.getState()).isInstanceOf(Draining.class);
-    }
-
-    @Test
-    void shouldTransitionDrainingToStoppedOnCompleteShutdown() {
-        // given
-        manager.initializationSucceeded();
-        manager.beginShutdown();
-
-        // when
-        manager.completeShutdown();
-
-        // then
-        assertThat(manager.getState()).isInstanceOf(Stopped.class);
-    }
-
-    @Test
-    void shouldTransitionFailedToStoppedOnBeginShutdown() {
-        // given
-        manager.initializationFailed(new RuntimeException("boom"));
-
-        // when
-        manager.beginShutdown();
-
-        // then
-        assertThat(manager.getState()).isInstanceOf(Stopped.class);
-    }
-
-    @Test
-    void shouldTransitionInitializingToStoppedOnBeginShutdown() {
-        // when
-        manager.beginShutdown();
-
-        // then
-        assertThat(manager.getState()).isInstanceOf(Stopped.class);
-    }
-
-    @Test
-    void shouldBeNoOpWhenBeginShutdownFromStopped() {
-        // given
-        manager.initializationFailed(new RuntimeException("boom"));
-        manager.stop();
-
-        // when
-        manager.beginShutdown();
-
-        // then
-        assertThat(manager.getState()).isInstanceOf(Stopped.class);
-    }
-
-    @Test
-    void shouldBeNoOpWhenCompleteShutdownFromStopped() {
-        // given
-        manager.initializationFailed(new RuntimeException("boom"));
-        manager.stop();
-
-        // when
-        manager.completeShutdown();
-
-        // then
-        assertThat(manager.getState()).isInstanceOf(Stopped.class);
-    }
-
     static Stream<Arguments> invalidTransitions() {
         return Stream.of(
                 Arguments.of("initializationSucceeded from SERVING", (Runnable) () -> {