Skip to content

Commands

The following commands serve to create, alter and delete Kafka topics within the LogMan.io environment. All Kafka topics managed by LogMan.io next to the internal ones are specified in event lane *.yaml declarations inside /EventLanes folder in the library.

Prerequisites

All commands should be run from Kafka Docker container, that can be accessed via the following command:

docker exec -it kafka_container bash

The command utilizes Kafka Command Line interface, which is documented here: Kafka Command-Line Interface (CLI) Tools

Create a topic

In order to create a topic, specify the topic name, number of partitions and replication factor. The replication factor should be set to 1 and partitions to 6, which is the default for LogMan.io Kafka topics.

/usr/bin/kafka-topics --zookeeper locahost:2181 --create --topic "events.tenant.fortigate" --replication-factor 1 --partitions 6

Replace events.tenant.fortigate with your topic name.

Configure a topic

Retention

The following command changes the retention of data for Kafka topic to 86400000 milliseconds, that is 1 day. This means that data older than 1 day will be deleted from Kafka to spare storage space:

/usr/bin/kafka-configs --bootstrap-server localhost:9092 --entity-type topics --entity-name "events\.tenant\.fortigate" --alter --add-config retention.ms=86400000
Replace events\.tenant\.fortigate with your topic name.

Info

All Kafka topics in LogMan.io should have a retention for data set.

Info

When editing a topic setting in Kafka, the special characters like dot (.) should be escaped with slash (\).

Reseting a consumer group offset for a given topic

In order to reset the reading position, or the offset, for the given group ID (consumer group), use the following command:

/usr/bin/kafka-consumer-groups --bootstrap-server localhost:9092  --group "my-console-client"  --topic "events\.tenant\.fortigate"  --reset-offsets --to-datetime 2020-12-20T00:00:00.000 --execute

Replace events\.tenant\.fortigate with your topic name.

Replace my-console-client with the given group ID.

Replace 2020-12-20T00:00:00.000 with the time to reset the reading offset to.

Hint

To reset the group to the current offset, use --to-current instead of --to-datetime 2020-12-20T00:00:00.000.

Deleting a consumer group offset for a given topic

The offset for the given topic can be deleted from the consumer group, hence the consumer group would be effectively disconnected from the topic itself. Use the following command:

/usr/bin/kafka-consumer-groups --bootstrap-server localhost:9092  --group "my-console-client"  --topic "events\.tenant\.fortigate" --delete-offsets

Replace events\.tenant\.fortigate with your topic name.

Replace my-console-client with the given group ID.

Deleting the consumer group

A consumer group for ALL topics can be deleted with its offset information using the following command:

/usr/bin/kafka-consumer-groups --bootstrap-server localhost:9092 --delete --group my-console-client

Replace my-console-client with the given group ID.

Alter a topic

Change the number of partitions

The following command increases the number of partitions within the given topic.

/usr/bin/kafka-topics --zookeeper locahost:2181 --alter -partitions 6 --topic "events\.tenant\.fortigate"

Replace events\.tenant\.fortigate with your topic name.

Specify ZooKeeper node

Kafka reads and alters data stored in ZooKeeper. In case you've configured Kafka so its files are stored in specific ZooKeeper node, you will get this error.

Error while executing topic command : Topic 'events.tenant.fortigate' does not exist as expected
[2024-05-06 10:16:36,207] ERROR java.lang.IllegalArgumentException: Topic 'events.tenant.fortigate' does not exist as expected
at kafka.admin.TopicCommand$.kafka$admin$TopicCommand$$ensureTopicExists(TopicCommand.scala:539)
at kafka.admin.TopicCommand$ZookeeperTopicService.alterTopic(TopicCommand.scala:408)
at kafka.admin.TopicCommand$.main(TopicCommand.scala:66)
at kafka.admin.TopicCommand.main(TopicCommand.scala)
(kafka.admin.TopicCommand$)

Adjust the --zookeeper argument accordingly. E.g. Kafka data is stored in kafka node of ZooKeeper:

/usr/bin/kafka-topics --zookeeper lm11:2181/kafka --alter --partitions 6 --topic 'events\.tenant\.fortigate'

Try to remove escape characters (/) if the topic name is still not recognized.

Delete a topic

The topic can be deleted using the following command. Please keep in mind that Kafka topics are automatically created if new data are being produced/sent to it by any service.

/usr/bin/kafka-topics --zookeeper locahost:2181 --delete --topic "events\.tenant\.fortigate"

Replace events\.tenant\.fortigate with your topic name.

Troubleshooting

There are many logs in others and I cannot find the ones with "interface" attribute inside

Kafka Console Consumer can be used to obtain events from multiple topics, here from all topics starting with events..

Next it is possible the grep the field in doublequotes:

/usr/bin/kafka-console-consumer --bootstrap-server localhost:9092 --whitelist "events.*" | grep '"interface"'

This command gives you all incoming logs with "interface" attribute from all events topics.