Manage ksqlDB by using the Confluent CLI

In addition to the Cloud Console, you can use the Confluent CLI to create ksqlDB applications, create Kafka topics, and control access to cluster resources.

Install the Confluent CLI

After you have a working Kafka cluster in Confluent Cloud, you can use the Confluent CLI to interact with your cluster from your local computer. For example, you can produce and consume to a topic by using the Confluent CLI.

Install Confluent CLI as describe in CLI installation guide.

Tip

For examples of using multiple languages to connect Kafka client applications to Confluent Cloud, see the Code Examples.

Log in to your cluster

In the following steps, you log in to your Kafka cluster and connect to your cluster with an API key that you create by using the Confluent CLI.

For more information about Confluent CLI commands, see Confluent CLI Command Reference.

  1. Log in to your Confluent Cloud cluster.

    confluent login

    Your output should resemble:

    Enter your Confluent credentials:
Email: jdoe@myemail.io
Password: ***********************

Logged in as "jdoe@myemail.io" for organization "b0b21724-4586-4a07-b787-8087daacbf87" ("my-org").

  2. View your cluster.

    confluent kafka cluster list

    Your output should resemble:

      Current |     ID     |    Name    | Type  | Provider | Region  | Availability | Status
----------+------------+------------+-------+----------+---------+--------------+---------
          | lkc-bc456j | my-cluster | BASIC | azure    | westus2 | single-zone  | UP

  3. Set the active Kafka cluster. In this example, the cluster ID is lkc-bc456j.

    confluent kafka cluster use lkc-bc456j

    Tip

    The lkc prefix in the cluster ID is an acronym for “logical Kafka cluster”.

Create a Kafka API key and secret

Run the following command to create an API key and secret. When you create an API key by using the CLI, it’s automatically stored locally.

confluent api-key create --resource lkc-bc456j

Your output should resemble:

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 | S808KPS5CCOLIRCT                                                 |
| Secret  | R8rh7HHSzTr8088LxNrtwc4cGwIZPyTkBjp4HzLBayavzACD+57CGRI1qlFeTOFN |
+---------+------------------------------------------------------------------+

Save the API key and secret output in a secure location. The secret cannot be retrieved later.

Create a ksqlDB cluster by using the Confluent CLI

Use the confluent ksql cluster create command to create a new ksqlDB cluster.

Run the following commmand to create a new ksqlDB cluster that has access to the same resources as the user account or service account that’s associated with the key.

confluent ksql cluster create my-ksqldb-cluster --credential-identity <user-id>

Your output should resemble:

+-------------------------+--------------------------------------------------------+
| Id                      | lksqlc-123456                                          |
| Name                    | my-ksqldb-cluster                                      |
| Topic Prefix            | pksqlc-ab123                                           |
| Kafka                   | lkc-bc456j                                             |
| Storage                 |                                                    500 |
| Endpoint                | https://pksqlc-ab123.westus2.azure.confluent.cloud:443 |
| Status                  | PROVISIONING                                           |
| Detailed Processing Log | true                                                   |
+-------------------------+--------------------------------------------------------+

Note

It may take a few minutes for the state of your new ksqlDB cluster to transition from PROVISIONING to UP.

Create a resource API key for ksqlDB

Starting with Confluent CLI v0.198.0, you can create API keys for particular resources, like a ksqlDB cluster. For more information, see Create a resource API key.

Important

The API key and secret that you create in this step are distinct from the Kafka key pair that you created previously. This key pair is for the ksqlDB cluster specifically and can be created only by using the confluent api-key create --resource <ksqldb-cluster-id> command.

Run the following command to see a list of all API keys associated with your ksqlDB cluster.

confluent api-key list --resource <ksqldb-cluster-id>

If the output is empty, run the following command to create an API key that is specific to your ksqlDB cluster.

confluent api-key create --resource <ksqldb-cluster-id>

Your output should resemble:

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 | 3B0BXEMZDEQIQIJD                                                 |
| Secret  | H8o8ZSQ63KyLv0wHzkCF+rOUme89poHA5EZitxE3olgSzRalK5xedtF6twTadPk7 |
+---------+------------------------------------------------------------------+

Save this resource API key and secret in a secure location. The secret cannot be retrieved later.

Use the ksqlDB cluster’s resource API key to connect the ksqlDB CLI to your hosted ksqlDB cluster and to send HTTPS requests to it.

Access a ksqlDB cluster by using a resource API key

