Using Kafka CLI Tools

1. Managing Topics

Command	Description	Options	Example
kafka-topics.sh --create	Create a new topic	`--topic <topic-name>`, `--partitions <num-partitions>`, `--replication-factor <factor>`	`kafka-topics.sh --create --topic my-topic --partitions 3 --replication-factor 1`
kafka-topics.sh --list	List all topics	`--bootstrap-server <server>`	`kafka-topics.sh --list --bootstrap-server localhost:9092`
kafka-topics.sh --describe	Describe topic details	`--topic <topic-name>`, `--under-replicated-partitions`	`kafka-topics.sh --describe --topic my-topic`
kafka-topics.sh --alter	Alter topic configuration	`--topic <topic-name>`, `--config <key=value>`	`kafka-topics.sh --alter --topic my-topic --config retention.ms=86400000`
kafka-topics.sh --delete	Delete a topic	`--topic <topic-name>`	`kafka-topics.sh --delete --topic my-topic`

Example: Creating a Topic with Custom Configuration

kafka-topics.sh --create \
  --topic user-events \
  --partitions 6 \
  --replication-factor 2 \
  --config retention.ms=604800000 \
  --config max.message.bytes=1048588

Note: The --replication-factor must be less than or equal to the number of brokers in your cluster.

Warning: Deleting a topic is irreversible. Ensure you have backups or are certain about the action.

2. Working with Consumer Groups

Command	Description	Options
kafka-consumer-groups.sh --list	List all consumer groups	`--bootstrap-server` (required), `--verbose`
kafka-consumer-groups.sh --describe	Describe consumer group details	`--bootstrap-server` (required), `--group`, `--members`, `--offsets`
kafka-consumer-groups.sh --delete	Delete a consumer group	`--bootstrap-server` (required), `--group`
kafka-consumer-groups.sh --reset-offsets	Reset consumer group offsets	`--bootstrap-server` (required), `--group`, `--topic`, `--to-earliest`, `--to-latest`, `--by-duration`

Example: List consumer groups

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

Example: Describe consumer group

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group

Example: Reset consumer group offsets to earliest

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group my-group --topic my-topic --to-earliest

Note: The --reset-offsets command requires the --execute flag to actually perform the reset.

3. Managing ACLs

Command	Description	Options	Example
kafka-acls.sh	Manage ACLs for Kafka brokers	`--bootstrap-server`, `--command-config`, `--topic`, `--group`	`kafka-acls.sh --bootstrap-server localhost:9092 --list`

Example: Create ACL for a user to read a topic

kafka-acls.sh --bootstrap-server localhost:9092 \
  --add --allow-principal User:alice \
  --operation Read --topic test-topic

Example: List all ACLs

kafka-acls.sh --bootstrap-server localhost:9092 --list

Example: Delete an ACL

kafka-acls.sh --bootstrap-server localhost:9092 \
  --remove --allow-principal User:alice \
  --operation Read --topic test-topic

Note: ACLs are managed at the broker level and require appropriate permissions to configure.

4. Managing Configs

Config Type	Command	Description
Broker Configs	`kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-name 0 --alter --add-config log.flush.interval.messages=1000`	Alter broker-level configurations
Topic Configs	`kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --alter --add-config retention.ms=604800000`	Alter topic-level configurations
User Configs	`kafka-configs.sh --bootstrap-server localhost:9092 --entity-type users --entity-name user1 --alter --add-config sasl.mechanism=PLAIN`	Alter user-level configurations

Example: View Topic Configs

kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --describe</code>
</pre>
</div>
</div>
<div class="example">
<h2 id="example-set-retention-period-for-a-topic">Example: Set Retention Period for a Topic<a href="#example-set-retention-period-for-a-topic" class="toc__headingLink" aria-label="Link to this section"><svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M10 13a5 5 0 0 0 7.54.54l3-3a5 5 0 0 0-7.07-7.07l-1.72 1.71"></path><path d="M14 11a5 5 0 0 0-7.54-.54l-3 3a5 5 0 0 0 7.07 7.07l1.71-1.71"></path></svg></a></h2>
<div class="codeBlock" data-language="bash">
<pre><code>kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --alter --add-config retention.ms=604800000</code>
</pre>
</div>
</div>
<div class="note">
<strong>Note:</strong> Config changes may take time to propagate across the cluster.
</div>
<h2 id="5-reassigning-partitions">5. Reassigning Partitions<a href="#5-reassigning-partitions" class="toc__headingLink" aria-label="Link to this section"><svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M10 13a5 5 0 0 0 7.54.54l3-3a5 5 0 0 0-7.07-7.07l-1.72 1.71"></path><path d="M14 11a5 5 0 0 0-7.54-.54l-3 3a5 5 0 0 0 7.07 7.07l1.71-1.71"></path></svg></a></h2>
<div class="example">
<h2 id="example-reassigning-partitions-for-a-topic">Example: Reassigning partitions for a topic<a href="#example-reassigning-partitions-for-a-topic" class="toc__headingLink" aria-label="Link to this section"><svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M10 13a5 5 0 0 0 7.54.54l3-3a5 5 0 0 0-7.07-7.07l-1.72 1.71"></path><path d="M14 11a5 5 0 0 0-7.54-.54l-3 3a5 5 0 0 0 7.07 7.07l1.71-1.71"></path></svg></a></h2>
<div class="codeBlock" data-language="bash"><pre><code>kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --topic test-topic --generate --broker-list 0,1,2

