topicctl

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:

1. Run go install github.com/segmentio/topicctl/cmd/topicctl@latest
2. Clone this repo and run make install in the repo root
3. 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

1. Start up a 6 node Kafka cluster locally:

docker-compose up -d

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

./scripts/set_up_net_alias.sh

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

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

1. Send some test messages to the topic-default topic:

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

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

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

1. Run some test commands:

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

1. 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

1. 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:

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:

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:

1. 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.

2. 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:

1. --cluster-config=[path], where the refererenced path is a cluster configuration in the format expected by the apply command described above,
2. --zk-addr=[zookeeper address] and --zk-prefix=[optional prefix for cluster in zookeeper], or
3. --broker-addr=[bootstrap broker address]

All subcommands support the cluster-config pattern. The last two are also supported by the get, repl, reset-offsets, and tail subcommands since these can be run independently of an apply workflow.

Version compatibility

We've tested topicctl on Kafka clusters with versions between 0.10.1 and 2.7.1, inclusive.

Note, however, that clusters at versions prior to 2.4.0 cannot use broker APIs for applying and thus also require ZooKeeper API access for full functionality. See the cluster access details section below for more details.

If you run into any unexpected compatibility issues, please file a bug.

Config formats

topicctl uses structured, YAML-formatted configs for clusters and topics. These are typically source-controlled so that changes can be reviewed before being applied.

Clusters

Each cluster associated with a managed topic must have a config. These configs can also be used with the get, repl, reset-offsets, and tail subcommands instead of specifying a broker or ZooKeeper address.

The following shows an annotated example:

…

Note that the name, environment, region, and description fields are used for description/identification only, and don't appear in any API calls. They can be set arbitrarily, provided that they match up with the values set in the associated topic configs.

If the tool is run with the --expand-env option, then the cluster config will be prepreocessed using os.ExpandEnv at load time. The latter will replace references of the form $ENV_VAR_NAME or ${ENV_VAR_NAME} with the associated values from the environment.

Additionally, the Amazon Resource Name (ARN) of a secret in AWS Secrets Manager can be provided instead of the username and password. Topicctl will then retrieve the secret value from Secrets Manager and use it as the credentials. The secret in Secrets Manager must have a value in the format shown below, identical to what AWS MSK requires.

{
  "username": "alice",
  "password": "alice-secret"
}

An example of secrets manager being used can be seen below. Be sure to include the [6Random-Characters AWS Secrets Manager tac