Confluent Cloud CLI: Command Example for Apache Kafka®¶

In this tutorial, you will use the Confluent Cloud CLI to produces messages to and consumes messages from an Apache Kafka® cluster.

After you run the tutorial, use the provided source code as a reference to develop your own Kafka client application.

Prerequisites¶

Client¶

Local install of Confluent Cloud CLI v1.13.0 or later.
timeout: used by the bash scripts to terminate a consumer process after a certain period of time. timeout is available on most Linux distributions but not on macOS. macOS users should view the installation instructions for macOS.

Kafka Cluster¶

You can use this tutorial with a Kafka cluster in any environment:
- In Confluent Cloud
- On your local host
- Any remote Kafka cluster
If you are running on Confluent Cloud, you must have access to a Confluent Cloud cluster with an API key and secret.
- The first 20 users to sign up for Confluent Cloud and use promo code C50INTEG will receive an additional $50 free usage (details)
- For an automated way to create a Kafka cluster, credentials, and ACLs in Confluent Cloud, see ccloud-stack Utility for Confluent Cloud.

Setup¶

Clone the confluentinc/examples GitHub repository and check out the 6.1.0-post branch.

git clone https://github.com/confluentinc/examples
cd examples
git checkout 6.1.0-post

Change directory to the example for Confluent Cloud CLI.
```
cd clients/cloud/ccloud/
```
Log in to Confluent Cloud with the command ccloud login, and use your Confluent Cloud username and password. The --save argument saves your Confluent Cloud user login credentials or refresh token (in the case of SSO) to the local netrc file.
```
ccloud login --save
```

Basic Producer and Consumer¶

In this example, the producer application writes Kafka data to a topic in your Kafka cluster. If the topic does not already exist in your Kafka cluster, the producer application will use the Kafka Admin Client API to create the topic. Each record written to Kafka has a key representing a username (for example, alice) and a value of a count, formatted as json (for example, {"count": 0}). The consumer application reads the same Kafka topic and keeps a rolling sum of the count as it processes each record.

Produce Records¶

Create the topic in Confluent Cloud.
```
ccloud kafka topic create test1
```
Run the Confluent Cloud CLI producer, writing messages to topic test1, passing in arguments for:
- --parse-key --delimiter ,: pass key and value, separated by a comma
```
ccloud kafka topic produce test1 --parse-key --delimiter ,
```
Type a few messages, using a , as the separator between the message key and value:
```
alice,{"count":0}
alice,{"count":1}
alice,{"count":2}
```
When you are done, press Ctrl-C.
View the producer code.

Consume Records¶

Run the Confluent Cloud CLI consumer, reading messages from topic test1, passing in arguments for:
- -b: print all messages from the beginning of the topic -
- --print-key: print key and value (by default, it only prints value)
```
ccloud kafka topic consume test1 -b --print-key
```
Verify that the consumer received all the messages. You should see:
```
alice   {"count":0}
alice   {"count":1}
alice   {"count":2}
```
When you are done, press CTRL-C.
View the consumer code.

Avro And Confluent Cloud Schema Registry¶

This example is similar to the previous example, except the value is formatted as Avro and integrates with the Confluent Cloud Schema Registry.

Before using Confluent Cloud Schema Registry, check its availability and limits.

As described in the Quick Start for Schema Management on Confluent Cloud in the Confluent Cloud GUI, enable Confluent Cloud Schema Registry and create an API key and secret to connect to it.
Verify that your VPC can connect to the Confluent Cloud Schema Registry public internet endpoint.

Produce Avro Records¶

Create the topic in Confluent Cloud.
```
ccloud kafka topic create test2
```

Create a file, for example schema.json, that has the schema of your message payload.

echo '{"type":"record","name":"myrecord","fields":[{"name":"count","type":"int"}]}' > schema.json

Run the Confluent Cloud CLI producer writing messages to topic test2, passing in arguments for:
- --value-format avro: use Avro data format for the value part of the message
- --schema: the path to the schema file
- --parse-key --delimiter ,: pass key and value, separated by a comma
```
ccloud kafka topic produce test2 --value-format avro --schema schema.json --parse-key --delimiter ,
```
Note

The first time you run this command, you must provide user credentials for Confluent Cloud Schema Registry.
Type a few messages, using a , as the separator between the message key and value:
```
alice,{"count":3}
alice,{"count":4}
alice,{"count":5}
```
When you are done, press Ctrl-C.
View the producer Avro code.

Consume Avro Records¶

Run the Confluent Confluent CLI consumer reading messages from topic test2, passing in arguments for:
- -b: print all messages from the beginning of the topic
- --value-format avro: use Avro data format for the value part of the message
- --print-key: print key and value (by default, it only prints value)
```
ccloud kafka topic consume test2 -b --value-format avro --print-key
```
Verify that the consumer received all the messages. You should see:
```
alice   {"count":3}
alice   {"count":4}
alice   {"count":5}
```
When you are done, press Ctrl-C.
View the consumer Avro code.