Metadata-Version: 2.4
Name: kafka-util
Version: 0.1.4
Summary: CLI tool for managing Apache Kafka clusters
License-Expression: GPL-3.0-or-later
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: typer>=0.12.0
Requires-Dist: rich<14,>=13.7.0
Requires-Dist: questionary<3,>=2.0.1
Requires-Dist: confluent-kafka<3,>=2.3.0
Requires-Dist: httpx<1,>=0.27.0
Requires-Dist: pydantic<3,>=2.6.0
Requires-Dist: pyyaml<7,>=6.0.1
Requires-Dist: pyperclip<2,>=1.8.2
Provides-Extra: avro
Requires-Dist: fastavro; extra == "avro"
Provides-Extra: protobuf
Requires-Dist: protobuf; extra == "protobuf"
Provides-Extra: all
Requires-Dist: fastavro; extra == "all"
Requires-Dist: protobuf; extra == "all"
Provides-Extra: dev
Requires-Dist: pytest<9,>=8.0.0; extra == "dev"
Requires-Dist: pytest-cov<5,>=4.1.0; extra == "dev"
Requires-Dist: mypy<2,>=1.8.0; extra == "dev"
Requires-Dist: ruff>=0.2.0; extra == "dev"
Requires-Dist: pre-commit<4,>=3.6.0; extra == "dev"
Requires-Dist: types-pyyaml<7,>=6.0.12; extra == "dev"

<p align="center">
  <h1 align="center">Kafka CLI</h1>
  <p align="center">
    A modern, interactive command-line tool for managing Apache Kafka clusters.
    <br />
    <strong>Topics &middot; Consumer Groups &middot; Messages &middot; Schema Registry &middot; Multi-Cluster</strong>
  </p>
  <p align="center">
    <a href="https://pypi.org/project/kafka-util/"><img src="https://img.shields.io/pypi/v/kafka-util?color=blue" alt="PyPI version"></a>
    <a href="https://pypi.org/project/kafka-util/"><img src="https://img.shields.io/pypi/pyversions/kafka-util" alt="Python versions"></a>
    <a href="LICENSE"><img src="https://img.shields.io/pypi/l/kafka-util" alt="License"></a>
  </p>
  <p align="center">
    <a href="#installation">Installation</a> &middot;
    <a href="#quick-start">Quick Start</a> &middot;
    <a href="#commands">Commands</a> &middot;
    <a href="docs/EXAMPLES.md">Examples</a> &middot;
    <a href="CHANGELOG.md">Changelog</a>
  </p>
</p>

---

## Table of Contents

