> ## Documentation Index
> Fetch the complete documentation index at: https://docs.streamnative.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Connect to your cluster using Kafka CLI

This document describes how to connect to your StreamNative cluster using the Kafka CLI tool (v3.1.0) using either [OAuth2](#use-oauth2) or [API Keys](/cloud/security/authentication/service-accounts/use-api-keys/api-keys-overview#kafka-clients) authentication.

<Note title="Note">
  This QuickStart assumes that you have created a StreamNative 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.
</Note>

## Connect to your cluster using API keys

<span id="use-apikeys" />

This section describes how to connect to your StreamNative cluster using the Kafka Java client with [SASL/PLAIN authentication](/cloud/security/authentication/service-accounts/use-api-keys/api-keys-overview#kafka-clients).

### Before you begin

<Note title="Note">
  * Before using an API key, verify that the service account is authorized to access the resources, such as tenants, namespaces, and topics.
  * The password for different utilities as `kcat` will be equal to `token:<API KEY>`.
</Note>

You can follow the instructions to [create an API key](/cloud/security/authentication/service-accounts/use-api-keys/api-keys-overview#using-api-keys-to-connect-to-your-cluster) for the service account you choose to use.

### Steps

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

   ```bash theme={null}
   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
   ```

2. Download the supplementary libraries for the Kafka client.

   ```bash theme={null}
   cd ~/kafka/kafka_2.13-3.1.0
   # download supplementary libraries
   curl -O https://repo1.maven.org/maven2/io/streamnative/pulsar/handlers/oauth-client/3.1.0.1/oauth-client-3.1.0.1.jar --output-dir ./libs
   ```

3. Create a token configuration file.

   This example creates a file named `kafka-token.properties`, substituting `API-KEY` with an API key of your service account.

   Remind that the password is: `token:API-KEY`

   ```
   # 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
   ```

4. Connect to the cluster through the SASL/PLAIN authentication method.

   a. Open a terminal and run a Kafka consumer to receive a message from the `kop-test-topic` topic.

   ```bash theme={null}
   # 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 StreamNative cluster.

   b. Open another terminal and run a Kafka producer to send a message to the `test-topic` topic.

   ```bash theme={null}
   # 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.

## Connect to your cluster using OAuth2 authentication

<span id="use-oauth2" />

This section describes how to connect to your StreamNative 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 StreamNative 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

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

   ```bash theme={null}
   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
   ```

2. Download the supplementary libraries for the Kafka client.

   ```bash theme={null}
   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
   ```

3. 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
   ```

   * `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.

4. 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.

   ```bash theme={null}
   # 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 StreamNative cluster.

   b. Open another terminal and run a Kafka producer to send a message to the `test-topic` topic.

   ```bash theme={null}
   # 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.