Page History

...

Current State: Under Discussion

Motivation

Kafka brokers authenticate every client connection and store the authenticated principal (KafkaPrincipal) in memory on each KafkaChannel. However, there is no admin API, CLI command, JMX MBean, or log output that allows an operator to answer:

...

This is a fundamental observability gap. Every comparable system provides this capability:

Use Cases

• Security incident response: When a SASL credential is compromised, operators cannot determine if the compromised user has active connections without restarting brokers or using indirect workarounds.
• Credential rotation: During planned rotation, there is no way to verify that old credentials are no longer in use on active connections.
• Audit and compliance: Regulatory requirements mandate the ability to report who is connected to a system at any point in time.
• Capacity planning: Per-principal connection counts help with quota tuning and resource allocation.
• Debugging: Correlating connections with authenticated identities during troubleshooting.

...

SocketServer
  └── NetworkProcessorDataPlaneAcceptor (one per listener endpoint)
        └── Processor (one per network thread)
              └── Selector
                    └── channels: Map[String, KafkaChannel]
                          └── KafkaChannel
                                ├── principal(): KafkaPrincipal    ← authenticated user
                                ├── socketAddress(): InetAddress      ← client IP
                                ├── channelMetadataRegistry
                                │     └── clientInformation         ← software name/version (KIP-511)
                                └── id: String                      ← connection ID

The data is simply not surfaced through any external interface.

Existing Workarounds

Related Work

• KIP-511 (shipped, Kafka 2.4): Exposes client software name/version via JMX metrics — aggregate connection counts by client type only. No principals, no individual connections.
• KIP-714 (shipped, Kafka 3.7): Client-pushed telemetry via PushTelemetry RPC. Client-side metrics, not server-side connection listing. Requires client opt-in.
• KIP-1000 (accepted): ListClientMetricsResources API — lists telemetry configuration resources, not connections.
• KIP-567 (stalled): Kafka Cluster Audit — audit log of operations, not real-time connection state.
• KIP-1313 (under discussion): ClientInstanceId in all request headers — enriches request tracing but provides no API to query active connections.

...

{
  "apiKey": TBD93,
  "type": "request",
  "name": "ListClientConnectionsRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "PrincipalFilter", "type": "string", "versions": "0+",
      "nullableVersions": "0+",
      "about": "If set, only return connections for this principal. Null returns all." },
    { "name": "ClientAddressFilter", "type": "string", "versions": "0+",
      "nullableVersions": "0+",
      "about": "If set, only return connections from this client address. Null returns all." },
    { "name": "ListenerFilter", "type": "string", "versions": "0+",
      "nullableVersions": "0+",
      "about": "If set, only return connections on this listener. Null returns all." }
  ]
}

...

{
  "apiKey": TBD93,
  "type": "response",
  "name": "ListClientConnectionsResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "Duration in milliseconds for which the request was throttled." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+" },
    { "name": "Connections", "type": "[]ConnectionInfo", "versions": "0+",
      "about": "List of active connections on this broker.",
      "fields": [
        { "name": "ConnectionId", "type": "string", "versions": "0+",
          "about": "The broker-assigned connection ID." },
        { "name": "Principal", "type": "string", "versions": "0+",
          "about": "The authenticated principal, e.g. User:alice." },
        { "name": "PrincipalType", "type": "string", "versions": "0+",
          "about": "The principal type, e.g. User." },
        { "name": "ClientAddress", "type": "string", "versions": "0+",
          "about": "The remote client IP address." },
        { "name": "ClientPort", "type": "int32", "versions": "0+",
          "about": "The remote client port." },
        { "name": "ListenerName", "type": "string", "versions": "0+",
          "about": "The listener the client connected to." },
        { "name": "SecurityProtocol", "type": "string", "versions": "0+",
          "about": "The security protocol: PLAINTEXT, SSL, SASL_PLAINTEXT, or SASL_SSL." },
        { "name": "SoftwareName", "type": "string", "versions": "0+",
          "about": "The client software name (from ApiVersionsRequest, per KIP-511)." },
        { "name": "SoftwareVersion", "type": "string", "versions": "0+",
          "about": "The client software version (from ApiVersionsRequest, per KIP-511)." },
        { "name": "ConnectedSince", "type": "int64", "versions": "0+",
          "about": "Timestamp (epoch ms) when the connection was established." }
      ]
    }
  ]
}

...

1. Iterates over all DataPlaneAcceptor instances (one per listener endpoint)
2. For each acceptor, iterates its Processor pool
3. For each processor, calls selector.channels() to get a snapshot of open KafkaChannel objects
4. For each channel, extracts: id, principal(), socketAddress(), socketPort(), channelMetadataRegistry().clientInformation()
5. Returns a Seqcollection of ConnectionInfo objects with all extracted data

The handler in KafkaApis receives a reference to SocketServer (added as a constructor parameter), calls collectClientConnections(), applies request filters, and builds the response.

...

The implementation adds a collectClientConnections() method on SocketServer that iterates all processors and their selectors, collecting channel metadata into a response-ready structure. This keeps the access pattern encapsulated within the kafka.network package where Selector is accessible (it is package-private to kafka.network).

ConnectedSince Tracking

...

The SoftwareName and SoftwareVersion fields are populated from data sent in the client's ApiVersionsRequest (per KIP-511, shipped in Kafka 2.4). Connections that have not yet completed the ApiVersions handshake (e.g., raw TCP connections or very early in the connection lifecycle) will report empty strings for these fields.

Key Source Files

Reference Implementation

A working POC patch is available at: KIP-1329 POC branch. The POC validates:

...

This API exposes information about authenticated connections, including principals and client addresses.

This is consistent with other cluster-level describe operations (e.g., DescribeCluster, ListClientMetricsResources).

...