Package io.kroxylicious.proxy.internal

Class VirtualClusterRegistry

java.lang.Object
io.kroxylicious.proxy.internal.VirtualClusterRegistry
All Implemented Interfaces:
AutoCloseable
public class VirtualClusterRegistry extends Object implements AutoCloseable
Owns the virtual cluster configuration tree and lifecycle state.

This is the single source of truth for which virtual clusters exist and their current lifecycle state. It does not manage networking, endpoint registration, metrics, or any Netty infrastructure — those remain with KafkaProxy.

The onVirtualClusterStopped callback notifies the owner (typically KafkaProxy) when a virtual cluster reaches the terminal VirtualClusterLifecycleState.Stopped state, allowing the owner to apply proxy-level policy (e.g. serve: none). During reload, the Draining → Initializing → Serving cycle is managed internally without involving the callback — reload never reaches Stopped.

Each virtual cluster's per-cluster state machine is a VirtualClusterLifecycle.

  • Constructor Details

    • VirtualClusterRegistry

      public VirtualClusterRegistry(List<VirtualClusterModel> virtualClusterModels, BiFunction<Configuration,String,VirtualClusterModel> rawModelResolver, BiConsumer<String,Optional<Throwable>> onVirtualClusterStopped)
      Creates a new VirtualClusterRegistry for the given set of virtual clusters.
      Parameters:
      virtualClusterModels - the complete set of virtual cluster configurations
      rawModelResolver - builds a VirtualClusterModel for a (config, clusterName) pair without any threading concerns. The registry wraps every invocation in a dispatch to the lifecycle executor — see resolveModel(io.kroxylicious.proxy.config.Configuration, java.lang.String). Captured at construction time so plugin-factory wiring is the responsibility of the registry's owner (typically KafkaProxy), not the registry itself.
      onVirtualClusterStopped - callback invoked with (clusterName, priorFailureCause) whenever a virtual cluster reaches the terminal Stopped state. The cause is empty for clean stops (e.g. drain completed during shutdown) and present for failure-driven stops. The callback must not throw exceptions.
      Throws:
      NullPointerException - if any argument is null
      IllegalArgumentException - if the list contains duplicate cluster names

  • Method Details

    • resolveModel

      public VirtualClusterModel resolveModel(Configuration config, String clusterName)
      Builds a VirtualClusterModel for the named virtual cluster, dispatched onto the lifecycle thread. Used by OperationsPlanner during reconfigure so each filter's initialize() runs on a non-event-loop thread regardless of which thread invoked reconfigure().
      Throws:
      RuntimeException - the same RuntimeException the underlying resolver threw

    • close

      public void close()
      Shuts down the lifecycle executor. Must be called after shutdownAllClusters() has drained — that method dispatches close work through the executor, so tearing the executor down first would reject straggling submissions.
      Specified by:
      close in interface AutoCloseable

    • virtualClusterModels

      public Collection<VirtualClusterModel> virtualClusterModels()
      Returns the currently-tracked virtual cluster models. The collection reflects the constructor- supplied models PLUS any added at runtime via addVirtualCluster(VirtualClusterModel).

      Iteration order is unspecified (the backing map is concurrent). Callers that need order-stable output should sort the result themselves.

      Returns:
      weakly-consistent snapshot of currently-tracked virtual cluster models

    • initializationSucceeded

      public void initializationSucceeded(String clusterName)
      Signals that the named virtual cluster initialized successfully. Transitions the cluster from Initializing to Serving.
      Parameters:
      clusterName - the virtual cluster name
      Throws:
      IllegalArgumentException - if no cluster with that name exists

    • initializationFailed

      public void initializationFailed(String clusterName, Throwable cause)
      Signals that the named virtual cluster failed to initialize. Transitions the cluster from Initializing to Failed, then immediately to Stopped (no recovery path exists today), and fires the onVirtualClusterStopped callback.
      Parameters:
      clusterName - the virtual cluster name
      cause - the failure cause
      Throws:
      IllegalArgumentException - if no cluster with that name exists

    • shutdownAllClusters

      public List<Throwable> shutdownAllClusters()
      Transitions all virtual clusters toward draining/stopped as appropriate for shutdown.
      • Serving → Draining
      • Draining → Draining (a pre-existing drain, e.g. from hot-reload, is left to complete)
      • Initializing → Stopped (fires callback with empty cause)
      • Failed → Stopped (fires callback with cause)
      • Stopped → Stopped (no-op)

    • lifecycleFor

      @Nullable public VirtualClusterLifecycle lifecycleFor(String clusterName)
      Returns the lifecycle for the given virtual cluster name.
      Parameters:
      clusterName - the virtual cluster name
      Returns:
      the lifecycle, or null if no cluster with that name exists

    • modelFor

      @Nullable public VirtualClusterModel modelFor(String clusterName)
      Returns the model for the given virtual cluster name.
      Parameters:
      clusterName - the virtual cluster name
      Returns:
      the model, or null if no cluster with that name exists

    • registerConnection

      public boolean registerConnection(String clusterName, ClientConnectionStateMachine ccsm)
      Attempts to register a new connection for clusterName.
      Returns:
      true iff the cluster is known to this registry AND its lifecycle is in a state that accepts new connections (i.e. SERVING). An unknown cluster is treated as a rejection rather than an error so that KafkaProxyInitializer's existing false → rejectConnection path covers both "not serving" and "no such cluster" without depending on the bookkeeping-vs-binding ordering invariant being preserved by future changes.

    • deregisterConnection

      public void deregisterConnection(String clusterName, ClientConnectionStateMachine ccsm)
      Decrements the active-connections count for clusterName if the cluster is no longer known to this registry. Called from a Netty channel-close listener, which can race against entry removal in a future cleanup-on-Stopped

    • removeVirtualCluster

      public CompletableFuture<Void> removeVirtualCluster(String clusterName)
      Drives an existing virtual cluster through SERVING → DRAINING → STOPPED. Invoked by ConfigurationReloadOrchestrator for clusters present in the running configuration but absent in the submitted one.

      The cluster's entry is not removed from the registry on reaching Stopped — see shutdownCluster(String, VirtualClusterLifecycle) for the rationale.

      Parameters:
      clusterName - the virtual cluster to remove; must name an existing cluster
      Returns:
      a future that completes when the cluster has reached Stopped
      Throws:
      IllegalArgumentException - if clusterName does not name a registered cluster

    • addVirtualCluster

      public CompletableFuture<Void> addVirtualCluster(VirtualClusterModel newModel)
      Creates a VirtualClusterLifecycle in INITIALIZING for the given model. Endpoint binding and the transition to SERVING are the orchestrator's responsibility — once gateway registration succeeds it calls initializationSucceeded(String); on failure it calls initializationFailed(String, Throwable) and rolls back the gateway bindings.

      If an entry already exists for this name AND its lifecycle is Stopped, the entry is replaced — this is how ReplaceCluster's add half re-establishes the cluster after the remove half drove it to Stopped. The retained-Stopped- entry policy (see shutdownCluster(java.lang.String, io.kroxylicious.proxy.internal.VirtualClusterLifecycle)) interacts with re-add by name reuse, and "replace the dead entry" is the natural reconciliation.

      Parameters:
      newModel - the model for the new cluster
      Returns:
      an already-completed future (the operation is synchronous; the CompletableFuture shape is preserved for caller symmetry with removeVirtualCluster(java.lang.String))
      Throws:
      IllegalStateException - if an entry already exists AND its lifecycle is in any state OTHER than Stopped — re-adding an actively-serving (or initializing, draining, or failed) cluster would be a contract violation. The exception message names the current state to aid diagnosis.