SamBarker
diff --git a/‎kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/KafkaProxy.java‎
Lines changed: 28 additions & 0 deletions b/‎kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/KafkaProxy.java‎
Lines changed: 28 additions & 0 deletions
diff --git a/‎kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/VirtualClusterLifecycleManager.java‎
Lines changed: 127 additions & 0 deletions b/‎kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/VirtualClusterLifecycleManager.java‎
Lines changed: 127 additions & 0 deletions
diff --git a/‎kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/VirtualClusterLifecycleState.java‎
Lines changed: 96 additions & 0 deletions b/‎kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/VirtualClusterLifecycleState.java‎
Lines changed: 96 additions & 0 deletions
@@ -57,6 +57,8 @@
 import io.kroxylicious.proxy.internal.KafkaProxyInitializer;
 import io.kroxylicious.proxy.internal.MeterRegistries;
 import io.kroxylicious.proxy.internal.PortConflictDetector;
+import io.kroxylicious.proxy.internal.VirtualClusterLifecycleManager;
+import io.kroxylicious.proxy.internal.VirtualClusterManager;
 import io.kroxylicious.proxy.internal.admin.ManagementInitializer;
 import io.kroxylicious.proxy.internal.config.Features;
 import io.kroxylicious.proxy.internal.net.DefaultNetworkBindingOperationProcessor;
