Instaclustr has several different tools that you can use to manage Topics in Apache Kafka. These include a CLI tool, the Instaclustr API, and the Instaclustr Terraform Provider.

Table of Contents

Kafka Topic Management via CLI

Kafka comes with its own topic management tools that can be downloaded from here. For our offering of Kafka 2.6.1 and newer, you can use Kafka-topics.sh tool that they have provided. However, for our older offerings, you will need to use ic-Kafka-topics.

In the following instructions, we will assume the use of ic-Kafka-topics.

Introduction to Kafka Topic Management

Ic-Kafka-topics is a tool developed by Instaclustr that can be used to manage Kafka topics using a connection to a Kafka broker.  Ic-Kafka-topics is based on the standard Kafka-topics tool, but unlike Kafka-topics, it does not require a zookeeper connection to work.

The core “actions” supported by ic-Kafka-topics include:

• list – list the topics available on the cluster
• create – create a topic
• describe – provide details of one or more topics
• alter – change the properties of one or more topics
• delete – delete one or more topics

Installing ic-kafka-topics

Ic-Kafka-topics is available as a tarfile, downloadable from the Instaclustr Console cluster connection page.  To install:

1. Download the tarfile from the Instaclustr console
2. Unpack the tarfile:
   tar -xf kafka_2.11-2.1.2.tgz
3. Go to the bin directory of the Kafka installation
   > cd kafka_2.11-2.1.2/bin
4. Run ic-kafka-topics.sh
   > ./ic-kafka-topics.sh

This should display help text for ic-Kafka-topic.sh:

Using ic-kafka-topics

Once you have downloaded and installed ic-Kafka-topics, you can use it to manage topics with your cluster.

To enable ic-Kafka-topics to connect to your cluster you will always need to provide two options to the tool:

A minimal properties file requires your userid and password in the following format:

(carefully note the placement of the backslashes and semicolon at the end of the lines)

If your Kafka cluster has client ⇆ broker encryption enabled your properties file should look like this, ensuring the password and truststore location are correct:

You can find out more about using certificates with Kafka clients including where to find them here.

All usage of the ic-Kafka-topics command will specify these details.  For example, to list the topics on your cluster, run:

The following sections describe each of the actions supported by the tool.  You can get help at any time at the command line by running ic-Kafka-topics.sh with no arguments.

Note: To connect to your Kafka cluster over the private network, use port 9093 instead of 9092.

List

The List action provides a listing of the topics that are available on the cluster.  To use the List action, include the ‑‑list option

Example:

Create

The “Create” action is used to create a new topic.

For simple topic creation

For simple topic creation: use the Create action.  You must provide the following details:

Example:

The following example will create a topic called “mytopic” with 9 partitions and a replication factor of 3.

Kafka will automatically assign partitions to available brokers.

Creating topics with a replica-assignment

If you want more control about the assignment of the new topic’s partitions to existing brokers, you may provide a manual partition to broker assignment using the ‑‑replica-assignment option.

Describe

The “describe” action is used to provide a detailed description of a topic.  You need to provide the following details

Only configurations which are not set to defaults will be displayed.  This includes values inherited from broker default configs or values which have been set for the topic.

For more details see the description of the output format for the standard Kafka-topics tool here:

Example:

Alter

The “alter” action is used to set custom configurations on the topic.  You can configure additional custom configurations on a property using the ‑‑alter action.

To set a configuration on a topic you must specify the following options:

Examples

The following command will alter the retention.ms configuration on the topic called “mytopic“.

The following command will alter two values (retention.ms and compression.type) on the topic called “mytopic”

The following command will set retention.ms to 50000000 on all topics starting with the string “test”

The following command will change the min.insync.replicas property value for this topic and override the default min.insync.replicas value of the broker.

Deleting a configuration

To remove a custom configuration from a topic, use ‑‑alter combined with the ‑‑delete-config option.  The ‑‑delete-config option should specify the name of the configuration to remove.  Removing a custom configuration will cause the value to revert to the default value.

