A reporting tool for kafka topics.

Show any or all of the following attributes for each topic in a cluster:

• sizes (DescribeLogDirsRequest)
• retention policies
• earliest/latest message timestamps (watermarks)

With larger teams and installations, misconfiguration often results in lots of resource waste. Auditing with this tool has helped.

Installation

pip install kafkareport

Usage

Tested only with SASL_SSL + SCRAM_SHA_512 u/p, and with AWS_MSK_IAM on AWS MSK provisioned. MSK Serverless is not supported because it doesn't expose DescribeLogDirs, which the size report depends on. See Addenda for details.

All times should be UTC.

Configuration

SASL/SCRAM

Json file of a dict passed to confluent_kafka's consumer, so it should follow the same format except for group.id. It is automatically set. See env.json for a sample conf file.

AWS MSK IAM

Set sasl.mechanism to the sentinel AWS_MSK_IAM and drop the sasl.username / sasl.password keys. See env-iam.json for a sample conf file.

Credentials come from the default AWS chain (env, profile, instance profile). Region resolves from AWS_REGION / AWS_DEFAULT_REGION / ~/.aws/config. KafkaReport(debug=True)/kafkareport -v also enables the MSK signer's credential debug, which logs the AWS credentials being used to sign the OAUTHBEARER token, handy when access is denied.

CLI

% kafkareport ~/aws.json --csv report.csv
+----------------+-------+----------------------------------+----------------------------------+--------------+-----------------+---------------------+
| topic          | bytes | earliest                         | latest                           | retention.ms | retention.bytes | delete.retention.ms |
+----------------+-------+----------------------------------+----------------------------------+--------------+-----------------+---------------------+
| kafkareportdue | 2330  | 2024-06-30 20:04:44.486000+00:00 | 2024-06-30 20:04:46.041000+00:00 | 16800001     | 16800001        | 16800001            |
| kafkareportuno | 2268  | 2024-06-30 20:04:42.880000+00:00 | 2024-06-30 20:04:44.431000+00:00 | 16800001     | 16800001        | 16800001            |
+----------------+-------+----------------------------------+----------------------------------+--------------+-----------------+---------------------+

% cat report.csv
topic,bytes,earliest,latest,retention.ms,retention.bytes,delete.retention.ms
kafkareportdue,2330,2024-06-30 20:04:44.486000+00:00,2024-06-30 20:04:46.041000+00:00,16800001,16800001,16800001
kafkareportuno,2268,2024-06-30 20:04:42.880000+00:00,2024-06-30 20:04:44.431000+00:00,16800001,16800001,16800001

Lib

Docs

>>> from kafkareport import KafkaReport

>>> conf = {"bootstrap.servers": "localhost:9092", "ssl.endpoint.identification.algorithm": "none"}

>>> report = KafkaReport(conf)

>>> report.get_topicnames()
dict_keys(['kafkareportdue', 'kafkareportuno', '__consumer_offsets'])

>>> for topic in report.get_topicnames():
...     print(topic)
...     report.retentions(topic)
...     report.watermarks(topic)
...
kafkareportdue
{'retention.ms': 16800001, 'retention.bytes': 16800001, 'delete.retention.ms': 16800001}
{'earliest': datetime.datetime(2024, 6, 30, 20, 8, 57, 554000, tzinfo=datetime.timezone.utc), 'latest': datetime.datetime(2024, 6, 30, 20, 8, 59, 99000, tzinfo=datetime.timezone.utc)}
kafkareportuno
{'retention.ms': 16800001, 'retention.bytes': 16800001, 'delete.retention.ms': 16800001}
{'earliest': datetime.datetime(2024, 6, 30, 20, 8, 55, 975000, tzinfo=datetime.timezone.utc), 'latest': datetime.datetime(2024, 6, 30, 20, 8, 57, 503000, tzinfo=datetime.timezone.utc)}
__consumer_offsets
{'retention.ms': 604800000, 'retention.bytes': -1, 'delete.retention.ms': 86400000}
{'earliest': '', 'latest': ''}

>>> report.topic_sizes()
[{'topic': 'kafkareportuno', 'bytes': 2690}, {'topic': 'kafkareportdue', 'bytes': 2328}, {'topic': '__consumer_offsets', 'bytes': 0}]

>>> report.watermarks("kafkareportuno")
{'earliest': datetime.datetime(2024, 6, 30, 20, 8, 55, 975000, tzinfo=datetime.timezone.utc), 'latest': datetime.datetime(2024, 6, 30, 20, 8, 57, 503000, tzinfo=datetime.timezone.utc)}

Development

This is a poetry project, so it should be butter once you get that sorted. Install pre-commit for lint/format on commit, typing on push.

pre-commit install --hook-type pre-push for typing pre-push.

Testing

Testing runs against kafka/zookeper containers (plaintext only; no auth), as you can see in the Github actions. docker compose up should do the trick.

pytest can use --conf=/some/file.json instead of default for localstack. This will manipulate pytest.topics on the kafka servers.

See Live testing against AWS MSK for details on how to quickly smoke test against actual clusters.

Occasionally, e.g. on laptop wake, the kafka container will be in a weird state. docker compose down && docker compose up always did the trick.

Addenda

Watermarks use a thread for each topic partition, but it can still take a while. -v for gory details along the way, KafkaReport(debug=True) for the lib.

This project started with confluent_kafka, but DescribeLogDirsRequest still isn't implemented in confluent_kafka or the underlying librdkafka. kafka-python has had a native describe_log_dirs since 2.1.0, but the stable 2.x signature can't target a specific broker — that landed on master/3.0.0.dev via PR #2881. Until that ships, logdirs.py monkey-patches a per-broker wrapper onto KafkaAdminClient.describe_log_dirs.

As a result, there is a janky confluent_kafka to kafka-python AdminClient conf map, now lifted into kafkareport/auth.py since IAM doubled its complexity. It handles plaintext, SCRAM, and the AWS_MSK_IAM sentinel (translated to OAUTHBEARER + an MSK signer-backed token provider for kafka-python and an oauth_cb for confluent_kafka).