Feat/idle timeouts (kroxylicious#3046)

Type of change
Enhancement / new feature
Description
Harden the proxy by disconnecting idle timeouts.

Additional Context
This PR starts to model a KafkaSession to which allows the runtime and filters to understand what stage of a connections life cycle it is in. This is somewhat higher level than the current ProxyChanellStateMachine but they do overlap.

The KafkaSession model allows this PR to draw a distinction between new and or anonymous clients and those which have been authenticated so that we can apply more relaxed timeouts to the authenticated sessions - those which are less likely to be problematic and or malicious.

Signed-off-by: Sam Barker <sam@quadrocket.co.uk>

Author: SamBarker

.idea/inspectionProfiles/Project_Default.xml

@@ -7,6 +7,7 @@ Format `<github issue/pr number>: <short description>`.
 
 ## SNAPSHOT
 
+* [#3046](https://github.com/kroxylicious/kroxylicious/pull/3046): Add configurable idle connection timeouts for client connections
 * [#3242](https://github.com/kroxylicious/kroxylicious/pull/3242): chore: remove deprecated template kek selector brace style
 * [#3224](https://github.com/kroxylicious/kroxylicious/pull/3224): Add support for using Secret in `trustAnchorRef` field of the KafkaService and the VirtualKafkaCluster CRs.
 * [#3171](https://github.com/kroxylicious/kroxylicious/pull/3171): build(deps): bump io.strimzi:api from 0.48.0 to 0.50.0
@@ -27,6 +28,14 @@ Format `<github issue/pr number>: <short description>`.
 * A JSON Web Signature (JWS) Signature validator has been added. WARNING: This validator does NOT include JSON Web Token (JWT) validation (expiration, issuer, etc. are NOT checked).
 * Curly-brace style topicName tokens are no longer supported in the Record Encryption TemplateKekSelector template. `template` should use `$(topicName)` instead of `${topicName}`.
   The was deprecated in version 0.11.0.
+* Idle connection timeout support added with two optional configuration properties under `network.proxy`:
+  * `unauthenticatedIdleTimeout` - Applies to connections where authentication cannot be detected
+  * `authenticatedIdleTimeout` - Applies to connections with established identities
+  Both properties use Go-style duration format (e.g., `30s`, `5m`, `1h30m`) with supported units: `d`, `h`, `m`, `s`, `ms`, `μs`/`us`, `ns`.
+* A new metric `kroxylicious_client_to_proxy_disconnects_total` tracks client-to-proxy disconnections with a `cause` label to distinguish between:
+  * `idle_timeout` - Connection exceeded the configured idle timeout duration
+  * `client_closed` - Client initiated the connection close
+  * `server_closed` - Backend server closed the connection, causing the proxy to close the client connection
 
 ## 0.18.0
 

CHANGELOG.md

@@ -59,6 +59,10 @@ public <P extends Principal> Optional<P> uniquePrincipalOfType(Class<P> uniquePr
         return uniquePrincipalOfType(this.principals, uniquePrincipalType);
     }
 
+    public boolean isAnonymous() {
+        return this.principals.isEmpty();
+    }
+
     private static <P extends Principal> Optional<P> uniquePrincipalOfType(Set<Principal> principals, Class<P> uniquePrincipalType) {
         if (uniquePrincipalType.isAnnotationPresent(Unique.class)) {
             return principals.stream()

kroxylicious-api/src/main/java/io/kroxylicious/proxy/authentication/Subject.java

@@ -70,4 +70,34 @@ void canExtractPrincipals() {
         Assertions.assertThat(subject2.allPrincipalsOfType(FakeUniquePrincipal.class)).isEmpty();
         Assertions.assertThat(Subject.anonymous().allPrincipalsOfType(FakeUniquePrincipal.class)).isEmpty();
     }
+
+    @Test
+    void shouldConsiderEmptySetOfPrinciplesAnonymous() {
+        // Given
+        Subject emptySubject = new Subject(Set.of());
+
+        // When
+        // Then
+        Assertions.assertThat(emptySubject.isAnonymous()).isTrue();
+    }
+
+    @Test
+    void shouldNotConsiderSetOfPrinciplesAnonymous() {
+        // Given
+        Subject emptySubject = new Subject(user1, foo);
+
+        // When
+        // Then
+        Assertions.assertThat(emptySubject.isAnonymous()).isFalse();
+    }
+
+    @Test
+    void shouldConsiderEmptySetOfPrinciplesEqual() {
+        // Given
+        Subject emptySubject = new Subject(Set.of());
+
+        // When
+        // Then
+        Assertions.assertThat(emptySubject).isEqualTo(Subject.anonymous());
+    }
 }

kroxylicious-api/src/test/java/io/kroxylicious/proxy/authentication/SubjectTest.java

@@ -17,5 +17,7 @@ include::../_modules/configuring/con-configuring-vc-target-tls.adoc[leveloffset=
 include::../_modules/configuring/con-configuring-vc-transport-subject-builder.adoc[leveloffset=+1]
 include::../_modules/configuring/con-configuring-vc-other-settings.adoc[leveloffset=+1]
 include::../_modules/configuring/con-configuring-toplevel-other-settings.adoc[leveloffset=+1]
+include::../_modules/configuring/con-configuring-network-settings.adoc[leveloffset=+1]
+include::../_modules/configuring/con-configuring-idle-timeouts.adoc[leveloffset=+1]
 
 include::../_modules/configuring/ref-configuring-proxy-example.adoc[leveloffset=+1]

kroxylicious-docs/docs/_assemblies/assembly-configuring-proxy.adoc

@@ -0,0 +1,92 @@
+:_mod-docs-content-type: CONCEPT
+
+[id='con-configuring-idle-timeouts-{context}']
+= Configuring idle connection timeouts
+
+[role="_abstract"]
+The proxy can automatically disconnect idle client connections to reclaim resources.
+Idle timeout configuration is completely optional and disabled by default, allowing you to opt in only when needed for your deployment.
+
+== Two-stage timeout mechanism
+
+The proxy supports two independent idle timeout settings that apply at different stages of the connection lifecycle:
+
+* **Unauthenticated timeout** (`unauthenticatedIdleTimeout`) - Applies to connections where the proxy cannot detect the completion of authentication. The proxy considers authentication to be complete if either of the following hold true:
+1. A transport subject builder creates a subject with an identity. Usually this would be from a client TLS certificate.
+2. A SASL inspection or termination filter has invoked `io.kroxylicious.proxy.filter.FilterContext.clientSaslAuthenticationSuccess` method.
+* **Authenticated timeout** (`authenticatedIdleTimeout`) - Applies to connections where an identity can be established. This timeout applies for the remainder of the connection's lifetime.
+
+Both timeout settings are optional and have no default values.
+You can configure one, both, or neither depending on your requirements.
+Timeout values use Go-style duration format (for example, `30s` for 30 seconds, `5m` for 5 minutes, `1h` for 1 hour).
+Supported units are: `d` (days), `h` (hours), `m` (minutes), `s` (seconds), `ms` (milliseconds), `μs` or `us` (microseconds), and `ns` (nanoseconds).
+Units can be combined, such as `1h30m` or `90s`.
+
+== Configuration examples
+
+.Example: Unauthenticated timeout only
+[source,yaml]
+----
+network:
+  proxy:
+    unauthenticatedIdleTimeout: 30s # <1>
+virtualClusters:
+  # ...
+----
+<1> Disconnect connections that remain unauthenticated for more than 30 seconds.
+
+.Example: Authenticated timeout only
+[source,yaml]
+----
+network:
+  proxy:
+    authenticatedIdleTimeout: 5m # <1>
+virtualClusters:
+  # ...
+----
+<1> Disconnect authenticated connections that are idle for more than 5 minutes.
+
+.Example: Both timeouts configured
+[source,yaml]
+----
+network:
+  proxy:
+    unauthenticatedIdleTimeout: 30s # <1>
+    authenticatedIdleTimeout: 10m # <2>
+virtualClusters:
+  # ...
+----
+<1> Disconnect unauthenticated connections after 30 seconds of inactivity.
+<2> Disconnect authenticated connections after 10 minutes of inactivity.
+
+== When to enable idle timeouts
+
+Consider enabling idle timeouts in the following scenarios:
+
+* **Misbehaving clients** - Clients that abandon connections without properly closing them, leaving resources allocated unnecessarily.
+* **High-scale deployments** - Environments with many clients where connection resources (memory, file descriptors) are constrained.
+* **Connection exhaustion prevention** - Deployments approaching operating system or network limits on concurrent connections.
+* **Network infrastructure requirements** - Environments where network infrastructure (firewalls, load balancers) drops idle connections, and you want the proxy to disconnect gracefully first.
+* **Different security postures** - Scenarios where unauthenticated connections require stricter timeouts than authenticated connections for security reasons.
+
+== When not to enable idle timeouts
+
+Avoid enabling idle timeouts in the following scenarios:
+
+* **Legitimate idle connections** - Applications where clients maintain long-lived connections with extended idle periods, such as consumers with long poll timeouts or applications using connection pooling.
+* **Stable network infrastructure** - Environments with reliable network infrastructure and no issues with idle connection management.
+* **Minimal overhead desired** - Deployments where the proxy's monitoring overhead should be kept to an absolute minimum.
+* **No resource constraints** - Systems with ample connection resources and no risk of connection exhaustion.
+
+== Monitoring idle disconnects
+
+The proxy tracks idle disconnects using the `kroxylicious_client_to_proxy_disconnects_total` metric with `cause="idle_timeout"`.
+This counter is incremented each time a connection is closed due to exceeding the configured idle timeout.
+
+The `kroxylicious_client_to_proxy_disconnects_total` metric also tracks other disconnect scenarios:
+
+* `cause="idle_timeout"` - Connection exceeded the configured idle timeout duration
+* `cause="client_closed"` - The downstream client initiated the connection close
+* `cause="server_closed"` - The upstream node closed the connection, causing the proxy to close the client connection
+
+For more information about connection metrics, see xref:con-prometheus-metrics-proxy-{context}[Overview of proxy metrics].

kroxylicious-docs/docs/_modules/configuring/con-configuring-idle-timeouts.adoc

@@ -0,0 +1,31 @@
+:_mod-docs-content-type: CONCEPT
+
+[id='con-configuring-network-settings-{context}']
+= Configuring network and Netty settings
+
+[role="_abstract"]
+The proxy allows configuration of network settings for both the proxy endpoints (client-facing) and management endpoints using the `network.proxy` and `network.management` properties.
+These settings control low-level Netty behavior and are optional, with sensible defaults provided.
+
+.Configuration fragment showing network settings
+[source,yaml]
+----
+network:
+  proxy: # <1>
+    workerThreadCount: 8 # <2>
+    shutdownQuietPeriodSeconds: 10 # <3>
+  management: # <4>
+    workerThreadCount: 2
+    shutdownQuietPeriodSeconds: 5
+virtualClusters:
+  # ...
+----
+<1> Network settings for the proxy endpoints that handle client connections.
+<2> Optional: Number of Netty worker threads for handling concurrent connections. Defaults to twice the number of available processors.
+<3> Optional: Grace period in seconds during which the proxy continues processing existing connections before shutting down. Defaults to 0.
+<4> Network settings for the management HTTP endpoints. Can be configured independently from proxy settings.
+
+* All network settings are optional. The proxy will use sensible defaults if not specified.
+* The `workerThreadCount` setting allows tuning for high-concurrency deployments. Increasing this value can improve throughput when handling many simultaneous client connections.
+* The `shutdownQuietPeriodSeconds` setting provides a graceful shutdown window, allowing in-flight requests to complete before the proxy terminates.
+* Proxy and management endpoints can have different thread pool sizes and shutdown behaviors based on their different workload characteristics.

kroxylicious-docs/docs/_modules/configuring/con-configuring-network-settings.adoc

@@ -34,12 +34,21 @@ For these specific errors, the `virtual_cluster` and `node_id` labels are set to
 
 NOTE: Error conditions signaled _within_ the Kafka protocol response (such as `RESOURCE_NOT_FOUND` or `UNKNOWN_TOPIC_ID`) are not classed as errors by these metrics.
 
-=== Understanding connection counter vs gauge metrics
+=== Understanding connection metrics relationships
 
-The proxy provides both counter and gauge metrics for connections:
+The proxy provides several related metrics for tracking connections:
 
 * **Connection counters** (`kroxylicious_*_connections_total`) track the total number of connection attempts over time. These values only increase and provide a historical view of connection activity.
 * **Active connection gauges** (`kroxylicious_*_active_connections`) show the current number of open connections at any given moment. These values increase when connections are established and decrease when connections are closed.
+* **Error counters** (`kroxylicious_*_errors_total`) track connections that closed due to errors.
+* **Disconnect counters** (`kroxylicious_client_to_proxy_disconnects_total`) track connections that closed without errors, categorized by cause.
+
+When a connection closes, it increments either the error counter or one of the disconnect counter causes, but never both.
+The active connection gauge decreases regardless of whether the closure was due to an error or a clean disconnect.
+
+The following relationship holds:
+
+`Active connections = Connections total - (Errors total + sum of all Disconnect causes)`
 
 .Connection metrics for client and broker interactions
 |===
@@ -78,6 +87,15 @@ The proxy provides both counter and gauge metrics for connections:
 |`virtual_cluster`, `node_id`
 |Shows the current number of active TCP connections from the proxy to servers. +
  This gauge reflects real-time connection state and decreases when connections are closed.
+
+|`kroxylicious_client_to_proxy_disconnects_total`
+|Counter
+|`virtual_cluster`, `node_id`, `cause`
+|Incremented by one every time a client connection is closed by the proxy. The `cause` label indicates the reason for disconnection: +
+ `idle_timeout` - Connection exceeded the configured idle timeout duration (requires idle timeouts configured via `network.proxy.unauthenticatedIdleTimeout` or `network.proxy.authenticatedIdleTimeout`). +
+ `client_closed` - Client initiated the connection close. +
+ `server_closed` - Backend server closed the connection, causing the proxy to close the client connection. +
+ Note: Error-based disconnects are tracked separately via `kroxylicious_client_to_proxy_errors_total`, not this metric.
 |===
 
 == Message metrics

kroxylicious-docs/docs/_modules/monitoring/con-prometheus-metrics-proxy.adoc

@@ -25,6 +25,7 @@
 import org.apache.kafka.common.protocol.Errors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.spi.LoggingEventBuilder;
 
 import io.kroxylicious.proxy.authentication.ClientSaslContext;
 import io.kroxylicious.proxy.authentication.SaslSubjectBuilder;
@@ -315,7 +316,6 @@ private CompletionStage<ResponseFilterResult> processSuccessfulAuthenticateRespo
                         .setMessage(
                                 "Server has accepted an expired SASL credentials on channel {}. Client must re-authenticate on the next request, or the server will disconnect.")
                         .addArgument(context::channelDescriptor)
-                        .addArgument(authorizationIdFromClient)
                         .log();
                 context.clientSaslAuthenticationFailure(state.saslObserver().mechanismName(), authorizationIdFromClient, new SaslException("expired credential"));
             }
@@ -372,6 +372,13 @@ public String authorizationId() {
                 e = exception;
             }
             else {
+                LoggingEventBuilder eventBuilder = LOGGER.atWarn()
+                        .setMessage("Exception caught while trying to build subject (enable debug to see the stacktrace). {}")
+                        .addArgument(throwable.getMessage());
+                if (LOGGER.isDebugEnabled()) {
+                    eventBuilder = eventBuilder.setCause(throwable);
+                }
+                eventBuilder.log();
                 e = new SubjectBuildingException("SaslSubjectBuilder " + subjectBuilder.getClass() + " threw an unexpected exception", throwable);
             }
             context.clientSaslAuthenticationFailure(saslObserver.mechanismName(),

kroxylicious-filters/kroxylicious-sasl-inspection/src/main/java/io/kroxylicious/filter/sasl/inspection/SaslInspectionFilter.java

@@ -78,6 +78,11 @@ private RequiringHandshakeRequest() {
         public AwaitingHandshakeResponse nextState(SaslObserver saslObserver) {
             return new AwaitingHandshakeResponse(saslObserver, NegotiationType.INITIAL);
         }
+
+        @Override
+        public String toString() {
+            return this.getClass().getSimpleName();
+        }
     }
 
     /** We're waiting for a SASL handshake response from the server. */
@@ -98,6 +103,11 @@ public RequiringAuthenticateRequest nextState() {
             return new RequiringAuthenticateRequest(saslObserver(), negotiationType);
         }
 
+        @Override
+        public String toString() {
+            return this.getClass().getSimpleName();
+        }
+
     }
 
     /**
@@ -123,6 +133,11 @@ public AwaitingAuthenticateResponse nextState(boolean authRequestApiSupportsReau
             var clientSupportsReauthentication = negotiationType == NegotiationType.REAUTH || authRequestApiSupportsReauth;
             return new AwaitingAuthenticateResponse(saslObserver(), negotiationType, clientSupportsReauthentication);
         }
+
+        @Override
+        public String toString() {
+            return this.getClass().getSimpleName();
+        }
     }
 
     /**
@@ -156,6 +171,11 @@ State nextState(boolean saslFinished) {
                 return new RequiringAuthenticateRequest(saslObserver(), negotiationType);
             }
         }
+
+        @Override
+        public String toString() {
+            return this.getClass().getSimpleName();
+        }
     }
 
     /**
@@ -182,6 +202,10 @@ public AwaitingHandshakeResponse nextState(SaslObserver saslObserver) {
             return new AwaitingHandshakeResponse(saslObserver, NegotiationType.REAUTH);
         }
 
+        @Override
+        public String toString() {
+            return this.getClass().getSimpleName();
+        }
     }
 
     /**
@@ -191,6 +215,11 @@ public AwaitingHandshakeResponse nextState(SaslObserver saslObserver) {
     final class DisallowingAuthenticateRequest implements State {
         private DisallowingAuthenticateRequest() {
         }
+
+        @Override
+        public String toString() {
+            return this.getClass().getSimpleName();
+        }
     }
 
     enum NegotiationType {