View Source

Status

Current State: Draft
Discussion Thread: TBD
JIRA: KAFKA-20526
Release: TBD

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:

"Which user principals currently have active connections to this broker?"

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

System	Command / API
MySQL	`SHOW PROCESSLIST`
PostgreSQL	`pg_stat_activity`
RabbitMQ	Management API `/api/connections`
MongoDB	`db.currentOp()`
Apache Kafka	Nothing

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.

Current State of the Data

The broker already holds all the data in memory:

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

The data is simply not surfaced through any external interface.

Existing Workarounds

Workaround	Limitation
Set `kafka.authorizer.logger=DEBUG` dynamically	Only logs principals when they make requests that trigger authorization. Truly idle connections are invisible.
Set `kafka.request.logger=DEBUG` dynamically	Extremely verbose. Still misses connections that send zero requests.
JMX quota metrics (`kafka.server:type=,user=`)	Sensors expire after 600s of inactivity. Requires quotas to be enabled.
Heap dump (`jmap`)	Causes GC pause. Requires post-processing. Not suitable for real-time use.