Get started with the Kafka protocol

This quick start assumes that you already have a StreamNative Cloud account with a valid form of payment.
If you cannot enable the Kafka protocol on an existing cluster, submit a ticket with this request to the support team.

This quick start guides you through how to do the following:

Set up a Kafka-enabled cluster in a new organization and a new instance.
Configure a Kafka client to produce and consume messages.

For a general quick start on creating a cluster and using a Pulsar Java client to produce and consume messages to the Pulsar cluster, see quick start - StreamNative Cloud.

Step 1: Log in to StreamNative Cloud Console

To log in to the Streamnative Cloud Console, navigate to the StreamNative Cloud Console login page, enter the email address and password, and then click Log in to log in to the StreamNative Cloud Console.

Step 2: Create an organization

On the left navigation pane, click Dashboard.
Click Create organization.
Enter a name for the organization and then click Create.

You might have to wait briefly for the organization to be created. After your new organization is created, proceed to creating an instance.

Step 3: Create an instance and a cluster

On the left navigation pane, click Dashboard.
Click the organization that you just created.
On the Create your first instance card, click List instances.
On the Instances page, click CREATE INSTANCE.
Select Deploy Hosted to start the instance creation process.
On the Instance Configuration page, enter a name for your instance, select an infrastructure pool and the Availability Zone (AZ), and then click Cluster Location.
On the Cluster Location page, enter a name for your cluster, select the cluster location, and then click Cluster Size.
On the Cluster Size page, configure the cluster, and then click Payment.
- On the Basic tab, select custom sizing options.
- On the Advanced tab, in the Features area, enable the Kafka Protocol option.
If needed, on the Payment page, in the Create Payment Method box, enter a valid credit card number, and then click Create Payment Method.
Click Finish.

The cluster page displays, showing the cluster creation process. The cluster is ready for use after all components have been successfully deployed.

Step 4: Create a service account

On the left navigation pane, click Service Accounts.
Click Create Service Account.
Enter a name for the service account, and then click Confirm.

After creating a service account, you need to grant the service account produce and consume permissions to a namespace on your Pulsar cluster.

Step 5: Grant service account permissions

On the left navigation pane, in the ADMIN section, click Tenants/Namespaces.
Select the Public tenant, then select the Default namespace under the tenant.
Select the POLICY tab.
In the Authorization area, click ADD ROLE and select the service account you just created in the previous section.
In the Authorization area, on the drop-down menu below the service name you just added, select the consume and produce roles. The roles are added to your service account.

Now, you can use the service account to connect to your cluster with the Kafka CLI tool or Kafka client.

Step 6: Connect to the cluster

This example shows how to connect to your cluster using the Kafka CLI tool (v3.1.0) through the OAuth2 authentication plugin. For details about how to connect to a cluster using other Kafka clients, see connect overview.

You can also view the code example through the StreamNative Cloud Console.

On the left navigation pane, under Resources, click Clients.
Select the Kafka tab to view code examples about connecting to your Pulsar cluster using a variety of Kafka clients.

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 Overview 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. For details about how to get the OAuth2 credential file, see before you begin.
- 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 before you begin.
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.

Next steps

Learn more about connecting to a Pulsar cluster using Kafka clients in the connect overview topic.