Admin

KafkaAdminClient covers cluster management: topics, partitions, consumer groups, and configs. Like the rest of the client, operations return an Either.

from pyrula.kafka import KafkaAdminClient, KafkaConfig, TopicSpec

admin = KafkaAdminClient(KafkaConfig(bootstrap_servers="localhost:9092"))

result = admin.create_topics([TopicSpec("events", num_partitions=6, replication_factor=3)])
if result.is_ok():
    for topic_result in result.value.to_list():
        print(topic_result)

Topics and partitions

admin.create_topics(specs)
admin.delete_topics(["events"])
admin.create_partitions(new_partitions)
admin.describe_topics(["events"])
admin.describe_topic_config(["events"])

Consumer groups

admin.list_consumer_groups()
admin.describe_consumer_groups(["my-group"])
admin.delete_consumer_groups(["my-group"])

Cluster and offsets

admin.describe_cluster()
admin.describe_configs(resources)
admin.incremental_alter_configs(resources)
admin.list_offsets(offset_specs, isolation_level="read_committed")
admin.delete_records(offsets)

Each call returns Ok(value) with the result type for that operation, or Err with an AdminError. Check is_ok() before reading value.