Replicate Pub/Sub messages to Kafka

This tutorial shows how to ingest messages from Pub/Sub into your Managed Service for Apache Kafka cluster using Kafka Connect.

Kafka Connect manages data movement between your Kafka cluster and other systems. In this tutorial, you create a Connect cluster , and a Pub/Sub Source connector . The Pub/Sub Source connector reads messages from your Pub/Sub topic and writes them to a Kafka topic.

Before you begin

Console

Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Go to project selector

Verify that billing is enabled for your Google Cloud project .

Enable the Managed Kafka API.

Enable the API

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Go to project selector

Verify that billing is enabled for your Google Cloud project .

Enable the Managed Kafka API.

Enable the API

Make sure that you have the following role or roles on the project: Managed Kafka Cluster Editor , Managed Kafka Connect Cluster Editor , Managed Kafka Connector Editor , Managed Kafka Topic Editor , Pub/Sub Editor
Check for the roles
1. In the Google Cloud console, go to the IAM page.
  Go to IAM
2. Select the project.
3. In the Principal column, find all rows that identify you or a group that you're included in. To learn which groups you're included in, contact your administrator.
4. For all rows that specify or include you, check the Role column to see whether the list of roles includes the required roles.
Grant the roles
1. In the Google Cloud console, go to the IAM page.
  Go to IAM
2. Select the project.
3. Click Grant access .
4. In the New principals field, enter your user identifier. This is typically the email address for a Google Account.
5. In the Select a role list, select a role.
6. To grant additional roles, click Add another role and add each additional role.
7. Click Save .

gcloud

Install the Google Cloud CLI.

If you're using an external identity provider (IdP), you must first sign in to the gcloud CLI with your federated identity .

To initialize the gcloud CLI, run the following command:

gcloud  
init

Create or select a Google Cloud project .

Create a Google Cloud project:
```
gcloud projects create PROJECT_ID 
```
Replace PROJECT_ID with a name for the Google Cloud project you are creating.
Select the Google Cloud project that you created:
```
gcloud config set project PROJECT_ID 
```
Replace PROJECT_ID with your Google Cloud project name.

Verify that billing is enabled for your Google Cloud project .

Enable the Managed Kafka API:

gcloud  
services  
 enable 
  
managedkafka.googleapis.com

Install the Google Cloud CLI.

If you're using an external identity provider (IdP), you must first sign in to the gcloud CLI with your federated identity .

To initialize the gcloud CLI, run the following command:

gcloud  
init

Create or select a Google Cloud project .

Create a Google Cloud project:
```
gcloud projects create PROJECT_ID 
```
Replace PROJECT_ID with a name for the Google Cloud project you are creating.
Select the Google Cloud project that you created:
```
gcloud config set project PROJECT_ID 
```
Replace PROJECT_ID with your Google Cloud project name.

Verify that billing is enabled for your Google Cloud project .

Enable the Managed Kafka API:

gcloud  
services  
 enable 
  
managedkafka.googleapis.com

Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/managedkafka.clusterEditor, roles/managedkafka.connectClusterEditor, roles/managedkafka.connectorEditor, roles/managedkafka.topicEditor, roles/pubsub.editor
```
gcloud  
projects  
add-iam-policy-binding  
 PROJECT_ID 
  
--member = 
 "user: USER_IDENTIFIER 
" 
  
--role = 
 ROLE 
```
Replace the following:
- PROJECT_ID : your project ID.
- USER_IDENTIFIER : the identifier for your user account—for example, myemail@example.com .
- ROLE : the IAM role that you grant to your user account.

Create a Pub/Sub topic and subscription

In this step, you create a Pub/Sub topic with a subscription.

Console

Go to the Pub/Sub > Topicspage.

Go to Topics
Click Create topic.
In the Topic IDbox, enter a name for the topic.
Make sure that the Add a default subscriptioncheckbox is selected.
Click Create.

gcloud