- [Why Kafka CLI?](#why-kafka-cli)
- [Installation](#installation)
  - [From Source](#from-source)
- [Quick Start](#quick-start)
- [Commands](#commands)
  - [`kafka init`](#kafka-init)
  - [`kafka config`](#kafka-config)
  - [`kafka cluster`](#kafka-cluster)
  - [`kafka topic`](#kafka-topic)
  - [`kafka group`](#kafka-group)
  - [`kafka produce`](#kafka-produce)
  - [`kafka consume`](#kafka-consume)
  - [`kafka schema`](#kafka-schema)
  - [`kafka history`](#kafka-history)
  - [Global Flags](#global-flags)
- [Multi-Cluster Workflow](#multi-cluster-workflow)
- [Authentication](#authentication)
- [Configuration](#configuration)
- [Development](#development)
- [Project Structure](#project-structure)
- [Documentation](#documentation)
- [Built With](#built-with)
- [License](#license)

---

## Why Kafka CLI?

- **Multi-cluster management** — organize clusters by organization, switch contexts in one command
- **Interactive & scriptable** — rich terminal UI for humans, JSON output for pipelines
- **Schema Registry built-in** — Avro/JSON Schema/Protobuf with automatic (de)serialization
- **Safe by default** — destructive operations require confirmation, offset resets default to dry-run
- **Zero config to start** — guided `kafka init` wizard gets you connected in seconds

## Installation

**Requires Python 3.10+**

```bash
pip install kafka-util
```

Or with [pipx](https://pipx.pypa.io/) (recommended):

```bash
pipx install kafka-util
```

With optional serialization support:

```bash
pip install "kafka-util[avro]"        # Avro support (fastavro)
pip install "kafka-util[protobuf]"    # Protobuf support
pip install "kafka-util[all]"         # All serialization formats
```

Verify installation:

```bash
kafka --version
```

### From Source

```bash
git clone https://github.com/haonguyen1915/kafka-cli.git
cd kafka-cli
pip install -e ".[all]"
```

## Quick Start

```bash
# 1. Set up your first cluster
kafka init

# 2. Check cluster health
kafka cluster info

# 3. List topics
kafka topic list

# 4. Create a topic
kafka topic create my-topic --partitions 6 --replication 3

# 5. Produce a message
kafka produce my-topic --message "hello world"

# 6. Consume messages (shows the last 100 per partition by default)
kafka consume my-topic
```

## Commands

### `kafka init`

Interactive setup wizard. Creates or adds organizations and clusters to `~/.kafka-cli/config.yaml`.

```bash
kafka init
```

Prompts for organization, cluster name, bootstrap servers, authentication method, and optional Schema Registry URL. Validates the connection before saving.

---

### `kafka config`

Manage configurations and switch between clusters.

```bash
kafka config list                              # List all orgs & clusters
kafka config current                           # Show active configuration
kafka config use                               # Switch cluster (interactive)
kafka config use acme-corp prod                # Switch cluster (direct)
kafka config test                              # Test current connection
kafka config delete acme-corp staging --yes    # Remove a cluster
```

---

### `kafka cluster`

Cluster information and health checks.

```bash
kafka cluster info                             # Cluster metadata & broker details
kafka cluster health                           # Broker connectivity check
kafka cluster brokers                          # List all brokers
```

---

### `kafka topic`

Full topic lifecycle management.

```bash
kafka topic list                               # List all topics
kafka topic list --filter "order*"             # Glob filter
kafka topic create my-topic -p 6 -r 3         # Create with partitions & replicas
kafka topic describe my-topic                  # Partitions, configs, offsets
kafka topic alter my-topic --config retention.ms=604800000
kafka topic delete my-topic                    # Delete (with confirmation)
kafka topic consume-offsets my-topic           # Consumer group lag per topic
```

---

### `kafka group`

Consumer group inspection and management.

```bash
kafka group list                               # List all groups
kafka group list --state stable                # Filter by state
kafka group describe my-group                  # Members, assignments, lag
kafka group delete my-group                    # Delete group
kafka group reset-offsets my-group \
  --topic my-topic --to-earliest --execute     # Reset offsets
```

Offset reset strategies: `--to-earliest`, `--to-latest`, `--to-offset N`, `--shift-by N`. Defaults to **dry-run** — pass `--execute` to apply.

---

### `kafka produce`

Send messages to a topic.

```bash
kafka produce my-topic                                # Interactive stdin mode
kafka produce my-topic -m "hello world"               # Single message
kafka produce my-topic -k "key1" -m '{"data": 1}'    # With key
kafka produce my-topic --file data.jsonl              # Bulk from file
kafka produce my-topic --file data.jsonl \
  --key-field "id" --rate 100                         # Bulk with rate limit
```

Supports custom headers (`--header KEY:VALUE`), partition targeting, and Avro serialization via Schema Registry.

---

### `kafka consume`

Read messages from a topic.

By default it shows the **last 100 messages per partition**, grouped by partition,
then exits — handy for inspecting a topic (e.g. a DLQ) that has data but no live traffic.

```bash
kafka consume                                         # Pick topic interactively
kafka consume --interactive                           # Prompt for every option
kafka consume my-topic                                # Last 100/partition (default)
kafka consume my-topic --tail 20                      # Last 20/partition
kafka consume my-topic --from-beginning               # Every message from the start
kafka consume my-topic --limit 50                     # Cap total messages at 50
kafka consume my-topic --latest                       # Only new messages from now on
kafka consume my-topic --follow                       # Continuous (like tail -f)
kafka consume my-topic --group my-reader              # With consumer group
kafka consume my-topic --output json                  # JSON output for piping
kafka consume my-topic --output raw                   # Values only
kafka consume my-topic --dump out.json                # Write to a file (JSON array)
kafka consume my-topic --use-avro                     # Auto-deserialize Avro
```

| Flag | Short | Description |
| --- | --- | --- |
| `--interactive` | `-i` | Prompt for every option interactively |
| `--tail` | `-n` | Messages to read back per partition (default 100, the default mode) |
| `--from-beginning` | `-b` | Read every message from the earliest offset |
| `--latest` | | Only read new messages produced from now on |
| `--limit` | `-l` | Hard cap on total messages (any mode) |
| `--follow` | `-f` | Keep consuming, like `tail -f` |
| `--dump` | `-d` | Write consumed messages to a file as a JSON array |
| `--output` | `-O` | Output format: `table` (default), `json`, `raw` |
| `--group` | `-g` | Consume using a specific consumer group ID |
| `--use-avro` | | Deserialize values as Avro via Schema Registry |

Notes:

- `--from-beginning` and `--latest` are mutually exclusive.
- Grouped-by-partition display applies to the default tail mode; `--follow`,
  `--from-beginning`, and `--latest` stream messages as they arrive.

---

### `kafka schema`

Confluent Schema Registry management.

```bash
kafka schema list                                     # List all subjects
kafka schema get my-topic-value                       # Latest schema
kafka schema get my-topic-value --version 3           # Specific version
kafka schema versions my-topic-value                  # List all versions
kafka schema create my-topic-value --file schema.avsc # Register new schema
kafka schema test my-topic-value --file schema.avsc   # Compatibility check
kafka schema delete my-topic-value                    # Soft delete
kafka schema config --level BACKWARD                  # Set compatibility level
```

---

### `kafka history`

Built-in command history.

```bash
kafka history                  # Show recent commands (default: 20)
kafka history -n 50            # Last 50 commands
kafka history open             # Open in editor (vim, nano, code)
kafka history clear            # Clear history
```

---

### Global Flags

All cluster commands support context overrides:

```bash
kafka topic list --org acme-corp --cluster prod
kafka group describe my-group -o acme-corp -c staging
```

## Multi-Cluster Workflow

```bash
# Set up dev cluster
kafka init
# -> Org: acme-corp, Cluster: dev, Bootstrap: localhost:9092

# Add prod cluster to the same org
kafka init
# -> Org: acme-corp, Cluster: prod, Bootstrap: kafka-prod:9092

# Switch between clusters
kafka config use acme-corp dev
kafka config use acme-corp prod

# One-off command against a different cluster
kafka topic list --org acme-corp --cluster prod
```

## Authentication

| Method | Flag |
|--------|------|
| No auth (PLAINTEXT) | `mechanism: none` |
| SASL/PLAIN | `mechanism: sasl_plain` |
| SASL/SCRAM-SHA-256 | `mechanism: sasl_scram_256` |
| SASL/SCRAM-SHA-512 | `mechanism: sasl_scram_512` |
| SSL / mTLS | `mechanism: ssl` |

Security protocols: `PLAINTEXT`, `SASL_PLAINTEXT`, `SASL_SSL`, `SSL`

SSL options include CA certificate, client certificate, client key, and key password.

Schema Registry supports basic auth and token-based authentication.

See [Authentication Guide](docs/AUTHENTICATION.md) for detailed configuration.

## Configuration

Config file location: `~/.kafka-cli/config.yaml`

```yaml
organizations:
  acme-corp:
    name: Acme Corporation
    clusters:
      dev:
        bootstrap_servers: localhost:9092
        auth:
          mechanism: none
        schema_registry:
          url: http://localhost:8081
      prod:
        bootstrap_servers: kafka-prod-1:9092,kafka-prod-2:9092
        auth:
          mechanism: sasl_scram_256
          username: ${KAFKA_USER}
          password: ${KAFKA_PASSWORD}
          security_protocol: SASL_SSL
default:
  organization: acme-corp
  cluster: dev
```

See [Configuration Guide](docs/CONFIGURATION.md) for the full schema reference.

## Development

```bash
# Clone and install
git clone https://github.com/haonguyen1915/kafka-cli.git
cd kafka-cli
poetry install          # or: pip install -e ".[dev,all]"

# Run tests
make test               # or: pytest

# Run tests with coverage
make test-cov           # or: pytest --cov=src --cov-report=html

# Lint & format
make lint               # ruff check + mypy
make format             # ruff fix + format

# Build & publish
make build
make publish
```

## Project Structure

```
src/                       # → maps to kafka_cli package
  __init__.py              # Package metadata & version
  __main__.py              # python -m kafka_cli support
  main.py                  # Typer app entry point & command registration
  commands/
    init.py                # kafka init — setup wizard
    config.py              # kafka config — configuration management
    cluster.py             # kafka cluster — cluster operations
    topic.py               # kafka topic — topic management
    group.py               # kafka group — consumer group management
    produce.py             # kafka produce — message production
    consume.py             # kafka consume — message consumption
    schema.py              # kafka schema — Schema Registry operations
    history.py             # kafka history — command history
  core/
    client.py              # KafkaClientFactory — connection management
    admin_client.py        # Admin operations (topics, groups, ACLs)
    producer.py            # Producer wrapper
    consumer.py            # Consumer wrapper
    config.py              # Config load/save (~/.kafka-cli/)
    context.py             # Current org/cluster context
    schema_registry.py     # Schema Registry HTTP client
    history.py             # Command history tracking
  models/                  # Pydantic data models
  serializers/             # String, JSON, Avro, Protobuf (de)serializers
  ui/                      # Rich console, interactive prompts, table formatting
  utils/                   # Validators, formatters, clipboard helpers
```

## Documentation

| Guide | Description |
|-------|-------------|
| [Command Reference](docs/COMMAND_REFERENCE.md) | All commands, flags, and output examples |
| [Configuration](docs/CONFIGURATION.md) | Config schema and multi-cluster setup |
| [Authentication](docs/AUTHENTICATION.md) | SASL, SSL, mTLS, Confluent Cloud |
| [Schema Registry](docs/SCHEMA_REGISTRY.md) | Avro, JSON Schema, Protobuf workflows |
| [Examples](docs/EXAMPLES.md) | Real-world usage examples |
| [Project Structure](docs/PROJECT_STRUCTURE.md) | Architecture and module details |
| [Changelog](CHANGELOG.md) | Version history and release notes |

## Built With

- [Typer](https://typer.tiangolo.com/) — CLI framework
- [Rich](https://rich.readthedocs.io/) — Terminal formatting and tables
- [Questionary](https://questionary.readthedocs.io/) — Interactive prompts
- [confluent-kafka](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/) — Kafka client
- [Pydantic](https://docs.pydantic.dev/) — Data validation
- [httpx](https://www.python-httpx.org/) — HTTP client (Schema Registry)

## License

This project is licensed under the [GPL-3.0-or-later](LICENSE) license.
