Connect to cluster through Kafka CLI

This document assumes that you have created a Pulsar cluster and a service account, and have granted the service account produce and consume permissions to a namespace on your Pulsar cluster. For details, see get started with the Kafka protocol.

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

Prerequisites

Get the OAuth2 credential file. For details, see get an OAuth2 credential file.
Get the service URL of your Pulsar cluster. For details, see get a service URL.

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. For details about how to get the OAuth2 credential file, see get an 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 plugin.
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 service URL of your Pulsar cluster. For details, see get a service URL.
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.