Package io.kroxylicious.proxy.internal.net

Class EndpointRegistry

java.lang.Object
io.kroxylicious.proxy.internal.net.EndpointRegistry
All Implemented Interfaces:
EndpointBindingResolver, EndpointReconciler, AutoCloseable
public class EndpointRegistry extends Object implements EndpointReconciler, EndpointBindingResolver, AutoCloseable
The endpoint registry is responsible for associating network endpoints with broker/bootstrap addresses of virtual clusters. Pictorially the registry looks like this: 

 Registry
    ├─ Endpoint (tls 9092)
    │    ╰───┬──→ cluster1.kafka.com         ──→ Virtual Cluster A bootstrap
    │        ├──→ broker1-cluster1.kafka.com ──→ Virtual Cluster A broker node 1
    │        ╰──→ broker2-cluster1.kafka.com ──→ Virtual Cluster A broker node 2
    ├─ Endpoint (plain 19092)
    │        ╰───→  (null)                   ──→ Virtual Cluster B bootstrap
    ├─ Endpoint (plain 19093)
    │        ╰───→  (null)                   ──→ Virtual Cluster B broker 1
    ╰─ Endpoint (plain 12094)
             ╰───→  (null)                   ──→ Virtual Cluster B broker 2
Key points about the implementation: The registry is thread safe for all operations.
  • virtual cluster registration uses java.util.concurrent features to ensure registration is single threaded.
  • virtual cluster de-registration uses java.util.concurrent.atomic to ensure de-registration is single threaded.
  • virtual cluster reconciliation uses java.util.concurrent.atomic to ensure reconciliation is single threaded.
  • updates to the binding mapping (attached to channel) are made only whilst holding an intrinsic lock on the EndpointRegistry.ListeningChannelRecord.
  • updates to the binding mapping are published safely to readers (i.e. threads calling EndpointBindingResolver.resolve(Endpoint, String)). This relies the fact that the binding map uses concurrency safe data structures (ConcurrentHashMap and the exclusive use of immutable objects within it.

  • Field Details

    • NO_CHANNEL_BINDINGS_MESSAGE

      public static final String NO_CHANNEL_BINDINGS_MESSAGE
      See Also:

    • VIRTUAL_CLUSTER_CANNOT_BE_NULL_MESSAGE

      public static final String VIRTUAL_CLUSTER_CANNOT_BE_NULL_MESSAGE
      See Also:

    • OS_ASSIGNED_PORT

      public static final int OS_ASSIGNED_PORT
      Sentinel port value meaning "let the OS assign a free port at bind time". Use with KafkaProxy.listeningPort(String, int) to discover the actual port after startup.
      See Also:

    • CHANNEL_BINDINGS

      protected static final io.netty.util.AttributeKey<Map<io.kroxylicious.proxy.internal.net.EndpointRegistry.RoutingKey,EndpointBinding>> CHANNEL_BINDINGS

  • Constructor Details

  • Method Details

    • registerVirtualCluster

      public CompletionStage<Endpoint> registerVirtualCluster(EndpointGateway virtualClusterModel)
      Registers a virtual cluster with the registry. The registry refers to the endpoint configuration of the virtual cluster to understand its port requirements, binding new listening ports as necessary. This operation returns a CompletionStage<Endpoint>. The completion stage will complete once all necessary network bind operations are completed. The returned Endpoint refers to the bootstrap endpoint of the virtual cluster.
      Parameters:
      virtualClusterModel - virtual cluster to be registered.
      Returns:
      completion stage that will complete after registration is finished.

    • deregisterVirtualCluster

      public CompletionStage<Void> deregisterVirtualCluster(EndpointGateway virtualClusterModel)
      De-registers a virtual cluster from the registry, removing all existing endpoint bindings, and closing any listening sockets that are no longer required. This operation returns a CompletionStage<Endpoint>. The completion stage will complete once any necessary network unbind operations are completed.
      Parameters:
      virtualClusterModel - virtual cluster to be deregistered.
      Returns:
      completion stage that will complete after registration is finished.

    • reconcile

      public CompletionStage<Void> reconcile(EndpointGateway virtualClusterModel, Map<Integer,HostPort> upstreamNodes)
      Reconciles the current set of bindings for this virtual cluster against those required by the current set of nodes.
      Specified by:
      reconcile in interface EndpointReconciler
      Parameters:
      virtualClusterModel - virtual cluster
      upstreamNodes - current map of node id to upstream host ports
      Returns:
      CompletionStage yielding true if binding alterations were made, or false otherwise.

    • upstreamAddress

      public Optional<HostPort> upstreamAddress(EndpointGateway gateway, int upstreamNodeId)
      Description copied from interface: EndpointReconciler
      Returns the upstream address for the given upstream node ID (note this may be a physical node id, or a virtual node id if routing enabled), as last reconciled for the virtual cluster identified by gateway. Returns empty if the address is not yet known.
      Specified by:
      upstreamAddress in interface EndpointReconciler
      Parameters:
      gateway - the virtual cluster's endpoint gateway
      upstreamNodeId - the upstream node ID
      Returns:
      the upstream host and port, or empty if not yet known

    • localPortFor

      @VisibleForTesting public int localPortFor(Endpoint endpoint)

    • resolve

      public CompletionStage<EndpointBinding> resolve(Endpoint endpoint, @Nullable String sniHostname)
      Uses channel metadata (port, SNI name etc.) from the incoming connection to resolve a BootstrapEndpointBinding.
      Specified by:
      resolve in interface EndpointBindingResolver
      Parameters:
      endpoint - endpoint being resolved
      sniHostname - SNI hostname, may be null.
      Returns:
      completion stage yielding the BootstrapEndpointBinding or exceptionally a EndpointResolutionException.

    • shutdown

      public CompletionStage<Void> shutdown()
      Signals that the registry should be shut down. The operation returns a CompletionStage that will complete once shutdown is complete (endpoints unbound).
      Returns:
      CompletionStage that completes once shut-down is complete.

    • close

      public void close() throws Exception
      Specified by:
      close in interface AutoCloseable
      Throws:
      Exception