Kafka Console CLI

A command-line interface for Apache Kafka operations.

Installation

npm install -g kafka-console

Or run without installing:

npx kafka-console [command]

Quick Start

# List topics
kafka-console list -b localhost:9092

# Consume messages
kafka-console consume my-topic -b localhost:9092 --from 0 --count 10

# Produce a message
echo '{"key":"k1","value":"hello"}' | kafka-console produce my-topic -b localhost:9092

Global Options

These options apply to all commands:

| Option | Description | Default | Env | |--------|-------------|---------|-----| | -b, --brokers <brokers> | Comma-separated broker addresses | localhost:9092 | KAFKA_BROKERS | | -t, --timeout <ms> | Operation timeout in milliseconds | 0 (no timeout) | KAFKA_TIMEOUT | | --ssl | Enable TLS connection | false | | | --insecure | Disable TLS certificate verification (requires --ssl) | false | | | --mechanism <mech> | SASL mechanism: plain, scram-sha-256, scram-sha-512, oauthbearer | | KAFKA_MECHANISM | | --username <user> | SASL username (for plain/scram) | | KAFKA_USERNAME | | --password <pass> | SASL password (for plain/scram) | | KAFKA_PASSWORD | | --oauth-bearer <token> | SASL OAuth bearer token | | KAFKA_OAUTH_BEARER |

Commands

consume

Consume messages from a Kafka topic. Outputs one JSON object per line (JSONL).

kafka-console consume <topic> [options]

Each output line contains: partition, offset, timestamp, headers, key, value, ahead.

| Option | Description | Default | |--------|-------------|---------| | -g, --group <group> | Consumer group name | auto-generated | | -f, --from <from> | Start position: 0 (beginning), timestamp in ms, or ISO 8601 date | latest | | -c, --count <count> | Maximum number of messages to consume | unlimited | | -s, --skip <skip> | Number of messages to skip before outputting | 0 | | -o, --output <file> | Write output to a file instead of stdout | stdout | | -d, --data-format <fmt> | Value format: json, raw, or path to custom formatter | json | | --snapshot | Consume all existing messages and exit (records high watermark on start) | false |

Examples:

# Consume from the beginning, stop after 10 messages
kafka-console consume my-topic --from 0 --count 10

# Snapshot: dump all existing messages to a file and exit
kafka-console consume my-topic --snapshot -o dump.jsonl

# Consume from a specific timestamp
kafka-console consume my-topic --from "2024-06-01T00:00:00Z"

# Skip first 100 messages, take next 50
kafka-console consume my-topic --from 0 --skip 100 --count 50

# Pipe to jq for filtering
kafka-console consume my-topic | jq 'select(.value.level == "ERROR")'

# Use raw format for non-JSON message values
kafka-console consume my-topic -d raw

produce

Produce messages to a Kafka topic. Reads JSONL from stdin or a JSON array from a file.

kafka-console produce <topic> [options]

Each input line must be a JSON object with a value field. Optional fields: key, headers.

| Option | Description | Default | |--------|-------------|---------| | -d, --data-format <fmt> | Value format: json, raw, or path to custom formatter | json | | -i, --input <file> | Read messages from a JSON array file instead of stdin | stdin | | -w, --wait <ms> | Delay between messages in milliseconds | 0 | | -H, --header <header> | Static header added to every message (key:value), repeatable | none |

Examples:

# Produce a single message
echo '{"key":"user-1","value":{"name":"Alice"}}' | kafka-console produce my-topic

# Produce with static headers
echo '{"value":"data"}' | kafka-console produce my-topic -H source:cli -H env:prod

# Produce from a file (JSON array of messages)
kafka-console produce my-topic --input messages.json

# Produce raw text values
echo '{"value":"plain text"}' | kafka-console produce my-topic -d raw

# Pipe from another topic
kafka-console consume source --snapshot | kafka-console produce dest

list

List topic names. Outputs one topic name per line (JSONL).

kafka-console list [options]
kafka-console ls [options]

| Option | Description | Default | |--------|-------------|---------| | -a, --all | Include internal topics (e.g. __consumer_offsets) | false |

metadata

Display cluster metadata: brokers, controller, topics, partitions, replicas, and ISR.

kafka-console metadata

Output is a single JSONL object with brokers, clusterId, controllerId, and topicMetadata fields.

config

Describe the configuration of a Kafka resource.

kafka-console config [options]

| Option | Description | |--------|-------------| | -r, --resource <type> | Resource type: topic, broker, broker_logger (required) | | -n, --resourceName <name> | Resource name: topic name or broker ID (required) |

Examples:

kafka-console config -r topic -n my-topic
kafka-console config -r broker -n 1

topic:create

Create a new Kafka topic.

kafka-console topic:create <topic> [options]

| Option | Description | Default | |--------|-------------|---------| | -p, --partitions <n> | Number of partitions | 1 | | -r, --replicas <n> | Replication factor | 1 |

topic:delete

Delete a Kafka topic.

kafka-console topic:delete <topic>

topic:offsets

Show topic partition offsets.

kafka-console topic:offsets <topic> [timestamp] [options]

Without arguments: shows high/low watermarks per partition. With a timestamp (ms or ISO 8601): shows offsets at that point in time. With --group: shows committed offsets for a consumer group.

| Option | Description | |--------|-------------| | -g, --group <group> | Show committed offsets for this consumer group |

Examples:

# High/low watermarks
kafka-console topic:offsets my-topic

# Offsets at a specific time
kafka-console topic:offsets my-topic "2024-06-01T00:00:00Z"

# Consumer group committed offsets
kafka-console topic:offsets my-topic -g my-consumer-group

Authentication

# TLS (certificate verification enabled by default)
kafka-console consume my-topic -b broker:9093 --ssl

# TLS with self-signed certificate (skip verification)
kafka-console consume my-topic -b broker:9093 --ssl --insecure

# SASL/PLAIN
kafka-console consume my-topic -b broker:9093 --ssl \
  --mechanism plain --username myuser --password mypass

# SASL/SCRAM-SHA-256
kafka-console consume my-topic -b broker:9093 --ssl \
  --mechanism scram-sha-256 --username myuser --password mypass

# OAuth Bearer
kafka-console consume my-topic -b broker:9093 --ssl \
  --mechanism oauthbearer --oauth-bearer "eyJhbG..."

Message Formats

json (default) - values are JSON-serialized on produce and JSON-parsed on consume.

raw - values are passed as plain text strings, no serialization.

Custom formatter - provide a path to a module exporting encode and decode functions:

// formatter/custom.js
module.exports = {
  encode: (value) => Buffer.from(JSON.stringify(value)),
  decode: (buffer) => JSON.parse(buffer.toString())
};

kafka-console consume my-topic -d ./formatter/custom.js

License

License The MIT License