Example:

The following command will remove the retention.ms configuration, reverting it to the default value.

Delete (Topic Deletion)

The Delete action is used to a delete a topic.  To delete a topic you must specify the following options:

Examples

The following command will delete the topic called “mytopic“.

The following command will delete the all topics starting with “dev”.

Kafka Topic Management via Instaclustr API

Instaclustr provides Topic Management API for Kafka clusters to help you with managing topics. The Topic Management API supports the following functions:

• List Topics
• Create a Topic
• Delete a Topic
• Get Topic information (number of partitions and replication factor)
• Get Topic configuration
• Modify Topic configuration

These examples show how to use the features of the Instaclustr Topic Management API.

For each endpoint listed below, all requests must include basic authentication details:

List Topics

1. To retrieve a list of topics currently available in the Kafka cluster, make a GET request to:
   
   https://api.instaclustr.com/provisioning/v1/<cluster_id>/kafka/topics

   1	https://api.instaclustr.com/provisioning/v1/<cluster_id>/kafka/topics

2. On a successful request, the API should respond with a 200 status code and a JSON containing the list of topics. E.g.:
   
   { "topics": [ "test-topic" ] }

   1 2 3 4 5

   {   "topics": [      "test-topic"   ] }

Create a Topic

1. To create a new topic to the Kafka cluster make a POST request to:
   
   https://api.instaclustr.com/provisioning/v1/<cluster_id>/kafka/topics

   1	https://api.instaclustr.com/provisioning/v1/<cluster_id>/kafka/topics

   A request body should be included, specifying the topic name, number of partitions, and replication factor. E.g.:
   

   { "topic": "test-topic", "partitions": 3, "replicationFactor": 3 }

   1 2 3 4 5

   {   "topic": "test-topic",   "partitions": 3,   "replicationFactor": 3 }

   Number of partitions and the replication factor should be non-zero integers. The replication factor is limited by the number of available brokers in the clusters. Be aware of creating a topic with too many partitions, as it can slow down the cluster.

2. On successful topic creation, the API should respond with a 201 status code and a short message. E.g.
   
   { "message": "Kafka topic test-topic created." }

   1 2 3

   {   "message": "Kafka topic test-topic created." }

   Once this message is received, the topic should be immediately available in the Kafka cluster.

Delete a Topic

1. To delete an existing topic from the Kafka cluster make a DELETE request to:
   
   https://api.instaclustr.com/provisioning/v1/<cluster_id>/kafka/topics/<topic-name>

   1	https://api.instaclustr.com/provisioning/v1/<cluster_id>/kafka/topics/<topic-name>

   No request body is required. The topic to be deleted is specified in the URL.

2. When the topic is successfully deleted, the API should respond with a 200 status code and a short message. E.g.
   
   { "message": "Kafka topic test-topic has been deleted." }

   1 2 3

   {   "message": "Kafka topic test-topic has been deleted." }

Get Topic Information

1. To retrieve the topic information of a particular topic available in the Kafka cluster, make a GET request to:
   
   https://api.instaclustr.com/provisioning/v1/<cluster_id>/kafka/topics/<topic-name>

   1	https://api.instaclustr.com/provisioning/v1/<cluster_id>/kafka/topics/<topic-name>

2. The API should respond with a 200 status code and a JSON containing the information of the requested topic. E.g.:
   
   { "topic": "test-topic", "partitions": 3, "replicationFactor": 3 }

   1 2 3 4 5

   {   "topic": "test-topic",   "partitions": 3,   "replicationFactor": 3 }

Get Topic Configuration

• To retrieve the topic configuration of a particular topic available in the Kafka cluster, make a GET request to:
  
  https://api.instaclustr.com/provisioning/v1/<cluster_id>/kafka/topics/<topic-name>/config

  1	https://api.instaclustr.com/provisioning/v1/<cluster_id>/kafka/topics/<topic-name>/config

