Package io.kroxylicious.proxy.internal

Class ClientConnectionStateMachine

java.lang.Object
io.kroxylicious.proxy.internal.ClientConnectionStateMachine
public class ClientConnectionStateMachine extends Object

The state machine for a single client's proxy session. The "session state" is held in the state field and is represented by an immutable subclass of ClientConnectionState which contains state-specific data. Events which cause state transitions are represented by the on*() family of methods. Depending on the transition the frontend or backend handlers may get notified via one if their in*() methods. 

   «start»
      │
      ↓ frontend.channelActive
     ClientActive ╌╌╌╌⤍ error ╌╌╌╌⤍
  ╭───┤
  ↓   ↓ frontend.channelRead receives a PROXY header
  │  HaProxy ╌╌╌╌⤍ error ╌╌╌╌⤍
  ╰───┤
      ↓ frontend.channelRead receives any KRPC request
     Forwarding ╌╌╌╌⤍ error ╌╌╌╌⤍
  ╭───┤
  │   ↓ onDraining
  │  Draining ╌╌╌╌⤍ error ╌╌╌╌⤍
  │   │ onDrainCompleted (drained naturally)
  │   │ or onDrainTimeout (force-closed after timeout)
  ╰───┤
      ↓ backend.channelInactive
      ↓ or frontend.channelInactive
     Closed ⇠╌╌╌╌ error ⇠╌╌╌╌

The Draining state is optional: a connection only enters it when drain(Duration) is invoked externally (typically by VirtualClusterLifecycle during proxy shutdown or virtual-cluster hot-reload). The on* methods that perform the actual state transitions (onDraining, onDrainCompleted, onDrainTimeout) are private and orchestrated internally by drain. Any channelInactive or error event that arrives while in Draining routes through toClosed(java.lang.Throwable) the same way it would from Forwarding; the merged-edge label applies to both paths.

In addition to the "session state" this class manages the client-side of TCP backpressure via the clientReadsBlocked field. Server-side backpressure is managed by the ServerConnectionStateMachine.

When either side of the proxy starts applying back pressure the proxy should propagate that fact to the other peer(s). Thus, when the proxy is notified that any peer is applying back pressure it results in action on the channels with the opposite peer(s). Concretely this means:

  • When any server channel becomes unwritable, client reads are paused (don't accept requests we can't forward).
  • Client reads resume only when all server channels are writable.
  • When the client channel becomes unwritable, reads are paused on all server channels (don't accept responses we can't deliver).

  • Constructor Details

    • ClientConnectionStateMachine

      public ClientConnectionStateMachine(EndpointBinding endpointBinding, io.kroxylicious.proxy.authentication.TransportSubjectBuilder transportSubjectBuilder, KafkaSession kafkaSession)

  • Method Details

    • toString

      public String toString()
      Overrides:
      toString in class Object

    • currentState

      public String currentState()

    • onClientUnwritable

      public void onClientUnwritable()
      Notify the state machine when the client applies back pressure.

    • onClientWritable

      public void onClientWritable()
      Notify the state machine when the client stops applying back pressure

    • sessionId

      public String sessionId()
      Returns:
      Return the session ID which connects a frontend channel with a backend channel

    • kafkaSession

      public KafkaSession kafkaSession()
      Returns:
      Return the session for this connection.

    • onSessionTransportAuthenticated

      public void onSessionTransportAuthenticated()

    • onSessionSaslAuthenticated

      public void onSessionSaslAuthenticated()

    • clientTlsContext

      public Optional<io.kroxylicious.proxy.tls.ClientTlsContext> clientTlsContext()

    • clientSaslAuthenticationSuccess

      public void clientSaslAuthenticationSuccess(String mechanism, io.kroxylicious.proxy.authentication.Subject subject)

    • clientSaslContext

      public Optional<io.kroxylicious.proxy.authentication.ClientSaslContext> clientSaslContext()

    • clientSaslAuthenticationFailure

      public void clientSaslAuthenticationFailure()

    • onClientTlsHandshakeSuccess

      public void onClientTlsHandshakeSuccess(SSLSession sslSession)

    • authenticatedSubject

      public io.kroxylicious.proxy.authentication.Subject authenticatedSubject()

    • clientChannel

      @Nullable public io.netty.channel.Channel clientChannel()

    • forwardToRoute

      public void forwardToRoute(String routeName, Object msg)
      Forward a message to the backend connection for the named route. Used by RouterDispatchHandler for both static and dynamic routing paths.

    • setRouterActive

      public void setRouterActive()
      Signals that a RouterDispatchHandler is active on this connection's pipeline. When active, responses bearing routing-range correlation IDs are not counted against the client in-flight limit (because they are synthetic, not client requests).

    • setUpstreamAddressResolver

      public void setUpstreamAddressResolver(Function<Integer,Optional<HostPort>> resolver)
      Sets the resolver used by forwardToNode(int, java.lang.String, java.lang.Object) to translate a virtual node ID to an upstream address. Must be set before any per-broker requests are sent.

    • onRoutedRequestComplete

      public void onRoutedRequestComplete()
      Signals that a dynamically-routed client request has been fully handled. Called by RouterDispatchHandler when the router's onRequest future completes and the response has been delivered to the client. Decrements the in-flight request count to maintain the 1:1 invariant even during fan-out routing.

    • forwardToNode

      public void forwardToNode(int virtualNodeId, String routeName, Object msg)
      Forward a message to the backend broker identified by the virtual node ID. Creates a new server connection if one does not already exist for the resolved upstream address.

    • onClientFilterChainComplete

      public void onClientFilterChainComplete(Object msg)
      A message has emerged from the filter chain and is ready to be forwarded to the upstream node.