Kafka CLI Tool

🚀🚀🚀 A comprehensive Kafka command line tool to query offsets, consume/produce messages, and manage Kafka clusters.

Features

This Go-based Kafka CLI tool supports all major Kafka client operations:

Core Operations

• Consume messages from topics with flexible partition and offset control
• Produce messages to topics with interactive mode and file input support
• Query offsets for topic partitions (oldest/newest)
• Retrieve metadata about clusters, topics, and brokers
• List topics with detailed partition and offset information
• Show compatibility information for client features

Advanced Features

• ✅ Interactive message production - Type messages in real-time
• ✅ Partition-specific operations - Target specific partitions
• ✅ Offset positioning - Start from beginning, latest, or specific offset
• ✅ JSON output format - Machine-readable output for all commands
• ✅ Consumer group support - Use consumer groups or manual assignment
• ✅ Message headers - Add custom headers to produced messages
• ✅ Bulk operations - Read messages from files for production
• ✅ Bootstrap server configuration - Connect to any Kafka cluster
• ✅ Client configuration - Set client ID and consumer group
• ✅ Producer configuration - Set acks, compression, batch size, linger time
• ✅ Consumer configuration - Set isolation level, fetch size
• ✅ Key/value separation - Parse keys from message content
• ✅ Stdin input support - Read messages from standard input
• ✅ File output - Save consumed messages to files
• ✅ SSL/SASL authentication - Secure connections to Kafka clusters
• ✅ Detailed metadata - Enhanced broker and topic information
• ✅ API version queries - Check broker supported features
• ✅ Flexible message formatting - Control key, timestamp, header display
• ✅ Bootstrap server management - Save and switch between clusters easily
• ✅ Cluster comments - Add descriptions to remember cluster purposes

Installation

Nightly Builds (Recommended)

Download pre-built binaries from our automated nightly builds:

1. Go to Releases

2. Download the latest Nightly Build for your platform:

   • Linux AMD64: kafka-*-linux-amd64.tar.gz
   • Linux ARM64: kafka-*-linux-arm64.tar.gz
   • macOS AMD64: kafka-*-darwin-amd64.tar.gz
   • macOS ARM64 (Apple Silicon): kafka-*-darwin-arm64.tar.gz
   • Windows AMD64: kafka-*-windows-amd64.exe.zip

3. Extract and install:

   # Linux/macOS
   tar -xzf kafka-*-linux-amd64.tar.gz
   sudo mv kafka /usr/local/bin/
   
   # Windows (PowerShell)
   Expand-Archive kafka-*-windows-amd64.exe.zip
   # Add kafka.exe to your PATH

Docker

# Pull latest nightly image
docker pull ghcr.io/worthies/kafka:nightly

# Run directly
docker run --rm ghcr.io/worthies/kafka:nightly --help

# Interactive usage with config
docker run --rm -v ~/.kafka.yaml:/home/kafka/.kafka.yaml \
  ghcr.io/worthies/kafka:nightly bootstrap list

Build from Source

git clone https://github.com/worthies/kafka.git
cd kafka
go build -o kafka .

Install with Go

go install github.com/worthies/kafka@latest

Usage

Global Configuration

All commands support these global flags:

kafka [command] \
  --bootstrap-servers localhost:9092,broker2:9092 \
  --client-id my-client \
  --consumer-group my-group \
  --verbose

Version Information

Check your installation and get detailed version information:

# Show detailed version info
kafka version

# Output:
# Kafka CLI Tool
# Version:    2024.01.15-nightly.abc1234
# Build Time: 2024-01-15T02:30:45Z
# Git Commit: abc1234567890abcdef1234567890abcdef123456
# Go Version: go1.21.0
# OS/Arch:    linux/amd64

# Quick version check
kafka --version

Commands

1. Consume Messages

# Consume from all partitions using consumer group
kafka consume --topic my-topic

# Consume from specific partition starting from beginning
kafka consume --topic my-topic --partition 0 --offset oldest

# Consume with JSON output and message limit
kafka consume --topic my-topic --json --limit 100

# Consume from latest with specific consumer group
kafka consume --topic my-topic --consumer-group my-app-group

# Consume with custom formatting (no timestamps, keys only)
kafka consume --topic my-topic \
  --print-timestamps=false \
  --print-headers=false \
  --output /tmp/messages.log

# Consume with read-committed isolation
kafka consume --topic my-topic \
  --isolation read_committed \
  --fetch-size 1048576

2. Produce Messages

# Interactive mode - type messages in real-time
kafka produce --topic my-topic --interactive

# Single message
kafka produce --topic my-topic --value "Hello, Kafka!"

# Message with key and headers
kafka produce --topic my-topic \
  --key user123 \
  --value "User logged in" \
  --header "event-type=login" \
  --header "timestamp=2024-01-01T12:00:00Z"

# Produce to specific partition with compression
kafka produce --topic my-topic --partition 2 \
  --value "Partition-specific message" \
  --compression gzip \
  --acks 1

