kafka-util 0.1.4

CLI tool for managing Apache Kafka clusters

Kafka CLI

A modern, interactive command-line tool for managing Apache Kafka clusters.
Topics · Consumer Groups · Messages · Schema Registry · Multi-Cluster

Installation · Quick Start · Commands · Examples · Changelog

---

Table of Contents

• Why Kafka CLI?
• Installation
  • From Source
• Quick Start
• Commands
  • kafka init
  • kafka config
  • kafka cluster
  • kafka topic
  • kafka group
  • kafka produce
  • kafka consume
  • kafka schema
  • kafka history
  • Global Flags
• Multi-Cluster Workflow
• Authentication
• Configuration
• Development
• Project Structure
• Documentation
• Built With
• 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+

pip install kafka-util

Or with pipx (recommended):

pipx install kafka-util

With optional serialization support:

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:

kafka --version

From Source

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

Quick Start

# 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.

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.

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.

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.

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.

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.

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.

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.

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.

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:

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

Multi-Cluster Workflow

# 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 for detailed configuration.

Configuration

Config file location: ~/.kafka-cli/config.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 for the full schema reference.

Development

# 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	All commands, flags, and output examples
Configuration	Config schema and multi-cluster setup
Authentication	SASL, SSL, mTLS, Confluent Cloud
Schema Registry	Avro, JSON Schema, Protobuf workflows
Examples	Real-world usage examples
Project Structure	Architecture and module details
Changelog	Version history and release notes

Built With

• Typer — CLI framework
• Rich — Terminal formatting and tables
• Questionary — Interactive prompts
• confluent-kafka — Kafka client
• Pydantic — Data validation
• httpx — HTTP client (Schema Registry)

License

This project is licensed under the GPL-3.0-or-later license.