• On a successful request, the API should respond with a 200 status code and a JSON containing the configuration of the requested topic. E.g.:
  
  { "topic": "test-topic", "config": { "compression.type": "producer", "leader.replication.throttled.replicas": "", "remote.storage.enable": "false", "message.downconversion.enable": "true", "min.insync.replicas": "2", "segment.jitter.ms": "0", "local.retention.ms": "-2", "cleanup.policy": "delete", "flush.ms": "9223372036854775807", "follower.replication.throttled.replicas": "*", "segment.bytes": "1073741824", "retention.ms": "604800000", "flush.messages": "9223372036854775807", "message.format.version": "3.0-IV1", "max.compaction.lag.ms": "9223372036854775807", "file.delete.delay.ms": "60000", "max.message.bytes": "1048588", "min.compaction.lag.ms": "0", "message.timestamp.type": "CreateTime", "local.retention.bytes": "-2", "preallocate": "false", "min.cleanable.dirty.ratio": "0.5", "index.interval.bytes": "4096", "unclean.leader.election.enable": "false", "retention.bytes": "-1", "delete.retention.ms": "86400000", "segment.ms": "604800000", "message.timestamp.difference.max.ms": "9223372036854775807", "segment.index.bytes": "10485760" } }

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34

  {   "topic": "test-topic",   "config": {     "compression.type": "producer",     "leader.replication.throttled.replicas": "",     "remote.storage.enable": "false",     "message.downconversion.enable": "true",     "min.insync.replicas": "2",     "segment.jitter.ms": "0",     "local.retention.ms": "-2",     "cleanup.policy": "delete",     "flush.ms": "9223372036854775807",     "follower.replication.throttled.replicas": "*",     "segment.bytes": "1073741824",     "retention.ms": "604800000",     "flush.messages": "9223372036854775807",     "message.format.version": "3.0-IV1",     "max.compaction.lag.ms": "9223372036854775807",     "file.delete.delay.ms": "60000",     "max.message.bytes": "1048588",     "min.compaction.lag.ms": "0",     "message.timestamp.type": "CreateTime",     "local.retention.bytes": "-2",     "preallocate": "false",     "min.cleanable.dirty.ratio": "0.5",     "index.interval.bytes": "4096",     "unclean.leader.election.enable": "false",     "retention.bytes": "-1",     "delete.retention.ms": "86400000",     "segment.ms": "604800000",     "message.timestamp.difference.max.ms": "9223372036854775807",     "segment.index.bytes": "10485760"    } }

Modify a Topic Configuration

1. To modify the configuration of a topic in the Kafka cluster, make a PUT request to:
   
   https://api.instaclustr.com/provisioning/v1/<cluster_id>/kafka/topics/<topic-name>/config

   1	https://api.instaclustr.com/provisioning/v1/<cluster_id>/kafka/topics/<topic-name>/config

   A request body should be included, specifying the configuration that you want to change. E.g.:
   

   { "config": { "min.insync.replicas": 2, } }

   1 2 3 4 5

   {   "config": {     "min.insync.replicas": 2,   } }

   We do not need to provide all of the available keys in the request body, only the keys of interest are required in the body.

   For available configuration keys and values, consult https://kafka.apache.org/23/documentation.html#topicconfigs . Alternatively, available configuration keys can also be inferred from the output of the get topic configuration endpoint. Note if you are using Kafka version that is newer than 2.3: Currently, we can’t change the message.format.version to a version that is newer than 2.3. Please use the Kafka CLI tools (kafka-configs.sh) to modify this configuration.

2. The API should respond with a 200 status code and a short message. E.g.
   
   { "message": "Kafka topic test-topic has been modified." }

   1 2 3

   {   "message": "Kafka topic test-topic has been modified." }

Kafka Topic Management via Instaclustr Terraform Provider

Instaclustr also provides a way to manage Kafka ACL using Terraform. For more information and examples, have a look at

The documentation for the ACL resource is located in docs/resources/kafka_topic.md and the ACL data source is located in docs/data-sources/kafka_topic.md .