kafkareport package

Submodules

Module contents

Reports on topic sizes, watermarks, and retention policies.

class kafkareport.KafkaReport(conf: dict[str, str | bool | Callable], debug: bool = False)

Bases: object

The only class. _TIMEOUT classvar is default unless passed via kwargs to the methods.

param conf:

A dict passed to confluent_kafka’s consumer, so it should follow the same format, except for group.id. It is automatically set.

param debug:

sets the logger level to DEBUG

Typical usage:

from kafkareport import KafkaReport

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

report = KafkaReport(conf)

report.get_topicnames()

for topic in report.get_topicnames():

report.retentions(topic)

report.watermarks(topic)

report.topic_sizes()

report.watermarks(“kafkareportuno”)

get_topicnames(timeout: int = 30) → list[str]

Returns a list of all topic names from the kafka servers.

Returns:: A list of all topic names

retentions(topic: str, timeout: int = 30) → dict[str, int]

Retrieves the retention settings for given topic.

Parameters:

topic – A string of the topic name.
timeout – Seconds to wait for timeout.

Returns:

A dict of the topic’s retention settings and values.

topic_sizes() → list[dict[str, str | int]]

Retrieves the size each topic takes up on the servers.

At time of first writing, neither kafka-python nor confluent_kafka implemented the DescribeLogDirsRequest, available in java. A recent PR for kafka-python supports it, so I used that experimental code. Not pretty, but it works.

Parameters:: timeout – Seconds to wait for timeout.
Returns:: A dict of the topic’s names and sizes in bytes,

watermarks(topic: str, timeout: int = 30) → dict[str, datetime]

Retrieves the earliest and latest message times for each topic.

Spans all partitions. Launches a thread for each partition. UTC times. Empty strings if no result.

Parameters:

topic – A string of the topic name.
timeout – Seconds to wait for timeout.

Raises:

KafkaException – If there was a problem getting topic metadata.

Returns:

A dict of the topic’s earliest and latest message times.