Important

You are viewing documentation for an older version of Confluent Platform. For the latest, click here.

Google Cloud Pub/Sub Source Connector for Confluent Platform¶

The Kafka Connect Google Cloud Pub/Sub source connector reads messages from Pub/Sub topic and writes them to Kafka topic. The schema for the Kafka record key and value is described in the section Record Schema. The Pub/Sub Connector uses pull strategy to get messages from the Pub/Sub topic. The messages are pulled synchronously.

Features¶

The Google Cloud Pub/Sub Source connector offers the following features:

Atleast Once Delivery: The connector guarantees that messages from Pub/Sub are delivered at least once to the Kafka topic.
No Ordering Guarantees: Google Cloud Pub/Sub provides a highly-available, scalable message delivery service. The tradeoff for having these properties is that the order in which messages are received by subscribers is not guaranteed. It is possible that the records written to Kafka topic end up in a different order.
Fetch Multiple Messages In every poll cycle, the connector fetches gcp.pubsub.message.max.count number of messages. By default, this value is 10000. However, if your message size is exceptionally large, you may want to reduce this to a lower number.

Note

While creating the subscription for the Pub/Sub topic, the acknowledgement deadline should be set to a high enough value to avoid duplicating records in Kafka topic. This allows the connector to commit the records and send acknowledgement for each Pub/Sub message processed. In the case when connector fails to write records to Kafka topic, the messages in the Pub/Sub topic are again made available.

Prerequisites¶

The following are required to run the Kafka Connect Google Cloud Pub/Sub Source Connector:

Kafka Broker: Confluent Platform 3.3.0 or above, or Kafka 0.11.0 or above
Connect: Confluent Platform 4.0.0 or above, or Kafka 1.0.0 or above
Java 1.8
A GCP service account that can access the Pub/Sub project. You can create this service account in the Google Cloud Console.

Install the Google Cloud Pub/Sub Connector¶

You can install this connector by using the Confluent Hub client (recommended) or you can manually download the ZIP file.

Install the connector using Confluent Hub¶

Prerequisite: Confluent Hub Client must be installed. This is installed by default with Confluent Enterprise.

Navigate to your Confluent Platform installation directory and run the following command to install the latest (latest) connector version. The connector must be installed on every machine where Connect will run.

confluent-hub install confluentinc/kafka-connect-gcp-pubsub:latest

You can install a specific version by replacing latest with a version number. For example:

confluent-hub install confluentinc/kafka-connect-gcp-pubsub:1.0.0-preview

Install the connector manually¶

Download and extract the ZIP file for your connector and then follow the manual connector installation instructions.

License¶

You can use this connector for a 30-day trial period without a license key.

After 30 days, this connector is available under a Confluent enterprise license. Confluent issues enterprise license keys to subscribers, along with providing enterprise-level support for Confluent Platform and your connectors. If you are a subscriber, please contact Confluent Support at support@confluent.io for more information.

See Confluent Platform license for license properties and License topic configuration for information about the license topic.

Configuration Properties¶

For a complete list of configuration properties for this connector, see Google Cloud Pub/Sub Source Connector Configuration Properties.

Note

For an example of how to get Kafka Connect connected to Confluent Cloud, see Distributed Cluster in Connect Kafka Connect to Confluent Cloud.

Quick Start¶

This quick start uses the Google Cloud Pub/Sub Source Connector to read messages from Pub/Sub and write them to a Kafka topic. You must download the Confluent Platform before starting the steps below.

Start Confluent¶

Start the Confluent services using the following Confluent CLI command:

confluent local start

Important

Do not use the Confluent CLI in production environments.

Set up Credentials¶

Create a service account and service account key under the GCP project.

Open the IAM & Admin page in the GCP Console.
Select your project and click Continue.
In the left navigation panel, click Service accounts.
In the top toolbar, click Create Service Account.
Enter the service account name and description; for example test-service-account.
Click Create and on the next page select the role Pub/Sub Subscriber under Pub/Sub.
On the next page click Create Key and download the JSON file.
For this quickstart, save the file under your $home directory and name it credentials.json.

For more information on service account keys, see the Google documentation.

Publish messages to Pub/Sub topic¶

To publish messages to Pub/Sub topic, a topic and a subscription should be created. Make sure you have the correct permission to create a Pub/Sub topic and a Pub/Sub subscription. You need to initialize your Google Cloud CLI with gcloud init if you haven’t used the gcloud CLI. Review Quickstart using the gcloud command-line tool.

