Add virtual cluster lifecycle state model

Introduce VirtualClusterLifecycleState as a sealed interface with
records for each state (Initializing, Serving, Draining, Failed, Stopped).
Each state exposes typed transition methods enforcing the state machine
from proposal 016. Failed carries its cause, and Stopped retains the
prior failure cause for diagnostics.

VirtualClusterLifecycleManager wraps the state in an AtomicReference
for thread-safe transitions, with structured logging on each transition.

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/internal/VirtualClusterLifecycleManager.java

@@ -0,0 +1,122 @@
+/*
+ * 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.concurrent.atomic.AtomicReference;
+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 atomically via {@link AtomicReference}.
+ * </p>
+ */
+public class VirtualClusterLifecycleManager {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(VirtualClusterLifecycleManager.class);
+
+    private final String clusterName;
+    private final AtomicReference<VirtualClusterLifecycleState> state = new AtomicReference<>(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 from {@link Failed} to {@link Stopped}.
+     */
+    public void stop() {
+        transition(current -> {
+            if (current instanceof Failed s) {
+                return s.toStopped();
+            }
+            throw unexpectedState(current, "stop");
+        });
+    }
+
+    public VirtualClusterLifecycleState getState() {
+        return state.get();
+    }
+
+    public String getClusterName() {
+        return clusterName;
+    }
+
+    private void transition(UnaryOperator<VirtualClusterLifecycleState> transitionFn) {
+        VirtualClusterLifecycleState previous = state.getAndUpdate(transitionFn);
+        VirtualClusterLifecycleState current = state.get();
+        LOGGER.atInfo()
+                .addKeyValue("virtualCluster", clusterName)
+                .addKeyValue("from", previous.getClass().getSimpleName())
+                .addKeyValue("to", current.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());
+    }
+}

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

@@ -0,0 +1,103 @@
+/*
+ * 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);
+        }
+    }
+
+    /** 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);
+        }
+
+        /**
+         * Drain complete, cluster reinitialising with new configuration.
+         * @return the new initializing state
+         */
+        public Initializing toInitializing() {
+            return new Initializing();
+        }
+    }
+
+    /** 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);
+        }
+
+        /**
+         * A retry is requested with corrected configuration.
+         * @return the new initializing state
+         */
+        public Initializing toInitializing() {
+            return new Initializing();
+        }
+    }
+
+    /** Terminal state. The cluster has been permanently removed. */
+    record Stopped(@Nullable Throwable priorFailureCause) implements VirtualClusterLifecycleState {}
+}

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

@@ -0,0 +1,168 @@
+/*
+ * 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.stream.Stream;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class VirtualClusterLifecycleManagerTest {
+
+    private static final String CLUSTER_NAME = "test-cluster";
+    private VirtualClusterLifecycleManager manager;
+
+    @BeforeEach
+    void setUp() {
+        manager = new VirtualClusterLifecycleManager(CLUSTER_NAME);
+    }
+
+    @Test
+    void shouldStartInInitializingState() {
+        assertThat(manager.getState()).isInstanceOf(Initializing.class);
+    }
+
+    @Test
+    void shouldExposeClusterName() {
+        assertThat(manager.getClusterName()).isEqualTo(CLUSTER_NAME);
+    }
+
+    @Test
+    void shouldTransitionToServingOnSuccess() {
+        // when
+        manager.initializationSucceeded();
+
+        // then
+        assertThat(manager.getState()).isInstanceOf(Serving.class);
+    }
+
+    @Test
+    void shouldTransitionToFailedOnError() {
+        // given
+        var cause = new RuntimeException("filter init failed");
+
+        // when
+        manager.initializationFailed(cause);
+
+        // then
+        assertThat(manager.getState())
+                .isInstanceOfSatisfying(Failed.class, failed -> assertThat(failed.cause()).isSameAs(cause));
+    }
+
+    @Test
+    void shouldTransitionFromServingToDraining() {
+        // given
+        manager.initializationSucceeded();
+
+        // when
+        manager.startDraining();
+
+        // then
+        assertThat(manager.getState()).isInstanceOf(Draining.class);
+    }
+
+    @Test
+    void shouldTransitionFromDrainingToStopped() {
+        // given
+        manager.initializationSucceeded();
+        manager.startDraining();
+
+        // when
+        manager.drainComplete();
+
+        // then
+        assertThat(manager.getState()).isInstanceOf(Stopped.class);
+    }
+
+    @Test
+    void shouldTransitionFromFailedToStopped() {
+        // given
+        manager.initializationFailed(new RuntimeException("boom"));
+
+        // when
+        manager.stop();
+
+        // then
+        assertThat(manager.getState()).isInstanceOf(Stopped.class);
+    }
+
+    @Test
+    void shouldRetainFailureCauseAfterStop() {
+        // given
+        var cause = new RuntimeException("boom");
+        manager.initializationFailed(cause);
+
+        // when
+        manager.stop();
+
+        // then
+        assertThat(manager.getState())
+                .isInstanceOfSatisfying(Stopped.class, stopped -> assertThat(stopped.priorFailureCause()).isSameAs(cause));
+    }
+
+    @Test
+    void shouldHaveNoPriorFailureCauseWhenStoppedFromDraining() {
+        // given
+        manager.initializationSucceeded();
+        manager.startDraining();
+
+        // when
+        manager.drainComplete();
+
+        // then
+        assertThat(manager.getState())
+                .isInstanceOfSatisfying(Stopped.class, stopped -> assertThat(stopped.priorFailureCause()).isNull());
+    }
+
+    static Stream<Arguments> invalidTransitions() {
+        return Stream.of(
+                Arguments.of("initializationSucceeded from SERVING", (Runnable) () -> {
+                    var m = new VirtualClusterLifecycleManager("c");
+                    m.initializationSucceeded();
+                    m.initializationSucceeded();
+                }),
+                Arguments.of("startDraining from INITIALIZING", (Runnable) () -> {
+                    var m = new VirtualClusterLifecycleManager("c");
+                    m.startDraining();
+                }),
+                Arguments.of("drainComplete from SERVING", (Runnable) () -> {
+                    var m = new VirtualClusterLifecycleManager("c");
+                    m.initializationSucceeded();
+                    m.drainComplete();
+                }),
+                Arguments.of("stop from SERVING", (Runnable) () -> {
+                    var m = new VirtualClusterLifecycleManager("c");
+                    m.initializationSucceeded();
+                    m.stop();
+                }),
+                Arguments.of("initializationSucceeded from STOPPED", (Runnable) () -> {
+                    var m = new VirtualClusterLifecycleManager("c");
+                    m.initializationFailed(new RuntimeException("x"));
+                    m.stop();
+                    m.initializationSucceeded();
+                }));
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("invalidTransitions")
+    void shouldRejectInvalidTransition(String description, Runnable action) {
+        assertThatThrownBy(action::run)
+                .isInstanceOf(IllegalStateException.class);
+    }
+}