Versions Compared

Old Version 3

changes.mady.by.user Mehari Beyene

Saved on

compared with

New Version 4

changes.mady.by.user Mehari Beyene

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

Current State: Under Discussion

Item

Value

Discussion Thread

TBD — post to dev@kafka.apache.org and link here

JIRA

KAFKA-20526

Release


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:

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.

...

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.

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.

...

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

File

Change

clients/src/main/resources/common/message/ListClientConnectionsRequest.json

New request schema

clients/src/main/resources/common/message/ListClientConnectionsResponse.json

New response schema

clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java

Register new API key (93)

clients/src/main/java/org/apache/kafka/common/requests/ListClientConnectionsRequest.java

Request wrapper class

clients/src/main/java/org/apache/kafka/common/requests/ListClientConnectionsResponse.java

Response wrapper class

clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java

Add parse case for API key 93

clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java

Add parse case for API key 93

core/src/main/scala/kafka/network/SocketServer.scala

Add collectClientConnections() method + ConnectionInfo case class

core/src/main/scala/kafka/server/KafkaApis.scala

Add socketServer param + handleListClientConnections handler

core/src/main/scala/kafka/server/BrokerServer.scala

Pass socketServer to KafkaApis

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

New admin client method (future)

tools/src/main/java/org/apache/kafka/tools/

New CLI tool (future)

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.

Operation

Resource

Permission

ListClientConnections

CLUSTER

DESCRIBE

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

...