To create a Pub/Sub topic, run the gcloud pubsub topics create command.
```
 gcloud  
pubsub  
topics  
create  
 TOPIC_ID 
 
```
Replace TOPIC_ID with a name for your Pub/Sub topic.
To create a subscription to your topic, run the gcloud pubsub subscriptions create command:
```
 gcloud  
pubsub  
subscriptions  
create  
--topic  
 TOPIC_ID 
  
 SUBSCRIPTION_ID 
 
```
Replace SUBSCRIPTION_ID with a name for your Pub/Sub subscription.

For information about how to name Pub/Sub topics and subscriptions, see Guidelines to name a topic or a subscription .

Create Managed Service for Apache Kafka resources

In this section, you create the following Managed Service for Apache Kafka resources:

A Kafka cluster with a topic.
A Connect cluster with a Pub/Sub connector.

Create a Kafka cluster

In this step, you create a Managed Service for Apache Kafka cluster . Creating a cluster can take up to 30 minutes.

Console

Go to the Managed Service for Apache Kafka > Clusterspage.

Go to Clusters
Click Create.
In the Cluster namebox, enter a name for the cluster.
In the Regionlist, select a location for the cluster.
For Network configuration, configure the subnet where the cluster is accessible:
1. For Project, select your project.
2. For Network, select the VPC network.
3. For Subnet, select the subnet.
4. Click Done.
Click Create.

After you click Create, the cluster state is Creating . When the cluster is ready, the state is Active .

gcloud

To create a Kafka cluster, run the managed-kafka clusters create command.

 gcloud  
managed-kafka  
clusters  
create  
 KAFKA_CLUSTER 
  
 \ 
--location = 
 REGION 
  
 \ 
--cpu = 
 3 
  
 \ 
--memory = 
3GiB  
 \ 
--subnets = 
projects/ PROJECT_ID 
/regions/ REGION 
/subnetworks/ SUBNET_NAME 
  
 \ 
--async

Replace the following:

KAFKA_CLUSTER : a name for the Kafka cluster
REGION : the location of the cluster
PROJECT_ID : your project ID
SUBNET_NAME : the subnet where you want to create the cluster, for example default

For information about supported locations, see Managed Service for Apache Kafka locations .

The command runs asynchronously and returns an operation ID:

 Check  
operation  
 [ 
projects/ PROJECT_ID 
/locations/ REGION 
/operations/ OPERATION_ID 
 ] 
  
 for 
  
status.

To track the progress of the create operation, use the gcloud managed-kafka operations describe command:

 gcloud  
managed-kafka  
operations  
describe  
 OPERATION_ID 
  
 \ 
  
--location = 
 REGION

When the cluster is ready, the output from this command includes the entry state: ACTIVE . For more information, see Monitor the cluster creation operation .

Create a Kafka topic

After the Managed Service for Apache Kafka cluster is created, create a Kafka topic.

Console

Go to the Managed Service for Apache Kafka > Clusterspage.

Go to Clusters
Click the name of the cluster.
In the cluster details page, click Create Topic.
In the Topic namebox, enter a name for the topic.
Click Create.

gcloud

To create a Kafka topic, run the managed-kafka topics create command.

 gcloud  
managed-kafka  
topics  
create  
 KAFKA_TOPIC_NAME 
  
 \ 
--cluster = 
 KAFKA_CLUSTER 
  
 \ 
--location = 
 REGION 
  
 \ 
--partitions = 
 10 
  
 \ 
--replication-factor = 
 3

Replace the following:

KAFKA_TOPIC_NAME : the name of the Kafka topic to create
KAFKA_CLUSTER : the name of the Kafka cluster
REGION : the region where you created the Kafka cluster

Create a Connect cluster

In this step, you create a Connect cluster. Creating a Connect cluster can take up to 30 minutes.

Before you start this step, make sure the Managed Service for Apache Kafka cluster is fully created.

Console

Go to the Managed Service for Apache Kafka > Connect Clusterspage.

Go to Connect Clusters
Click Create.
For the Connect cluster name, enter a string. Example: my-connect-cluster .
For Primary Kafka cluster, select the Kafka that you created earlier.
Click Create.

While the cluster is being created, the cluster state is Creating . When the cluster has finished being created, the state is Active .

gcloud

