Admin

KafkaAdminClient manages topics, partitions, consumer groups, configs, offsets and ACLs. Every call returns an Either: Ok(result) or Err(AdminError). Each request is bounded (30 s by default).

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)])
for topic_result in result.unwrap_or_raise().to_list():
    print(topic_result)

Topics and partitions

from pyrula.kafka import NewPartitions

admin.list_topics()                                    # IList[str]
admin.describe_topics(["events"])
admin.describe_topic_config(["events"])
admin.create_partitions([NewPartitions("events", total_count=12)])
admin.delete_topics(["events"])

create_topics and create_partitions take validate_only=True for a dry run and operation_timeout_ms (default 30000).

Consumer groups

from pyrula import IMap
from pyrula.kafka import TopicPartition

admin.list_consumer_groups()
admin.describe_consumer_groups(["billing"])
admin.list_consumer_group_offsets("billing")            # IMap[TopicPartition, int]; -1 = none committed
admin.alter_consumer_group_offsets(
    "billing", IMap({TopicPartition("events", 0): 1200}),
)                                                       # group must have no active members; offsets >= 0
admin.delete_consumer_groups(["billing"])

Offsets and records

from pyrula.kafka import OffsetSpec

admin.list_offsets(IMap({TopicPartition("events", 0): OffsetSpec.latest()}))
admin.delete_records(IMap({TopicPartition("events", 0): 1000}))   # delete before offset 1000

OffsetSpec has earliest(), latest(), max_timestamp() and for_timestamp(ms). list_offsets takes isolation_level (default "read_committed").

Configs

from pyrula import IList
from pyrula.kafka import AlterConfigOpType, ConfigEntry, ConfigResource, ResourceType

admin.describe_configs([ConfigResource(ResourceType.TOPIC, "events")])
admin.incremental_alter_configs([
    ConfigResource(
        ResourceType.TOPIC, "events",
        incremental_configs=IList([
            ConfigEntry("retention.ms", "86400000", incremental_operation=AlterConfigOpType.SET),
        ]),
    ),
])

AlterConfigOpType is SET, DELETE, APPEND or SUBTRACT.

ACLs

from pyrula.kafka import (
    AclCreation, AclFilter, AclOperation, AclPermissionType, AclResourceType,
)

admin.create_acls([
    AclCreation(AclResourceType.TOPIC, "events", "User:billing", "*",
                AclOperation.READ, AclPermissionType.ALLOW),
])
admin.describe_acls()                                    # all; narrow by any argument
admin.describe_acls(resource_type=AclResourceType.TOPIC, resource_name="events")
admin.delete_acls([AclFilter(principal="User:billing")])

describe_acls defaults to matching everything. delete_acls refuses, before sending anything, a filter that narrows nothing; to delete every binding on purpose, pass AclFilter.match_all().

describe_acls returns AclResources with their acls; create_acls and delete_acls return one AclResult per entry, and a non-empty error means that entry failed. Bindings have a pattern type: LITERAL (default for a creation) or PREFIXED. Filters default to ANY; MATCH finds every binding that applies to one named resource. Requires Kafka 2.0+.

Cluster

admin.describe_cluster()           # brokers, controller, cluster id

Threads

KafkaAdminClient can be shared across threads; each call opens its own connection. A call from the client’s background threads (error_cb, log handlers) raises AdminError with _conflicting_use.