# Kafka REST API Quick Start for Confluent Cloud The Apache Kafka® REST API is a set of cloud-native APIs for administering and using your Kafka clusters in Confluent Cloud. It’s a subset of the [Confluent Cloud REST APIs](/cloud/current/api.html#section/Introduction), listed under [KAFKA API (V3)](/cloud/current/api.html#tag/Cluster-(v3)) in the API reference. The API endpoint is available by default for all [cluster types](../clusters/cluster-types.md#cloud-cluster-types) and can be accessed via the Confluent Cloud Console. This quick start gets you up and running with administering your Apache Kafka® cluster using the [Confluent Cloud REST APIs](https://docs.confluent.io/cloud/current/ccloud/cluster-v-3/). #### NOTE Currently, these APIs support only BINARY, JSON and STRING. Therefore, the APIs do not currently support using [Schema Registry](../sr/schemas-manage.md#sr-prv) for validation of produced messages against the [supported schema formats](/platform/current/schema-registry/serdes-develop/index.html). ## What the Quick Start Covers With this quick start, you will learn how to use the [REST admin endpoints](https://docs.confluent.io/cloud/current/ccloud/cluster-v-3/) to: - View existing topics - Create a topic - Update the topic configuration - Get a full list of properties and values for a topic - Get the value for a specified property for a topic - Produce data to a topic - Delete the topic The Quick Start demos a small subset of the endpoints available, and shows how to use a few different types of calls. This is a primer, and therefore not comprehensive. A much wider range of functionality is available through the API. ## Prerequisites As a prerequisite, you’ll need a Kafka cluster in Confluent Cloud. You can use an existing cluster or create a new one, as described in [step 1 of the Quick Start for Apache Kafka using Confluent Cloud](../get-started/index.md#cloud-create-kafka-cluster). ## Usage Tips for the Examples The Quick Start shows how to use [curl](https://curl.haxx.se/) commands to test API calls. - For your API testing, you may want to use the `--silent` flag with `curl` and pipe the entire command through [jq](https://stedolan.github.io/jq/) to get nicely formatted output. For example: ```none curl --silent -H "Authorization: Basic " \ --request GET --url 'https:///kafka/v3/clusters//topics//configs' | jq ``` Output for commands in this quick start is shown as formatted, but you have to use `curl --silent` and `|jq` to get this, and those are not included in most of the example commands so as not to confuse this nice-to-have formatting with the core commands. - To send output to a file and also view it on-screen, use ` | tee notes.txt` to create a new file or overwrite a same-named file. Thereafter, use the `tee` command with the `-a` flag if you want to append to an existing file; for example, ` | tee -a notes.txt`. You can chain this after the `jq` pipe. ## Step 1: Find the REST endpoint address and cluster ID 1. Get the REST endpoint and cluster ID from the Confluent CLI or the Cloud Console UI: ### CLI 1. Log on to the Confluent CLI with `confluent login`. 2. Make sure you are viewing the environment you want with `confluent environment list`, `confluent environment use`. 3. List the clusters to get the IDs with the command `confluent kafka cluster list`. 4. Enter this command to get cluster details: ```bash confluent kafka cluster describe ``` ### Confluent Cloud 1. Sign in to the [Confluent Cloud Console](https://confluent.cloud/login). 2. Navigate to the cluster you want to use, and click **Cluster settings**. ![image](images/rest-api-qs-endpoint-on-ui.png)
2. Note the REST endpoint. This is the server where the REST APIs are hosted. Use this endpoint ID in place of the example server in the examples below. 3. Note the cluster ID. This is the ID of the Kafka cluster that will be administered using the REST API and should be used in place of the example cluster ID in the examples below. ## Step 2: Create credentials to access the Kafka cluster resources 1. Create a Confluent Cloud cluster-scoped API key and secret. To communicate with the REST API, you need a Confluent Cloud API key and API secret. You can generate an API key pair by using the CLI or Cloud Console. ### CLI 1. If you have not done so already, log on to the Confluent CLI with `confluent login`. 2. Run the following command to create the API key and secret for the cluster. ```bash confluent api-key create --resource --description ``` Your output should resemble: ```none It may take a couple of minutes for the API key to be ready. Save the API key and secret. The secret is not retrievable later. +---------+------------------------------------------------------------------+ | API Key | 1234WXYZ4321ZYXW | | Secret | 1aBcDEFG234Hy7CeGBoBDoBBsSttOMJ5oFUwwkhj7g7MlS3p01c99C6ao84pQb8X | +---------+------------------------------------------------------------------+ ``` ### Confluent Cloud 1. In [Confluent Cloud Console](https://confluent.cloud/login), navigate to the cluster you want to use, and from the left menu, select **Cluster Overview -> API Keys**. 2. Click **Create key** and follow the prompts to create a Global access API key. ![Confluent Cloud API keys](images/rest-qs-cloud-api-keys-on-cluster-ui.png)
For more information, see [Use API Keys to Authenticate to Confluent Cloud](../security/authenticate/workload-identities/service-accounts/api-keys/overview.md#cloud-api-keys). 2. Use the API key ID and corresponding secret to create the base64 encoded string used in the authorization header that will be included in the REST calls to the Confluent Cloud API. To learn more, see [Authentication](https://docs.confluent.io/cloud/current/api.html/#object-model) in the API documentation, which describes Cloud and Cluster API keys and base64 encoding. For example, to generate a base64 header from an API key on Mac OS: ```bash echo -n ":" | base64 ``` Your output from the above command will be a long alphanumeric string: the base64 encoded key and secret, which you will use as basic authorization in the REST API calls that follow. Save this so that you can use it in the next steps. 3. (Optional) Store the base64 encoded key and secret in an environment variable for use in commands. To make rest of the steps in this quick start more efficient, you can optionally store the base64 encoded key and secret in an environment variable, then use the variable in subsequent commands. For example, to store the encoded key and secret in a variable called `MYKEY`: ```bash MYKEY= ``` You can check the contents of the variable with `echo $MYKEY`. Then, use it for authorization in subsequent commands: `"Authorization: Basic $MYKEY"`. ## Step 3: List the topics already available on the Kafka cluster List the topics in the Kafka cluster: ```bash curl -H "Authorization: Basic " --request GET --url 'https:///kafka/v3/clusters//topics' ``` For example: ```bash curl -H "Authorization: Basic ABC123ABC" --request GET --url 'https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics' ``` The response lists any existing topics. In this example, there are no existing topics in this Kafka cluster: ```bash {"kind":"KafkaTopicList", "metadata":{ "self":"https://pkc-lzvrd.us-west4.gcp.confluent.cloud/kafka/v3/clusters/lkc-vo9pz/topics", "next":null }, "data":[] } ``` ## Step 4: Create a topic using Cluster Administration for the Kafka REST API Create a topic: ```bash curl -H "Authorization: Basic " -H 'Content-Type: application/json' \ --request POST --url 'https:///kafka/v3/clusters//topics' \ -d '{"topic_name": "", "partitions_count": , "replication_factor": }' ``` For example: ```bash curl -H "Authorization: Basic ABC123ABC" -H 'Content-Type: application/json' \ --request POST --url 'https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics' \ -d '{"topic_name": "testTopic1", "partitions_count": 5, "replication_factor": 3}' ``` The response gives information about the new topic: ```none { "kind": "KafkaTopic", "metadata": { "self": "https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1”, "resource_name": "crn:///kafka=lkc-vo9pz/topic=testTopic1" }, "cluster_id": "lkc-vo9pz", "topic_name": "testTopic1", "is_internal": false, "replication_factor": 3, "partitions_count": 5, "partitions": { "related": "https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1/partitions" }, "configs": { "related": "https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1/configs" }, "partition_reassignments": { "related": "https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1//partitions/-/reassignment" } } ``` ## Step 5: Administer topic configurations 1. View the default configuration for the topic created in [Step 4](#rest-api-qs-create-topic). You can get the full list of topic configuration properties with this API call: ```bash curl -H "Authorization: Basic " \ --request GET --url 'https:///kafka/v3/clusters//topics//configs' ``` For example, the following command gets the full list of configurations for `testTopic1`, pipes it through [jq](https://stedolan.github.io/jq/) for more readable output, and uses `tee` to send the output to a new file (called `testTopic1-configs.txt`) and simultaneously display it on-screen. Sending the output to a file will enable you to browse and search all the original configs, which might be tangentially useful for the next steps. ```bash curl --silent -H "Authorization: Basic ABC123ABC" \ --request GET --url 'https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1/configs' | jq | tee testTopic1-configs.txt ``` The response shows a list of topic configuration properties and the values these are set to: ```none { "kind": "KafkaTopicConfigList", "metadata": { "self": "https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1/configs", "next": null }, "data": [ { "kind": "KafkaTopicConfig", "metadata": { "self": "https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1/configs/cleanup.policy", "resource_name": "crn:///kafka=lkc-vo9pz/topic=testTopic1/config=cleanup.policy" }, "cluster_id": "lkc-vo9pz", "name": "cleanup.policy", "value": "delete", "is_read_only": false, "is_sensitive": false, "source": "DYNAMIC_TOPIC_CONFIG", "synonyms": [ { "name": "cleanup.policy", "value": "delete", "source": "DYNAMIC_TOPIC_CONFIG" }, { "name": "log.cleanup.policy", "value": "delete", "source": "DEFAULT_CONFIG" } ], "topic_name": "testTopic1", "is_default": false }, ... ``` 2. Edit the topic configuration. You can find a full list of configuration properties in the [Topic Configurations](/platform/current/installation/configuration/topic-configs.html) reference in the Confluent Platform documentation. To edit the topic properties, add the property name to the end of the URL: ```bash curl -H "Authorization: Basic " -H 'Content-Type: application/json' \ --request PUT --url 'https:///kafka/v3/clusters//topics/‘/configs/ \ -d '{"value": “”}’ ``` For example, to change the value of [retention.ms](/platform/current/installation/configuration/topic-configs.html#topicconfigs_retention.ms), from its default value of 604800000 (7 days) to 259200000 (3 days): ```bash curl -H "Authorization: Basic ABC123ABC" -H 'Content-Type: application/json' --request PUT \ --url 'https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1/configs/retention.ms' \ -d '{"value": "259200000"}' ``` No response payload is returned for this call. 3. View the updated topic configuration. To view the value for a single property, specify the property name in the URL: ```none curl -H "Authorization: Basic " --request GET --url 'https:///kafka/v3/clusters//topics//configs/’ ``` For example: ```bash curl -H "Authorization: Basic ABC123ABC" --request GET --url 'https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1/configs/retention.ms' ``` The response payload shows the current value of the topic configuration property `retention.ms`: ```none { "kind": "KafkaTopicConfig", "metadata": { "self": “https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1/configs/retention.ms", "resource_name": "crn:///kafka=lkc-vo9pz/topic=testTopic1/config=retention.ms" }, "cluster_id": "lkc-vo9pz", "name": "retention.ms", "value": "259200000", "is_read_only": false, "is_sensitive": false, "source": "DYNAMIC_TOPIC_CONFIG", "synonyms": [ { "name": "retention.ms", "value": "259200000", "source": "DYNAMIC_TOPIC_CONFIG" } ], "topic_name": "testTopic1", "is_default": false } ``` 4. Batch update topic configurations. To update multiple configuration properties with a single REST call, use a batch update: ```none curl -H "Authorization: Basic " -H 'Content-Type: application/json' \ --request POST --url 'https:///kafka/v3/clusters//topics//configs:alter' \ -d '{"data": [{"name": “”, "value": "", {"name": “”, "value": “}…]}’ ``` For example, this API call updates [retention.ms](/platform/current/installation/configuration/topic-configs.html#topicconfigs_retention.ms) again, this time to 172800000 (2 days) and [segment.bytes](/platform/current/installation/configuration/topic-configs.html#topicconfigs_segment.bytes) from its default (1073741824) to 123456789: ```none curl -H "Authorization: Basic ABC123ABC" -H 'Content-Type: application/json' --request POST --url 'https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1/configs:alter' -d '{"data": [{"name": "retention.ms", "value": "172800000"}, {"name": "segment.bytes", "value": "123456789"}]}' ``` 5. View the updated values for topic configuration. You can view the updates as shown in the previous step, substituting `retention.ms` and then `segment.bytes` for `` in this API call: ```none curl -H "Authorization: Basic " --request GET --url 'https:///kafka/v3/clusters//topics//configs/’ ``` Alternatively, sign in to the [Confluent Cloud Console](https://confluent.cloud/login), navigate to `testTopic1` on your cluster, click the **Configuration** tab, and then click **Show full config**. ![image](images/rest-api-qs-topic-configs-ui.png) ## Step 6: Create a topic and configure properties simultaneously You can also update topic configurations as a part of creating a topic. To try this out, create a new topic. For example: ```bash curl --silent -H "Authorization: Basic TOKEN" -H 'Content-Type: application/json' --request POST --url \ 'https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics' \ -d '{"topic_name": "testTopic2", "partitions_count": 4, "replication_factor": 3, "configs":[{"name": "retention.ms", "value": 98765},{"name": "segment.bytes", "value":"98765432"}]}' | jq ``` This creates the topic `testTopic2`, specifies `replication_factor` and `partitions_count` (as in the previous example), and also provides new values for `retention.ms` and `segment.bytes` at time of topic creation. ## Step 7: Produce data to the topic To produce to a topic, provide JSON data and a base64-encoded API key and secret to the REST Produce endpoint ([/kafka/v3/clusters//topics//records](https://docs.confluent.io/cloud/current/ccloud/records-v-3/)) as shown in the examples below. As you work through the examples, have the Confluent Cloud Console opened to the cluster and topic **Messages** tab where you are sending data. This way, you can watch the messages show up when you send the data. You will miss the messages if you do not have the Confluent Cloud Console open as you make the `records` request. This example sends messages to `testTopic1` in `MY-FIRST-CLUSTER`. ### Streaming mode (recommended for sending a batch of records) For additional examples, see the following note: #### NOTE For additional examples, see the public GitHub site [kafka-rest/examples/produce_v3](https://github.com/confluentinc/kafka-rest/tree/master/examples/produce_v3). Streaming mode is the more efficient way to send multiple records. In streaming mode, you can open an interactive terminal connection as part of the API call, and send multiple records over a single stream. To use this mode, set an additional header `"Transfer-Encoding: chunked”` on the initial request: ```none curl -X POST -H "Transfer-Encoding: chunked" -H "Content-Type: application/json" \ -H "Authorization: Basic " \ /kafka/v3/clusters//topics//records -T- ``` For example: ```none curl -X POST -H "Transfer-Encoding: chunked" -H "Content-Type: application/json" \ -H "Authorization: Basic "\ https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1/records -T- {"value": {"type": "JSON", "data": "Hello World!"}} {"value": {"type": "JSON", "data": "Hola Mundo!"}} {"value": {"type": "JSON", "data": "Hola Mundo, otra vez!"}} ``` View the **Messages** tab for testTopic1 on the Confluent Cloud Console (as you are sending the message): ![image](images/rest-api-qs-produce-batch-records-streaming.png) ### Non-streaming mode (not recommended) The syntax to produce a single record to a topic is as follows: ```none curl -X POST -H "Content-Type: application/json" \ -H "Authorization: Basic " \ '/kafka/v3/clusters//topics//records' -d \ '{"value": {"type": "", "data": ""}}' ``` For `type`, current options are `BINARY`, `JSON` or `STRING`. For example, this call sends the message “Bonjour le monde!” as JSON data to the topic, testTopic1: ```none curl -X POST -H "Content-Type: application/json" \ -H "Authorization: Basic " "https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/testTopic1/records" -d \ '{"value": {"type": "JSON", "data": "Bonjour le monde!"}}' ``` View the **Messages** tab for testTopic1 on the Confluent Cloud UI: ![image](images/rest-api-qs-produce-single-record.png) You can use non-streaming mode to send multiple records in a single request by appending records together and in the data payload of a single, non-streamed request as shown in the following example: ```none curl -X POST -H "Content-Type: application/json" -H \ "Authorization: Basic ” \ "https:///kafka/v3/clusters//topics//records" -d \ '{"partition_id": "1", "value": {"type": "JSON", "data": "Bonjour le monde!"}} {"partition_id": "10", "value": {"type": "JSON", "data": "Bonjour le monde, de nouveau!"}}' ``` Note that each individual record sent over the endpoint can have its own partition specified, so you can produce to multiple partitions in the same `/records` call. That said, streaming mode is more efficient for sending multiple records, as shown in the next section. The performance difference is quite marked; under a hundred requests per second for individual calls, as compared to several 1000 per second for streamed. ## Step 8: Delete the topics When you are ready to quit the Quick Start demo or when the topic is no longer needed, you can delete the sample topics. 1. To delete a topic, use this API call: ```bash curl -H "Authorization: Basic " -H 'Content-Type: application/json' --request DELETE --url 'https:///kafka/v3/clusters//topics/' ``` For example: ```none curl -H "Authorization: Basic ABC123ABC" -H 'Content-Type: application/json' --request DELETE --url 'https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/‘testTopic1' ``` ```none curl -H "Authorization: Basic ABC123ABC" -H 'Content-Type: application/json' --request DELETE --url 'https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics/‘testTopic2' ``` This call has no response payload. 2. Confirm that topic was deleted by listing the topics in the cluster: ```bash curl -H "Authorization: Basic " --request GET --url 'https:///kafka/v3/clusters//topics' ``` For example: ```bash curl -H "Authorization: Basic ABC123ABC" --request GET --url 'https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics' ``` The response shows that there are no topics in the cluster: ```none { "kind": "KafkaTopicList", "metadata": { "self": "https://pkc-abcde.us-west4.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-vo9pz/topics", "next": null }, "data": [] } ``` ## Suggested Resources - [Confluent Cloud REST APIs](https://docs.confluent.io/cloud/current/ccloud/cluster-v-3/) - [Confluent Cloud APIs overview](https://docs.confluent.io/cloud/current/api.html/#introduction) - [Confluent CLI Install](https://docs.confluent.io/confluent-cli/current/install.html) - [Confluent CLI Command Reference](https://docs.confluent.io/confluent-cli/current/command-reference/confluent_completion.html) - [Use API Keys to Authenticate to Confluent Cloud](../security/authenticate/workload-identities/service-accounts/api-keys/overview.md#cloud-api-keys) - [Quick Start for Confluent Cloud](../get-started/index.md#cloud-quickstart)