# Read messages from file with key/value separator
kafka produce --topic my-topic --file messages.txt --separator ":"

# Read from stdin with custom producer settings
echo "key1:value1" | kafka produce --topic my-topic \
  --stdin \
  --separator ":" \
  --batch-size 100 \
  --linger-ms 50

3. Query Offsets

# Get offsets for all partitions
kafka offsets --topic my-topic

# Get offsets for specific partition
kafka offsets --topic my-topic --partition 0

# JSON output
kafka offsets --topic my-topic --json

4. Cluster Metadata

# Full cluster metadata
kafka metadata

# Only brokers information
kafka metadata --brokers

# Only topics information  
kafka metadata --topics

# Detailed metadata with partition leaders
kafka metadata --topics --detailed

# Broker API versions
kafka metadata --api-versions

# JSON output
kafka metadata --json

5. List Topics

# Simple topic list
kafka topics

# Detailed view with partition and offset information
kafka topics --detail

# JSON output
kafka topics --json

6. Bootstrap Server Management

# Add bootstrap servers with memorable names and comments
kafka bootstrap add local localhost:9092 --comment "Local development cluster"
kafka bootstrap add prod broker1:9092,broker2:9092,broker3:9092 --comment "Production cluster"

# List all saved configurations
kafka bootstrap list

# Switch between configurations
kafka bootstrap use prod
kafka bootstrap use local

# Show current configuration
kafka bootstrap current

# Remove a configuration
kafka bootstrap remove staging

7. Compatibility Information

# Show supported Kafka versions and features
kafka compatibility

# JSON output
kafka compatibility --json

Configuration

Configuration File

Create ~/.kafka.yaml for default settings:

bootstrap-servers:
  - localhost:9092
  - broker2:9092
client-id: my-kafka-client
consumer-group: my-default-group
verbose: true

Environment Variables

Set environment variables with KAFKA_ prefix:

export KAFKA_BOOTSTRAP_SERVERS=localhost:9092
export KAFKA_CLIENT_ID=my-client
export KAFKA_CONSUMER_GROUP=my-group
export KAFKA_SECURITY_PROTOCOL=SASL_SSL
export KAFKA_SASL_MECHANISM=SCRAM-SHA-256
export KAFKA_SASL_USERNAME=my-username
export KAFKA_SASL_PASSWORD=my-password

### Bootstrap Server Management

The CLI tool includes a built-in configuration manager for bootstrap servers, making it easy to work with multiple Kafka clusters:

```bash
# Add clusters with descriptive comments
kafka bootstrap add local localhost:9092 \
  --comment "Local development cluster"

kafka bootstrap add staging staging.company.com:9092 \
  --comment "Staging environment for testing"

kafka bootstrap add prod kafka1.prod.com:9092,kafka2.prod.com:9092,kafka3.prod.com:9092 \
  --comment "Production cluster - handle with care"

# List all saved configurations
kafka bootstrap list

# Switch between clusters
kafka bootstrap use prod

# All commands will now use the production cluster
kafka topics
kafka metadata

The configuration is saved in ~/.kafka.yaml and automatically used by all commands unless you explicitly specify --bootstrap-servers.

Security & Authentication

SSL/TLS Configuration

# Connect with SSL
kafka consume --topic secure-topic 
  --security-protocol SSL 
  --ssl-cert-file /path/to/client.crt 
  --ssl-key-file /path/to/client.key 
  --ssl-ca-file /path/to/ca.crt

SASL Authentication

# SASL/SCRAM-SHA-256
kafka produce --topic secure-topic 
  --security-protocol SASL_SSL 
  --sasl-mechanism SCRAM-SHA-256 
  --sasl-username myuser 
  --sasl-password mypassword

# SASL/PLAIN
kafka metadata 
  --security-protocol SASL_PLAINTEXT 
  --sasl-mechanism PLAIN 
  --sasl-username myuser 
  --sasl-password mypassword

Examples

## Examples

### Real-time Message Monitoring

```bash
# Monitor messages in real-time with JSON output
kafka consume --topic user-events --json --offset latest

Data Pipeline Testing

# Produce test data
echo -e "event1\nevent2\nevent3" | kafka produce --topic test-topic --file -

# Verify consumption
kafka consume --topic test-topic --offset oldest --limit 3

Cluster Health Check

# Check cluster status
kafka metadata --json | jq '.brokerCount'

# List all topics
kafka topics --json | jq '.topics[].name'

# Check topic lag
kafka offsets --topic important-topic --json | jq '.totalMessages'

Supported Kafka Versions

• ✅ Kafka 0.8.2+ through 3.8.x
• ✅ All major Kafka distributions (Apache Kafka, Confluent, MSK, etc.)
• ✅ SASL/SSL authentication support
• ✅ Consumer groups and manual partition assignment

Architecture

Built with:

• IBM Sarama - Pure Go Kafka client
• Cobra - CLI framework
• Viper - Configuration management

Contributing

1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Ensure all tests pass: go test ./...
5. Submit a pull request

License

MIT License - see LICENSE file for details.