To create a Connect cluster, run the gcloud alpha managed-kafka connect-clusters create command.

 gcloud  
alpha  
managed-kafka  
connect-clusters  
create  
 CONNECT_CLUSTER 
  
 \ 
  
--location = 
 REGION 
  
 \ 
  
--cpu = 
 12 
  
 \ 
  
--memory = 
12GiB  
 \ 
  
--primary-subnet = 
projects/ PROJECT_ID 
/regions/ REGION 
/subnetworks/ SUBNET_NAME 
  
 \ 
  
--kafka-cluster = 
 KAFKA_CLUSTER 
  
 \ 
  
--async

Replace the following:

CONNECT_CLUSTER : a name for the Connect cluster
REGION : the region where you created the Kafka cluster
PROJECT_ID : your project ID
SUBNET_NAME : the subnet where you created the Kafka cluster
KAFKA_CLUSTER : the name of your Kafka cluster

The command runs asynchronously and returns an operation ID:

 Check  
operation  
 [ 
projects/ PROJECT_ID 
/locations/ REGION 
/operations/ OPERATION_ID 
 ] 
  
 for 
  
status.

To track the progress of the create operation, use the gcloud managed-kafka operations describe command:

 gcloud  
managed-kafka  
operations  
describe  
 OPERATION_ID 
  
 \ 
  
--location = 
 REGION

For more information, see Monitor the cluster creation operation .

Grant IAM roles

Grant the following Identity and Access Management (IAM) roles to the Managed Kafka service account:

Pub/Sub Subscriber
Pub/Sub Viewer

These roles allow connectors to read messages from Pub/Sub.

Console

In the Google Cloud console, go to the IAMpage.

Go to IAM
Select Include Google-provided role grants.
Find the row for Managed Kafka Service Accountand click Edit principal.
Click Add another roleand select the role Pub/Sub Subscriber. Repeat this step for the Pub/Sub Viewerrole.
Click Save.

For more information about granting roles, see Grant an IAM role by using the console .

gcloud

To grant IAM roles to the service account, run the gcloud projects add-iam-policy-binding command.

 gcloud  
projects  
add-iam-policy-binding  
 PROJECT_ID 
  
 \ 
  
--member = 
serviceAccount:service- PROJECT_NUMBER 
@gcp-sa-managedkafka.iam.gserviceaccount.com  
 \ 
  
--role = 
roles/pubsub.subscriber

gcloud  
projects  
add-iam-policy-binding  
 PROJECT_ID 
  
 \ 
  
--member = 
serviceAccount:service- PROJECT_NUMBER 
@gcp-sa-managedkafka.iam.gserviceaccount.com  
 \ 
  
--role = 
roles/pubsub.viewer

Replace the following:

PROJECT_ID : your project ID
PROJECT_NUMBER : your project number

To find your project number, use the gcloud projects describe command.

Create a Pub/Sub Source connector

In this step, you create a Pub/Sub Source connector . This connector reads messages from Pub/Sub and writes them to a Kafka topic.

Console

Go to the Managed Service for Apache Kafka > Connect Clusterspage.

Go to Connect Clusters
Click the name of the Connect cluster.
Click Create connector.
For the Connector name, enter a string. Example: pubsub-source .
In the Connector pluginlist, select Pub/Sub Source .
For Cloud Pub/Sub subscription, select the default Pub/Sub that was created when you created the Pub/Sub topic.
For Kafka topic, select the Kafka topic that you created previously.
Click Create.

gcloud

To create a Pub/Sub Source connector, run the gcloud alpha managed-kafka connectors create command.

 gcloud  
alpha  
managed-kafka  
connectors  
create  
 PUBSUB_CONNECTOR_NAME 
  
 \ 
  
--connect_cluster = 
 CONNECT_CLUSTER 
  
 \ 
  
--location = 
 REGION 
  
 \ 
  
--configs = 
connector.class = 
com.google.pubsub.kafka.source.CloudPubSubSourceConnector, \ 
cps.project = 
 PROJECT_ID 
, \ 
cps.streamingPull.enabled = 
true, \ 
cps.subscription = 
 SUBSCRIPTION_ID 
