Managing Topics

1. Creating Topics

Example: Create a topic with default configuration

kafka-topics.sh --create --topic my-topic --bootstrap-server localhost:9092

Option	Values	Description
--topic	`my-topic`	Name of the topic to create
--bootstrap-server	`localhost:9092`	Bootstrap server address to connect to the Kafka cluster
--partitions	`1`	Number of partitions for the topic (default: 1)
--replication-factor	`1`	Replication factor for the topic (default: 1)
--config	`retention.ms=86400000`	Configuration properties for the topic (e.g., retention time)

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

Warning: Creating a topic with a replication factor higher than the number of brokers will result in an error.

2. Listing Topics

Command	Description	Options	Example
kafka-topics.sh --list	List all topics in the Kafka cluster	`--bootstrap-server` `--command-config`	`kafka-topics.sh --list --bootstrap-server localhost:9092`
kafka-topics.sh --describe	Describe topic details including partitions, replicas, and configurations	`--topic` `--bootstrap-server` `--under-replicated-partitions`	`kafka-topics.sh --describe --topic my-topic --bootstrap-server localhost:9092`

3. Describing Topics

Property	Value	Description
topic.name	`orders`	Name of the Kafka topic
partitions	`6`	Number of partitions in the topic
replication.factor	`3`	Number of replicas for each partition
retention.ms	`604800000`	Retention period in milliseconds (7 days)
cleanup.policy	`delete`	Cleanup policy for log compaction (delete or compact)

Example: Describe a Kafka Topic

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders

Note: The --describe command provides detailed metadata about the topic including partition distribution, leader information, and replica status.

4. Setting Partition Count

Partition count determines how many partitions a topic can have, which affects parallelism and scalability.

Property	Values	Description
num.partitions	`1`	Default number of partitions when creating a topic
partitions	`1`	Number of partitions for a topic (can be changed later)

Example: Creating a topic with 4 partitions

kafka-topics.sh --create --topic my-topic --partitions 4 --replication-factor 1

Example: Changing partition count for an existing topic

kafka-topics.sh --alter --topic my-topic --partitions 8

Note: Increasing partition count is allowed, but decreasing it is not supported.

Warning: Changing partition count affects consumer group rebalancing and may cause data loss if not handled carefully.

5. Setting Replication Factor

Property	Default Value	Description
replication.factor	`1`	Number of replicas for each partition

Example: Setting Replication Factor

kafka-topics.sh --create --topic my-topic --partitions 3 --replication-factor 3 --bootstrap-server localhost:9092

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

6. Altering Topic Configuration

Property	Values	Description
retention.ms	`-1` or `> 0`	Time to retain messages in topic (in milliseconds). `-1` means infinite retention.
retention.bytes	`-1` or `> 0`	Maximum size of topic logs (in bytes). `-1` means unlimited.
segment.bytes	`> 0`	Maximum size of a single log segment (in bytes).
segment.ms	`> 0`	Time after which a new log segment is created (in milliseconds).
compression.type	`none`, `gzip`, `snappy`, `lz4`, `zstd`	Compression type for log segments.
min.insync.replicas	`> 0`	Minimum number of replicas that must acknowledge a write for it to be considered successful.
max.message.bytes	`> 0`	Maximum size of a single message (in bytes).

Example: Altering topic configuration using CLI

kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --alter --add-config retention.ms=604800000

Note: Changes to topic configuration are asynchronous and may take some time to propagate across all brokers in the cluster.

7. Deleting Topics

Property	Value	Description
topic.name	`my-topic`	Name of the topic to delete
retention.ms	`604800000`	Retention period in milliseconds (default 7 days)

Example: Delete Topic via CLI

kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic my-topic

Note: Deletion is asynchronous and may take some time to complete. The topic will be marked for deletion but not immediately removed from all brokers.

Warning: Deleting a topic will permanently remove all data and configuration associated with it. Ensure that no consumers or producers are actively using the topic before deletion.

8. Configuring Topic Retention

Property	Default Value	Description
retention.ms	`-1`	Time to retain log segments before deletion. -1 means infinite retention.
retention.bytes	`-1`	Maximum size of the log before deletion. -1 means unlimited.
log.flush.interval.messages	`10000`	Number of messages to accumulate before flushing the log.
log.flush.interval.ms	`900000`	Time in milliseconds to wait before flushing the log.

Example: Setting retention policy for a topic

kafka-topics.sh --create --topic my-topic \
  --partitions 3 \
  --replication-factor 1 \
  --config retention.ms=604800000 \
  --config retention.bytes=-1

Note: Setting retention.ms to 0 will effectively delete all messages immediately after they are written.

9. Setting Cleanup Policy

Policy	Description	Default	Notes
delete	Messages are deleted after a configured retention period or size limit is reached.	delete	Most common policy for log compaction.
compact	Messages are retained for as long as the latest value for a key is present.	compact	Used for maintaining the latest state of keys in a topic.
compact, delete	Applies both compaction and deletion policies to the topic.	compact, delete	Allows for both key-based retention and size/age limits.

Example: Configuring Cleanup Policy for a Topic

kafka-topics.sh --create --topic my-topic \
  --partitions 3 \
  --replication-factor 1 \
  --config retention.ms=604800000 \
  --config cleanup.policy=compact,delete

Note: The cleanup.policy setting can be configured at the topic level or inherited from the broker configuration.

Warning: When using compact, ensure that the topic has a reasonable retention period to prevent indefinite storage of keys.

10. Configuring Segment Settings

Property	Default Value	Description
log.segment.bytes	`1073741824`	Maximum size of a single log segment file in bytes. NEW in Kafka 3.0
log.segment.ms	`604800000`	Maximum age of a log segment in milliseconds before it is eligible for deletion.
log.roll.hours	`168`	Deprecated in favor of `log.segment.ms`. Controls how often segments are rolled based on time.
log.retention.hours	`168`	Retention period for logs in hours. DEPRECATED since 2.0, replaced by `log.retention.ms`.
log.retention.ms	`604800000`	Retention period for logs in milliseconds. Default is 7 days.

Example: Setting Segment Size and Retention

kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name my-topic \
  --set log.segment.bytes=268435456 \
  --set log.retention.ms=2592000000

Note: Segment settings can be configured at the topic level or inherited from broker defaults. Adjusting log.segment.bytes affects how frequently log segments are rolled and can impact performance if set too low.