Confluent Cloud ksqlDB supports authentication with a ksqlDB resource API key. You can use a resource API key to access the hosted ksqlDB cluster by using the ksqlDB CLI or HTTPS requests.

Run the confluent ksql cluster list command to get the URL of the ksqlDB endpoint.

confluent ksql cluster list

Your output should resemble:

       ID      |   Name      | Topic Prefix |   Kafka    | Storage |                         Endpoint                         | Status
---------------+-------------+--------------+------------+---------+----------------------------------------------------------+---------
  lksqlc-ab123 | ksqldb-app1 | pksqlc-zz321 | lkc-bc456j |     500 | https://pksqlc-zz321.us-central1.gcp.confluent.cloud:443 | UP

Follow these guidelines for both the ksqlDB CLI and REST API commands:

  • For <cloud-ksqldb-url>, use the endpoint value provided by the confluent ksql cluster list command, for example, https://pksqlc-zz321.us-central1.gcp.confluent.cloud:443.

  • For <ksqldb-specific-api-key> and <ksqldb-specific-secret>, use an API key provided by the confluent api-key create --resource <ksqldb-cluster-id> command.

    Important

    You must use a resource-specific key created for the ksqlDB cluster. API keys for Confluent Cloud or the Kafka cluster don’t work and cause an authorization error.

Using the ksqlDB CLI

To connect the ksqlDB CLI to a hosted ksqlDB cluster, run the following command.

  • Send your ksqlDB-specific API key and secret by using the -u and -p parameters.
$CONFLUENT_HOME/bin/ksql \
    -u <ksqldb-specific-api-key> \
    -p <ksqldb-specific-secret> \
    <ccloud-ksql-server-url>

Using HTTPS Requests

You can communicate with your hosted ksqlDB cluster by using the ksqlDB REST API.

Run the following curl command to send a POST request to the ksql endpoint. In this example, the request runs the LIST STREAMS statement and the response contains details about the streams in the ksqlDB cluster.

  • Specify --basic authentication in the Accept header of your request.
  • Send your ksqlDB-specific API key and secret, separated by a colon, as the --user credentials.
curl --http1.1 \
    -X "POST" "https://<cloud-ksqldb-url>/ksql" \
    -H "Accept: application/vnd.ksql.v1+json" \
    -H "Content-Type: application/json" \
    --basic --user "<ksqldb-specific-api-key>:<ksqldb-specific-secret>" \
    -d $'{
  "ksql": "LIST STREAMS;",
  "streamsProperties": {}
}'

Your output should resemble:

[
{
    "@type": "streams",
    "statementText": "LIST STREAMS;",
    "streams": [
    {
        "type": "STREAM",
        "name": "KSQL_PROCESSING_LOG",
        "topic": "pksqlc-zz321-processing-log",
        "keyFormat": "KAFKA",
        "valueFormat": "JSON",
        "isWindowed": false
    }
    ],
    "warnings": []
}
]

For more information, see ksqlDB API.

For an example that shows fully-managed Confluent Cloud connectors in action with Confluent Cloud ksqlDB, see the Cloud ETL Demo. This example also shows how to use Confluent CLI to manage your resources in Confluent Cloud.

../_images/topology.png

Mitigate a Forbidden Access error

When you use the Confluent CLI to work with ksqlDB cluster resources in Confluent Cloud, you may receive a 403 Forbidden Access error. You can run the command with the -vvv option to see details about the error.

The following example command shows an attempt to describe the ksqlDB cluster named lksqlc-99999.

confluent -vvv ksql cluster describe lksqlc-99999

The following example output shows a 403 error.

2023-05-30T18:29:56.976-0700 [DEBUG] KSQLService.Describe request: GET https://confluent.cloud/api/ksqls/lksqlc-99999?XXX_sizecache=0&account_id=env-ab1cd2m&id=lksqlc-99999
2023-05-30T18:29:57.213-0700 [DEBUG] KSQLService.Describe response: 403 Forbidden X-Request-Id:eef69c54d863adee82db6f41077acc9b Body: {"error":{"code":403,"message":"Forbidden Access","error_code":"forbidden_access"}}
request: GET https://confluent.cloud/api/ksqls/lksqlc-99999?XXX_sizecache=0&account_id=env-ab1cd2&id=lksqlc-99999
Error: Confluent Cloud backend error: error describing ksql cluster: Forbidden Access

An error like this may be caused by an invalid cluster ID or an authorization issue. Check with your admin to ensure that you have the proper authorization to access the ksqlDB cluster and its resources.