topicctl

A tool for easy, declarative management of Kafka topics. Includes the ability to "apply" topic changes from YAML as well as a repl for interactive exploration of brokers, topics, consumer groups, messages, and more.

Motivation

Managing Kafka topics via the standard tooling can be tedious and error-prone; there is no standard, declarative way to define topics (e.g., YAML files that can be checked-in to git), and understanding the state of a cluster at any given point in time requires knowing and using multiple, different commands with different interfaces.

We created topicctl to make the management of our Kafka topics more transparent and user-friendly. The project was inspired by kubectl and other tools that we've used in non-Kafka-related contexts.

See this blog post for more details.

🆕 Upgrading from v0

We recently revamped topicctl to support ZooKeeper-less cluster access as well as some additional security options (TLS/SSL and SASL)! See this blog post to learn more about why and how we did this.

All changes should be backwards compatible, but you'll need to update your cluster configs if you want to take advantage of these new features; see the clusters section below for more details on the latest config format.

The code for the old version has been preserved in the v0 branch if you run into problems and need to revert.

Related projects

Check out the data-digger for a command-line tool that makes it easy to tail and summarize structured data in Kafka.

Getting started

Installation

Either:

Run go install github.com/segmentio/topicctl/cmd/topicctl@latest
Clone this repo and run make install in the repo root
Use the Docker image: docker pull segment/topicctl

If you use (1) or (2), the binary will be placed in $GOPATH/bin.

If you use Docker, you can run the tool via docker run -ti --rm segment/topicctl [subcommand] [flags]. Depending on your Docker setup and what you're trying to do, you may also need to run in the host network via --net=host and/or mount local volumes with -v.

Quick tour

Start up a 6 node Kafka cluster locally:

docker-compose up -d

Run the net alias script to make the broker addresses available on localhost:

./scripts/set_up_net_alias.sh

Apply the topic configs in examples/local-cluster/topics:

topicctl apply --skip-confirm examples/local-cluster/topics/*yaml

Send some test messages to the topic-default topic:

topicctl tester --broker-addr=localhost:9092 --topic=topic-default

Open up the repl (while keeping the tester running in a separate terminal):

topicctl repl --cluster-config=examples/local-cluster/cluster.yaml

Run some test commands:

get brokers
get topics
get partitions
get partitions topic-default
get offsets topic-default
tail topic-default

Increase the number of partitions in the topic-default topic by changing the partitions: ... value in topic-default.yaml to 9 and re-applying:

topicctl apply examples/local-cluster/topics/topic-default.yaml

Bring down the local cluster:

docker-compose down

Usage

Subcommands

apply

topicctl apply [path(s) to topic config(s)]

The apply subcommand ensures that the actual state of a topic in the cluster matches the desired state in its config. If the topic doesn't exist, the tool will create it. If the topic already exists but its cluster state is out-of-sync, then the tool will initiate the necessary changes to bring it into compliance.

See the Config formats section below for more information on the expected file formats.

bootstrap

topicctl [flags] bootstrap

The bootstrap subcommand creates apply topic configs from the existing topics in a cluster. This can be used to "import" topics not created or previously managed by topicctl. The output can be sent to either a directory (if the --output flag is set) or stdout.

The placement strategy for the bootstrapped topic configs will default to cross-rack unless a different one is set via the --placement-strategy flag.

By default, this does not include internal topics such as __consumer_offsets. If you would like to have these topics included, pass the --allow-internal-topics flag.

check

topicctl check [path(s) to topic config(s)]

The check command validates that each topic config has the correct fields set and is consistent with the associated cluster config. Unless --validate-only is set, it then checks the topic config against the state of the topic in the corresponding cluster.

create

topicctl create [flags] [command]

The create command creates resources in the cluster from a configuration file. Currently, only ACLs are supported. The create command is separate from the apply command as it is intended for usage with immutable resources managed by topicctl.

delete

topicctl delete [flags] [operation]

The delete subcommand deletes a particular resource type in the cluster. Currently, the following operations are supported: | Subcommand | Description | | --------- | ----------- | | delete acls [flags] | Deletes ACL(s) in the cluster matching the provided flags | | delete topic [topic] | Deletes a single topic in the cluster |

get

topicctl get [flags] [operation]

The get subcommand lists out the instances and/or details of a particular resource type in the cluster. Currently, the following operations are supported:

| Subcommand | Description | | --------- | ----------- | | get balance [optional topic] | Number of replicas per broker position for topic or cluster as a whole | | get brokers | All brokers in the cluster | | get config [broker or topic] | Config key/value pairs for a broker or topic | | get groups | All consumer groups in the cluster | | get lags [topic] [group] | Lag for each topic partition for a consumer group | | get members [group] | Details of each member in a consumer group | | get partitions [optional: topics] | Get all partitions for topics | | get offsets [topic] | Number of messages per partition along with start and end times | | get topics | All topics in the cluster | | get acls [flags] | Describe access control levels (ACLs) in the cluster | | get users | All users in the cluster |

rebalance

topicctl rebalance [flags]

The apply subcommand can be used with flag --rebalance rebalances a specified topics across a cluster.

The rebalance subcommand, on the other hand, performs a rebalance for all the topics defined at a given topic prefix path.

See the rebalancing section below for more information on rebalancing.

repl

topicctl repl [flags]

The repl subcommand starts up a shell that allows running the get and tail subcommands interactively.

reset-offsets

topicctl reset-offsets [topic] [group] [flags]

The reset-offsets subcommand allows resetting the offsets for a consumer group in a topic. There are 2 main approaches for setting the offsets:

Use a combination of --partitions, --offset, --to-earliest and --to-latest flags. --partitions flag specifies a list of partitions to be reset e.g. 1,2,3 .... If not used, the command defaults to resetting consumer group offsets for ALL of the partitions. --offset flag indicates the specific value that all desired consumer group partitions will be set to. If not set, it will default to -2. Finally, --to-earliest flag resets offsets of consumer group members to earliest offsets of partitions while --to-latest resets offsets of consumer group members to latest offsets of partitions. However, only one of the --to-earliest, --to-latest and --offset flags can be used at a time. This approach is easy to use but lacks the ability for detailed offset configuration.
Use --partition-offset-map flag to specify a detailed offset configuration for individual partitions. For example, 1=5,2=10,7=12,... means that the consumer group offset for partition 1 must be set to 5, partition 2 to offset 10, partition 7 to offset 12 and so on. This approach provides greater flexibility and fine-grained control for this operation. Note that --partition-offset-map flag is standalone and cannot be coupled with any of the previous flags.

tail

topicctl tail [flags] [topic]

The tail subcommand tails and logs out topic messages using the APIs exposed in kafka-go. It doesn't have the full functionality of kafkacat (yet), but the output is prettier and it may be easier to use in some cases.

tester

topicctl tester [flags]

The tester command reads or writes test messages in a topic. For testing/demonstration purposes only.

Specifying the target cluster

There are three ways to specify a target cluster in the topicctl subcommands:

--cluster-config=[path], where the refererenced path is a cluster configuration in the format expected by the apply command described above,
--zk-addr=[zookeeper address] and `--zk-prefix=[optional prefix for cluster in zoo

Topicctl

Install / Use

README