@@ -149,6 +151,7 @@ private static int resolveThreadCount(Configuration configuration, Function<Netw
     private final NetworkBindingOperationProcessor bindingOperationProcessor = new DefaultNetworkBindingOperationProcessor();
     private final EndpointRegistry endpointRegistry = new EndpointRegistry(bindingOperationProcessor);
     private final PluginFactoryRegistry pfr;
+    private final VirtualClusterManager virtualClusterManager;
     private @Nullable MeterRegistries meterRegistries;
     private @Nullable FilterChainFactory filterChainFactory;
     private @Nullable EventGroupConfig managementEventGroup;
@@ -160,6 +163,10 @@ public KafkaProxy(PluginFactoryRegistry pfr, Configuration config, Features feat
         this.virtualClusterModels = config.virtualClusterModel();
         this.managementConfiguration = config.management();
         this.micrometerConfig = config.getMicrometer();
+        this.virtualClusterManager = new VirtualClusterManager(virtualClusterModels, (clusterName, cause) -> STARTUP_SHUTDOWN_LOGGER.atWarn()
+                .addKeyValue("virtualCluster", clusterName)
+                .addKeyValue("error", cause.map(Throwable::getMessage).orElse(null))
+                .log("Virtual cluster reached terminal stopped state, proxy shutdown required"));
     }
 
     @VisibleForTesting
@@ -216,6 +223,7 @@ public KafkaProxy startup() {
 
             STARTUP_SHUTDOWN_LOGGER.atInfo()
                     .log("Kroxylicious is starting");
+
             meterRegistries = new MeterRegistries(pfr, micrometerConfig);
             initVersionInfoMetric();
 
@@ -252,6 +260,8 @@ public KafkaProxy startup() {
                             .toArray(CompletableFuture[]::new))
                     .join();
 
+            virtualClusterModels.forEach(model -> virtualClusterManager.initializationSucceeded(model.getClusterName()));
+
             STARTUP_SHUTDOWN_LOGGER.atInfo()
                     .log("Kroxylicious is started");
             return this;
@@ -260,6 +270,11 @@ public KafkaProxy startup() {
             STARTUP_SHUTDOWN_LOGGER.atError()
                     .setCause(e)
                     .log("Exception during startup, shutting down");
+            // TODO: the onVirtualClusterStopped callback should drive the serve:none policy (triggering proxy shutdown)
+            // rather than relying on the caller to call shutdown() separately. Currently the callback only logs.
+            // All VCs are failed with the same exception because startup is all-or-nothing:
+            // initializationSucceeded is only called after all VCs register successfully (line 263).
+            virtualClusterModels.forEach(model -> virtualClusterManager.initializationFailed(model.getClusterName(), e));
             shutdown();
             throw new LifecycleException("Startup completed exceptionally", e);
         }
@@ -357,6 +372,7 @@ public void shutdown() {
         try {
             STARTUP_SHUTDOWN_LOGGER.atInfo()
                     .log("Shutting down");
+            virtualClusterManager.transitionAllToDraining();
             endpointRegistry.shutdown().handle((u, t) -> {
                 bindingOperationProcessor.close();
                 var closeFutures = new ArrayList<Future<?>>();
@@ -378,6 +394,7 @@ public void shutdown() {
                 }
                 return null;
             }).toCompletableFuture().join();
+            virtualClusterManager.completeDraining();
             if (meterRegistries != null) {
                 meterRegistries.close();
             }
@@ -394,6 +411,17 @@ public void shutdown() {
         }
     }
 
+    /**
+     * Returns the lifecycle manager for the given virtual cluster name.
+     * @param clusterName the virtual cluster name
+     * @return the lifecycle manager, or null if no cluster with that name exists
+     */
+    @VisibleForTesting
+    @Nullable
+    VirtualClusterLifecycleManager lifecycleManagerFor(String clusterName) {
+        return virtualClusterManager.lifecycleManagerFor(clusterName);
+    }
+
     @Override
     public void close() throws Exception {
         if (running.get()) {
 
@@ -0,0 +1,127 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.internal;
+
+import java.util.Objects;
+import java.util.function.UnaryOperator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.kroxylicious.proxy.internal.VirtualClusterLifecycleState.Draining;
+import io.kroxylicious.proxy.internal.VirtualClusterLifecycleState.Failed;
+import io.kroxylicious.proxy.internal.VirtualClusterLifecycleState.Initializing;
+import io.kroxylicious.proxy.internal.VirtualClusterLifecycleState.Serving;
+import io.kroxylicious.proxy.internal.VirtualClusterLifecycleState.Stopped;
+
+/**
+ * Manages the lifecycle state of a single virtual cluster.
+ * <p>
+ * Thread-safe: state transitions are performed under synchronization so that the
+ * read-transition-log sequence is atomic. This is sufficient for the single-shot
+ * proxy lifecycle where contention is not a concern.
+ * </p>
+ */
+public class VirtualClusterLifecycleManager {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(VirtualClusterLifecycleManager.class);
+
+    private final String clusterName;
+    private VirtualClusterLifecycleState state = new Initializing();
+
+    public VirtualClusterLifecycleManager(String clusterName) {
+        this.clusterName = Objects.requireNonNull(clusterName);
+    }
+
+    /**
+     * Transitions from {@link Initializing} to {@link Serving}.
+     */
+    public void initializationSucceeded() {
+        transition(current -> {
+            if (current instanceof Initializing s) {
+                return s.toServing();
+            }
+            throw unexpectedState(current, "initializationSucceeded");
+        });
+    }
+
+    /**
+     * Transitions from {@link Initializing} to {@link Failed}, recording the cause.
+     */
+    public void initializationFailed(Throwable cause) {
+        Objects.requireNonNull(cause);
+        transition(current -> {
+            if (current instanceof Initializing s) {
+                return s.toFailed(cause);
+            }
+            throw unexpectedState(current, "initializationFailed");
+        });
+    }
+
+    /**
+     * Transitions from {@link Serving} to {@link Draining}.
+     */
+    public void startDraining() {
+        transition(current -> {
+            if (current instanceof Serving s) {
+                return s.toDraining();
+            }
+            throw unexpectedState(current, "startDraining");
+        });
+    }
+
+    /**
+     * Transitions from {@link Draining} to {@link Stopped}.
+     */
+    public void drainComplete() {
+        transition(current -> {
+            if (current instanceof Draining s) {
+                return s.toStopped();
+            }
+            throw unexpectedState(current, "drainComplete");
+        });
+    }
+
+    /**
+     * 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");
+        });
+    }
+
+    public synchronized VirtualClusterLifecycleState state() {
+        return state;
+    }
+
+    public String clusterName() {
+        return clusterName;
+    }
+
+    private synchronized void transition(UnaryOperator<VirtualClusterLifecycleState> transitionFn) {
+        VirtualClusterLifecycleState previous = state;
+        state = transitionFn.apply(state);
+        var logBuilder = (state instanceof Serving || state instanceof Failed || state instanceof Stopped) ? LOGGER.atInfo() : LOGGER.atDebug();
+        logBuilder
+                .addKeyValue("virtualCluster", clusterName)
+                .addKeyValue("from", () -> previous.getClass().getSimpleName())
+                .addKeyValue("to", () -> state.getClass().getSimpleName())
+                .log("Virtual cluster lifecycle transition");
+    }
+
+    private IllegalStateException unexpectedState(VirtualClusterLifecycleState current, String operation) {
+        return new IllegalStateException(
+                "Cannot " + operation + " for virtual cluster '" + clusterName + "' in state " + current.getClass().getSimpleName());
+    }
+}
@@ -0,0 +1,96 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.internal;
+
+import java.util.Objects;
+
+import edu.umd.cs.findbugs.annotations.Nullable;
+
+/**
+ * Lifecycle states for a virtual cluster.
+ * <p>
+ * Each virtual cluster has exactly one state at any time. States are immutable
+ * records; transitions produce a new state instance.
+ * </p>
+ *
+ * @see <a href="https://github.com/kroxylicious/design/blob/main/proposals/016-virtual-cluster-lifecycle.md">Proposal 016</a>
+ */
+public sealed interface VirtualClusterLifecycleState {
+
+    /** The cluster is being set up. Not yet serving traffic. */
+    record Initializing() implements VirtualClusterLifecycleState {
+
+        /**
+         * Configuration applied successfully.
+         * @return the new serving state
+         */
+        public Serving toServing() {
+            return new Serving();
+        }
+
+        /**
+         * Configuration could not be applied.
+         * @param cause the failure reason
+         * @return the new failed state
+         */
+        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. */
+    record Serving() implements VirtualClusterLifecycleState {
+
+        /**
+         * The cluster is being shut down or reconfigured.
+         * @return the new draining state
+         */
+        public Draining toDraining() {
+            return new Draining();
+        }
+    }
+
+    /** New connections are rejected. Existing in-flight requests are completing. */
+    record Draining() implements VirtualClusterLifecycleState {
+
+        /**
+         * Drain complete, cluster permanently removed.
+         * @return the new stopped state
+         */
+        public Stopped toStopped() {
+            return new Stopped(null);
+        }
+
+    }
+
+    /** Configuration was not viable. All resources have been released. */
+    record Failed(Throwable cause) implements VirtualClusterLifecycleState {
+
+        public Failed {
+            Objects.requireNonNull(cause);
+        }
+
+        /**
+         * The cluster is being permanently removed.
+         * @return the new stopped state, retaining the failure cause for diagnostics
+         */
+        public Stopped toStopped() {
+            return new Stopped(cause);
+        }
+    }
+
+    /** Terminal state. The cluster has been permanently removed. */
+    record Stopped(@Nullable Throwable priorFailureCause) implements VirtualClusterLifecycleState {}
+}