kafka-replay-cli

A lightweight, CLI tool for dumping and replaying Kafka messages using Parquet files. Built for observability, debugging, and safe testing of event streams.

Use it to:

• Safely test event streams without impacting production.
• Debug complex Kafka flows by replaying exact message sets.
• Apply filters, transformations, and throttling for precise control.

---

Features

• Dump Kafka topics into Parquet files
• Replay messages from Parquet back into Kafka
• Filter replays by timestamp range and key
• Optional throttling during replay
• Apply custom transform hooks to modify or skip messages
• Preview replays without sending messages using --dry-run
• Control output verbosity with --verbose and --quiet
• Query message dumps with DuckDB SQL

---

Installation

pip install kafka-replay-cli

Requires Python 3.8 or newer.

---

Kafka Broker Requirements

You must have access to a running Kafka broker.

By default, the CLI will attempt to connect to localhost:9092, but you can specify any broker using the --bootstrap-servers option:

--bootstrap-servers my.kafka.broker:9092

---

Usage

Dump messages from a topic to Parquet

kafka-replay-cli dump \
  --topic test-topic \
  --output test.parquet \
  --bootstrap-servers localhost:9092 \
  --max-messages 1000

Advanced Dump Settings

--fetch-max-bytes 1000000   # Max bytes to fetch from broker per poll

---

Replay messages from a Parquet file

kafka-replay-cli replay \
  --input test.parquet \
  --topic replayed-topic \
  --bootstrap-servers localhost:9092 \
  --throttle-ms 100 

Advanced Producer Settings

You can fine-tune how the replay produces messages to Kafka:

--acks all                 # Wait for all replicas to acknowledge (0, 1, or all)
--compression-type gzip    # Compress messages (gzip, snappy, lz4, zstd)
--linger-ms 10             # Delay to batch more messages (in milliseconds)
--producer-batch-size 5000 # Max batch size in bytes for Kafka producer

---

Preview messages without sending (--dry-run)

kafka-replay-cli replay \
  --input test.parquet \
  --topic replayed-topic \
  --dry-run

By default, shows up to 5 sample messages that would be replayed. Use --dry-run-limit to adjust the number of preview messages.

--dry-run-limit 10

Adjust verbosity

--verbose  # Show detailed logs, including skipped messages
--quiet    # Suppress all output except errors and final summary

Example:

kafka-replay-cli replay \
  --input test.parquet \
  --topic replayed-topic \
  --dry-run \
  --verbose

---

Add timestamp and key filters

kafka-replay-cli replay \
  --input test.parquet \
  --topic replayed-topic \
  --start-ts "2024-01-01T00:00:00Z" \
  --end-ts "2024-01-02T00:00:00Z" \
  --key-filter "user-123"

---

Add partition and offset filters

--partition 1            # Only replay messages from partition 1
--offset-start 1000      # Replay offsets >= 1000
--offset-end 2000        # Replay offsets <= 2000

Example:

kafka-replay-cli replay \
  --input test.parquet \
  --topic replayed-topic \
  --partition 0 \
  --offset-start 100 \
  --offset-end 200

---

Control batch size

--batch-size 500

Controls how many messages are processed per batch during replay.

---

Transform Messages Before Replay

You can modify, enrich, or skip Kafka messages during replay by passing a custom Python script that defines a transform(msg) function.

Basic Example

File: hooks/example_transform.py

def transform(msg):
    if msg["value"]:
        msg["value"] = msg["value"].upper()
    return msg

Run with:

kafka-replay-cli replay \
  --input messages.parquet \
  --topic replayed-topic \
  --transform-script hooks/example_transform.py

Skip Messages

If your function returns None, the message will be skipped.
Use --verbose to see skip notices.

def transform(msg):
    if b'"event":"login"' not in msg["value"]:
        return None
    return msg

Message Format

Each msg is a dictionary:

{
  "timestamp": datetime,
  "key": bytes,
  "value": bytes,
  "partition": int,
  "offset": int
}

You can modify key and value, or add additional fields.

---

Query Messages with DuckDB

Run SQL queries directly on dumped Parquet files:

kafka-replay-cli query \
  --input test.parquet \
  --sql "SELECT timestamp, CAST(key AS VARCHAR) FROM input WHERE CAST(value AS VARCHAR) LIKE '%login%'"

Note: Kafka key and value are stored as binary (BLOB) for fidelity.
To search or filter them, use CAST(... AS VARCHAR).

---

Output query results to file

kafka-replay-cli query \
  --input test.parquet \
  --sql "SELECT key FROM input" \
  --output results.json

---

License

MIT

This project is not affiliated with or endorsed by the Apache Kafka project.

---

Maintainer

Konstantinas Mamonas

Feel free to fork, open issues, or suggest improvements.

---

Version

Use:

kafka-replay-cli version

to check the installed version.

---

Roadmap

See the ROADMAP for upcoming features and plans.