# Admin

> KafkaAdminClient for topics, partitions, consumer groups, and configs.

Source: https://docs.pyrula.io/kafka/admin/

`KafkaAdminClient` manages topics, partitions, consumer groups, configs, offsets and ACLs.
Every call returns an [`Either`](https://docs.pyrula.io/primitives/either.md): `Ok(result)` or `Err(AdminError)`.
Each request is bounded (30 s by default).

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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 `AclResource`s 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

```python
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`.