Connect to your cluster using Kafka CLI

This document describes how to connect to your Pulsar cluster using the Kafka CLI tool (v3.1.0) through OAuth2 authentication or Token authentication.

Note

This QuickStart assumes that you have created a Pulsar cluster with the Kafka protocol enabled, created a service account, and granted the service account produce and consume permissions to a namespace for the target topic.

Connect to your cluster through OAuth2 authentication

This section describes how to connect to your Pulsar cluster using the Kafka CLI tool through OAuth2 authentication.

Before you begin

Get the OAuth2 credential file.
1. On the left navigation pane, click Service Accounts.
2. In the row of the service account you want to use, in the Key File column, click the Download icon to download the OAuth2 credential file to your local directory.
Get the service URL of your Pulsar cluster.
1. On the left navigation pane, in the Admin area, click Pulsar Clusters.
2. Select the Details tab, and in the Access Points area, click Copy at the end of the row of the Kafka Service URL (TCP).

Steps

Download Kafka 3.1.0 release and extract it to the ~/kafka folder.

mkdir -p ~/kafka && cd ~/kafka
# download Kafka 3.1.0
curl -O https://archive.apache.org/dist/kafka/3.1.0/kafka_2.13-3.1.0.tgz
tar xzf ./kafka_2.13-3.1.0.tgz

Download the supplementary libraries for the Kafka client.

cd ~/kafka/kafka_2.13-3.1.0
# download supplementary libraries
curl -O https://repo1.maven.org/maven2/io/streamnative/pulsar/handlers/oauth-client/2.9.1.5/oauth-client-2.9.1.5.jar --output-dir ./libs
curl -O https://repo1.maven.org/maven2/org/apache/pulsar/pulsar-client-admin-api/2.9.2/pulsar-client-admin-api-2.9.2.jar --output-dir ./libs
curl -O https://repo1.maven.org/maven2/org/apache/pulsar/pulsar-client/2.9.2/pulsar-client-2.9.2.jar --output-dir ./libs
curl -O https://repo1.maven.org/maven2/org/apache/pulsar/pulsar-client-api/2.9.2/pulsar-client-api-2.9.2.jar --output-dir ./libs

Create an OAuth2 configuration file.
This example creates a file named kafka.properties, substituting the path to your downloaded OAuth2 credential file and the audience respectively.
```
# configure kafka.properties file.
echo 'sasl.login.callback.handler.class=io.streamnative.pulsar.handlers.kop.security.oauth.OauthLoginCallbackHandler
security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule \
  required oauth.issuer.url="https://auth.streamnative.cloud/"\
  oauth.credentials.url="file:///YOUR-KEY-FILE-PATH"\
  oauth.audience="YOUR-AUDIENCE-STRING";' > ~/kafka/kafka.properties
```
- required oauth.issuer.url: the OAuth2 authentication provider. You can get the value from your downloaded OAuth2 credential file.
- oauth.credentials.url: the path to your downloaded OAuth2 credential file.
- oauth.audience: the audience parameter is a combination of the urn:sn:pulsar, your organization name, and your Pulsar instance name.
Connect to the cluster through the OAuth2 authentication method.
a. Open a terminal and run a Kafka consumer to receive a message from the test-topic topic.
```
# run consumer
~/kafka/kafka_2.13-3.1.0/bin/kafka-console-consumer.sh \
    --bootstrap-server "your-pulsar-service-url" \
    --consumer.config ~/kafka/kafka.properties \
    --topic test-topic
```
- bootstrap-server: the Kafka service URL of your Pulsar cluster.
b. Open another terminal and run a Kafka producer to send a message to the test-topic topic.
```
# run producer
~/kafka/kafka_2.13-3.1.0/bin/kafka-console-producer.sh \
    --bootstrap-server "your-pulsar-service-url" \
    --producer.config ~/kafka/kafka.properties \
    --topic test-topic
```
You can type some messages, for example Hello, Kafka on Pulsar! and then press the Enter key to produce the message to the test-topic topic. Then, you should see this message on the consumer terminal.

Connect to your cluster through Token authentication

This section describes how to connect to your Pulsar cluster using the Kafka CLI tool through Token authentication.

Before you begin

Note

Before getting the token of a service account, verify that the service account is authorized as a superuser or an admin of the tenants and namespaces.
A token has a system-defined Time-To-Live (TTL) of 7 days. Before a token expires, ensure that you generate a new token for your service account.

Get the JWT token.
1. On the left navigation pane, click Service Accounts.
2. In the row of the service account you want to use, in the Token column, click Generate new token, then click the Copy icon to copy the token to your clipboard.
Get the service URL of your Pulsar cluster.
1. On the left navigation pane, in the Admin area, click Pulsar Clusters.
2. Select the Details tab, and in the Access Points area, click Copy at the end of the row of the Kafka Service URL (TCP).

Steps

Download Kafka 3.1.0 release and extract it to the ~/kafka folder.

mkdir -p ~/kafka && cd ~/kafka
# download Kafka 3.1.0
curl -O https://archive.apache.org/dist/kafka/3.1.0/kafka_2.13-3.1.0.tgz
tar xzf ./kafka_2.13-3.1.0.tgz

Download the supplementary libraries for the Kafka client.

cd ~/kafka/kafka_2.13-3.1.0
# download supplementary libraries
curl -O https://repo1.maven.org/maven2/io/streamnative/pulsar/handlers/oauth-client/2.9.1.5/oauth-client-2.9.1.5.jar --output-dir ./libs
curl -O https://repo1.maven.org/maven2/org/apache/pulsar/pulsar-client-admin-api/2.9.2/pulsar-client-admin-api-2.9.2.jar --output-dir ./libs
curl -O https://repo1.maven.org/maven2/org/apache/pulsar/pulsar-client/2.9.2/pulsar-client-2.9.2.jar --output-dir ./libs
curl -O https://repo1.maven.org/maven2/org/apache/pulsar/pulsar-client-api/2.9.2/pulsar-client-api-2.9.2.jar --output-dir ./libs

Create a token configuration file.

This example creates a file named kafka-token.properties, substituting YOUR-TOKEN with the token of your service account.

# configure kafka-token.properties file.
echo 'security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="public/default" password="token:YOUR-TOKEN";' > ~/kafka-token.properties

Connect to the cluster through the Token authentication method.
a. Open a terminal and run a Kafka consumer to receive a message from the kop-test-topic topic.
```
# run consumer
~/kafka/kafka_2.13-3.1.0/bin/kafka-console-consumer.sh \
    --bootstrap-server "your-pulsar-service-url" \
    --consumer.config ~/kafka/kafka-token.properties \
    --topic kop-test-topic
```
- bootstrap-server: the Kafka service URL of your Pulsar cluster.
b. Open another terminal and run a Kafka producer to send a message to the test-topic topic.
```
# run producer
~/kafka/kafka_2.13-3.1.0/bin/kafka-console-producer.sh \
    --bootstrap-server "your-pulsar-service-url" \
    --producer.config ~/kafka/kafka-token.properties \
    --topic kop-test-topic
```
You can type some messages, for example Hello, Kafka on Pulsar! and then press the Enter key to produce the message to the kop-test-topic topic. Then, you should see this message on the consumer terminal.