Create a Pub/Sub topic called topic-1

gcloud pubsub topics create topic-1

Create a Pub/Sub subscription called subscription-1

gcloud pubsub subscriptions create --topic topic-1 subscription-1 --ack-deadline 20

Publish three messages to topic-1

gcloud pubsub topics publish topic-1 --message "Peter"
gcloud pubsub topics publish topic-1 --message "Megan"
gcloud pubsub topics publish topic-1 --message "Erin"

Property-based example¶

Create a pubsub-source.properties file with the following contents. This configuration is used typically along with standalone workers.

name=pubsub-source
connector.class=io.confluent.connect.gcp.pubsub.PubSubSourceConnector
tasks.max=1
kafka.topic=pubsub-topic

gcp.pubsub.project.id=project-1
gcp.pubsub.topic.id=topic-1
gcp.pubsub.subscription.id=subscription-1
gcp.pubsub.credentials.path=/home/some_directory/credentials.json

confluent.topic.bootstrap.servers=localhost:9092
confluent.topic.replication.factor=1

Run the connector with this configuration.

confluent local load pubsub-source -- -d pubsub-source.properties

Confirm that the connector is in a RUNNING state.

confluent local status pubsub-source

REST-based example¶

Use this setting with distributed workers. Write the following JSON to pubsub-source-source.json, configure all of the required values, and use the following command to post the configuration to one of the distributed connect workers. Check here for more information about the Kafka Connect REST API

{
    "name" : "pubsub-source",
    "config" : {
       "connector.class" : "io.confluent.connect.gcp.pubsub.PubSubSourceConnector",
       "tasks.max" : "1",
       "kafka.topic" : "pubsub-topic",
       "gcp.pubsub.project.id" : "project-1",
       "gcp.pubsub.topic.id" : "topic-1",
       "gcp.pubsub.subscription.id" : "subscription-1",
       "gcp.pubsub.credentials.path" : "/home/some_directory/credentials.json",
       "confluent.topic.bootstrap.servers" : "localhost:9092",
       "confluent.topic.replication.factor" : "1"
    }
}

Use curl to post the configuration to one of the Kafka Connect Workers. Change http://localhost:8083/ to the endpoint of one of your Kafka Connect worker(s).

curl -s -X POST -H 'Content-Type: application/json' --data @pubsub-source.json http://localhost:8083/connectors

Use the following command to update the configuration of existing connector.

curl -s -X PUT -H 'Content-Type: application/json' --data @pubsub-source.json http://localhost:8083/connectors/pubsub-source/config

To consume records written by connector to the configured Kafka topic, run the following command:

kafka-avro-console-consumer --bootstrap-server localhost:9092 --property schema.registry.url=http://localhost:8081  --topic pubsub-topic --from-beginning

Record Schema¶

Each Pub/Sub message is converted into exactly one Kafka record, with the following structure:

The Kafka key consists of project Id, message Id and the subscription Id of the Pub/Sub message.
Optionally, the attributes from attribute map of Pub/Sub message can also be made a part of the Kafka key.
The Kafka value consists of message data and message attributes of the Pub/Sub message.

Key Schema¶

The Key is a struct with the following fields:

Field Name	Schema Type	Description
`ProjectId`	String	The Pub/Sub project containing topic from which messages have to be polled.
`TopicId`	String	The Pub/Sub topic containing messages.
`SubscriptionId`	String	The Pub/Sub subscrpition of the Pub/Sub topic.
`MessageId`	String	A unique id for a Pub/Sub message

Value Schema¶

The Value is a struct with the following fields:

Field Name	Schema Type	Description
`MessageData`	Optional String	The body of the Pub/Sub message.
`AttributeMap`	Optional String	The attribute map associated with the Pub/Sub message.

Troubleshooting¶

Too many duplicates¶

When you have too many duplicate messages in the destination Kafka topic, increase the Acknowledgement deadline for the subscription. Go to Google Cloud Pub/Sub subscription, select the subscription used in this connector, and increase Acknowledgement deadline (for example, from 10 to 20 seconds). Or, you can use the following CLI command. For more information on GCP Pub/Sub CLI, see the Google documentation.

gcloud pubsub subscriptions modify-message-ack-deadline <subscription> --ack-deadline <seconds>