, \ 
kafka.topic = 
 KAFKA_TOPIC_NAME 
, \ 
key.converter = 
org.apache.kafka.connect.storage.StringConverter, \ 
tasks.max = 
 3 
, \ 
value.converter = 
org.apache.kafka.connect.converters.ByteArrayConverter

Replace the following:

PUBSUB_CONNECTOR_NAME : a name for the connector, such as pubsub-source-connector
CONNECT_CLUSTER : the name of your Connect cluster
REGION : the region where you created the Connect cluster
PROJECT_ID : your project ID
KAFKA_TOPIC_NAME : the name of your Kafka topic
SUBSCRIPTION_ID : the name of your Pub/Sub subscription

View results

To view the results, publish some messages to Pub/Sub.

Console

In the Google Cloud console, go to the Pub/Sub > Topicspage.

Go to Topics
In the topic list, click the name of your Pub/Sub topic.
Click Messages.
Click Publish messages.
For Number of messages, enter 10 .
For Message body, enter {"name": "Alice", "customer_id": 1} .
Click Publish.

gcloud

To publish messages to your Pub/Sub topic, use the gcloud pubsub topics publish command.

  for 
  
run  
 in 
  
 { 
 1 
..10 } 
 ; 
  
 do 
  
gcloud  
pubsub  
topics  
publish  
 TOPIC_ID 
  
--message = 
 '{"name": "Alice", "customer_id": 1}' 
 done

Replace TOPIC_ID with the name of your Pub/Sub topic.

Now you can consume the messages from the Kafka topic. For more information, see Produce and consume messages with the CLI .

Clean up

To avoid incurring charges to your Google Cloud account for the resources used in this tutorial, either delete the project that contains the resources, or keep the project and delete the individual resources.

Console

Delete the Pub/Sub topic.
1. Go to the Pub/Sub > Topicspage.
  
  Go to Topics
2. Select the topic and click Delete.
Delete the Pub/Sub subscription.
1. Go to the Pub/Sub > Subscriptionspage.
  
  Go to Subscriptions
2. Select the subscription created with your topic and click Delete.
Delete the Connect cluster.
1. Go to the Managed Service for Apache Kafka > Connect Clusterspage.
  
  Go to Connect Clusters
2. Select the Connect cluster and click Delete.
Delete the Kafka cluster.
1. Go to the Managed Service for Apache Kafka > Clusterspage.
  
  Go to Clusters
2. Select the Kafka cluster and click Delete.

gcloud

To delete the Pub/Sub subscription and topic, use the gcloud pubsub subscriptions delete and the gcloud pubsub topics delete commands.

 gcloud  
pubsub  
subscriptions  
delete  
 SUBSCRIPTION_ID 
gcloud  
pubsub  
topics  
delete  
 TOPIC_ID

To delete the Connect cluster, use the gcloud alpha managed-kafka connect-clusters delete command.

 gcloud  
alpha  
managed-kafka  
connect-clusters  
delete  
 CONNECT_CLUSTER 
  
 \ 
  
--location = 
 REGION 
  
--async

To delete the Kafka cluster, use the gcloud managed-kafka clusters delete command.

 gcloud  
managed-kafka  
clusters  
delete  
 KAFKA_CLUSTER 
  
 \ 
  
--location = 
 REGION 
  
--async

What's next

Troubleshoot a Pub/Sub connector .
Learn more about the Pub/Sub Source connector .
Learn more about Kafka Connect .

Replicate Pub/Sub messages to Kafka Stay organized with collections Save and categorize content based on your preferences.

Before you begin

Console

Check for the roles

Grant the roles

gcloud

Create a Pub/Sub topic and subscription

Console

gcloud

Create Managed Service for Apache Kafka resources

Create a Kafka cluster

Console

gcloud

Create a Kafka topic

Console

gcloud

Create a Connect cluster

Console

gcloud

Grant IAM roles

Console

gcloud

Create a Pub/Sub Source connector

Console

gcloud

View results

Console

gcloud

Clean up

Console

gcloud

What's next

Replicate Pub/Sub messages to Kafka