Option	Description	Values
--bootstrap-server	Bootstrap server address	`localhost:9092`
--topic	Topic to reassign partitions for	`test-topic`
--generate	Generate reassignment plan	`true`
--broker-list	Comma-separated list of broker IDs to reassign partitions to	`0,1,2`
--execute	Execute the reassignment plan	`true`
--verify	Verify the reassignment status	`true`

Note: The reassignment process is asynchronous and may take some time to complete. Use --verify to check the status.

Example: Executing a reassignment plan

kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file reassignment.json --execute

Example: Verifying a reassignment plan

kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file reassignment.json --verify

6. Preferred Leader Election

Preferred leader election is a Kafka feature that allows you to designate a specific broker as the preferred leader for a partition. This is useful for maintaining consistency and ensuring that a particular broker handles the load for a specific partition.

Property	Description	Default Value	Notes
preferred.replica.election.enable	Enables preferred replica election	false	Set to true to enable preferred leader election

Example: Enable Preferred Replica Election

kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --alter --add-config preferred.replica.election.enable=true

Note: Preferred leader election is a configuration that can be applied at the topic level. It does not automatically elect a preferred leader, but rather enables the election process.

7. Verifying Consumer Group

Command	Description	Options
kafka-consumer-groups.sh	Lists consumer groups and their status	`--bootstrap-server`, `--describe`

Example: Describe consumer groups

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group

Note: The --group option limits output to a specific consumer group.

Field	Description	Notes
GROUP	Name of the consumer group	Unique identifier for the consumer group
CONSUMER_ID	ID of the consumer instance	Unique within the group
CLIENT_ID	Client identifier	Used for logging and monitoring
TOPIC	Topic name	Consumed by this consumer
PARTITION	Partition number	Assigned to this consumer
CURRENT_OFFSET	Current offset in the partition	Offset of the last message consumed
LOG_END_OFFSET	End offset of the partition	Total messages in the topic
LAG	Number of unprocessed messages	Indicates consumer lag

8. Verifying Producer

Command	Description	Example
kafka-console-producer.sh	Starts a console producer to send messages to a Kafka topic	`kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic`
--bootstrap-server	Specifies the Kafka broker address to connect to	`--bootstrap-server localhost:9092`
--topic	Specifies the topic to send messages to	`--topic my-topic`
--message-format	Defines how messages are formatted (e.g., json, text)	`--message-format json`
--async	Enables asynchronous message sending (default is synchronous)	`--async`

Example: Send a message using the console producer

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic
> Hello, Kafka!
> This is a test message.
> ^C

Note: Messages sent via the console producer are printed to stdout when using --async. For synchronous mode, messages are sent and acknowledged before proceeding.

9. Dumping Log Segments

Example: Dump log segments for a topic

kafka-run-class.sh kafka.tools.DumpLogSegments --files /path/to/kafka/logs/my-topic-0/00000000000000000000.log

Option	Values	Description
--files	`<file-path>`	Path to the log segment file(s) to dump
--print-data	`true` \| `false`	Whether to print the actual log data (default: false)
--patch-v0	`true` \| `false`	Apply v0 patch to older log formats (default: false)
--max-message-size	`int`	Maximum message size to read (default: 5MB)

Note: The DumpLogSegments tool is primarily used for debugging and inspecting the internal structure of Kafka log segments. It does not support reading compressed messages directly without decompression.

10. Mirroring Data

Kafka provides the kafka-mirror-maker.sh tool to mirror data between clusters. This is useful for disaster recovery, cross-region replication, or migrating between Kafka versions.

Property	Description	Default Value
source.cluster.bootstrap.servers	Bootstrap servers of the source Kafka cluster	localhost:9092
target.cluster.bootstrap.servers	Bootstrap servers of the target Kafka cluster	localhost:9092
topics	Topics to mirror (comma-separated list or regex pattern)	*
replication.factor	Replication factor for mirrored topics	1

Example: Mirroring Topics

kafka-mirror-maker.sh \
  --consumer.config consumer.properties \
  --producer.config producer.properties \
  --whitelist=".*"

Note: Ensure that the consumer and producer configurations are set up to connect to the respective source and target clusters.

Warning: The mirror maker does not support mirroring topics with different replication factors or partitions than the source.

11. Running Performance Tests

Example: Kafka Test Tools Configuration

kafka-run-class.sh kafka.tools.TestApp \
  --topic test-topic \
  --num-records 10000 \
  --record-size 1024 \
  --throughput 1000

Parameter	Description	Default Value
--topic	Name of the topic to test	None
--num-records	Total number of records to send	1000
--record-size	Size of each record in bytes	1024
--throughput	Records per second to send (0 = unlimited)	0
--producer-props	Properties for the producer client	None

Note: The kafka-run-class.sh script must be run from the Kafka installation directory.

Warning: Running performance tests can consume significant system resources and may impact other services.

Performance Test Summary

Use kafka.tools.TestApp to simulate load on Kafka clusters
Adjust --num-records and --record-size to match desired test intensity
Monitor cluster metrics during tests to identify bottlenecks