# Google Cloud Pub/Sub Source Connector for Confluent Platform

The Kafka Connect Google Cloud Pub/Sub Source Connector reads messages from a Pub/Sub
topic and writes them to a Kafka topic. The schema for the Kafka record key and
value is described in the section [Record Schema](#pubsub-record-schema). The Pub/Sub
Connector uses [pull strategy](https://cloud.google.com/pubsub/docs/pull) to
get messages from the Pub/Sub topic. The messages are pulled [synchronously](https://cloud.google.com/pubsub/docs/pull).

#### IMPORTANT
<!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the
[Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8)
page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and
are no longer available for download.

## Features

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

- [At least once delivery](#google-pubsub-at-least-once-delivery)
- [Multiple tasks](#google-pubsub-multiple-tasks)
- [No ordering guarantees](#google-pubsub-no-ordering-guarantees)
- [Fetch Multiple Messages](#google-pubsub-fetch-mulitple-messages)
- [Client-side encryption](#google-pubsub-sink-csfle-sm)

<a id="google-pubsub-at-least-once-delivery"></a>

### At least once delivery

This connector guarantees that records are delivered to the Kafka topic at least
once. If the connector restarts, there may be some duplicate records in the Kafka
topic.

<a id="google-pubsub-multiple-tasks"></a>

### Multiple tasks

The Google Cloud Pub/Sub Source connector supports running one or more tasks. You can
specify the number of tasks in the `tasks.max` configuration parameter. This
can lead to performance gains when multiple files need to be parsed.

<a id="google-pubsub-no-ordering-guarantees"></a>

### 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](https://cloud.google.com/pubsub/docs/ordering). It is possible that the
records written to a Kafka topic end up in a different order.

<a id="google-pubsub-fetch-mulitple-messages"></a>

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

<a id="google-pubsub-sink-csfle-sm"></a>

### Client-side encryption

This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html).

## License

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

After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information.

See [Confluent Platform license](configuration_options.md#gcp-pubsub-connector-license-config)  and [Confluent License Properties](configuration_options.md#gcp-pubsub-license-topic-configuration) for information about the license topic.

## Configuration Properties

For a complete list of configuration properties for this connector, see [Configuration Reference for Google Cloud Pub/Sub Source Connector for Confluent Platform](configuration_options.md#pubsub-source-connector-config).

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

## Install the Google Cloud Pub/Sub Connector

You can install this connector by using the [confluent connect
plugin
install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html)
command, or by manually downloading the ZIP file.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- Kafka Broker: Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later.
- Connect: Confluent Platform 4.0.0 or later, or Kafka 1.0.0 or later.
- 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](https://console.cloud.google.com/iam-admin).
- You must grant the Google Cloud Pub/Sub Source connector the
  `roles/pubsub.subscriber` role in Google Cloud for it to start. For more
  information, see the [list of Pub/Sub IAM roles](https://cloud.google.com/pubsub/docs/access-control#roles).
- An installation of the latest (`latest`) connector version.

  To install the `latest` connector version, navigate to your Confluent Platform
  installation directory and run the following command:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-gcp-pubsub:latest
  ```

  You can install a specific version by replacing `latest` with a version
  number as shown in the following example:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-gcp-pubsub:1.2.0
  ```

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-gcp-pubsub) for your connector and then follow the manual connector installation [instructions](/kafka-connectors/self-managed/userguide.html#connect-installing-plugins).

## 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](/platform/current/quickstart/ce-quickstart.html#download-start-cp) before starting the steps below.

### Start Confluent

Start the Confluent services using the following [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) command:

```bash
confluent local start
```

#### IMPORTANT
Do not use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments.

### Set up Credentials

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

1. Open the **IAM & Admin** page in the GCP Console.
2. Select your project and click **Continue**.
3. In the left navigation panel, click **Service accounts**.
4. In the top toolbar, click **Create Service Account**.
5. Enter the service account name and description; for example `test-service-account`.
6. Click **Create** and on the next page select the role `Pub/Sub Subscriber` under **Pub/Sub**.
7. On the next page click **Create Key** and download the JSON file.
8. 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](https://cloud.google.com/dataproc/docs/concepts/configuring-clusters/service-accounts).

### 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](https://cloud.google.com/pubsub/docs/quickstart-cli).

Create a Pub/Sub topic called `topic-1`

```bash
gcloud pubsub topics create topic-1
```

Create a Pub/Sub subscription called `subscription-1`

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

Publish three messages to `topic-1`

```bash
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](/platform/current/connect/concepts.html#standalone-workers).

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

```bash
confluent local load pubsub-source --config pubsub-source.properties
```

Confirm that the connector is in a `RUNNING` state.

```bash
confluent local status pubsub-source
```

### REST-based example

Use this setting with [distributed workers](/platform/current/connect/concepts.html#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](/platform/current/connect/references/restapi.html)

```json
{
    "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).

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

```bash
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:

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

<a id="pubsub-record-schema"></a>

### Record Schema

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

1. The Kafka key consists of `project Id`, `message Id` and the
   `subscription Id` of the Pub/Sub message.
2. 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](https://console.cloud.google.com/cloudpubsub/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](https://cloud.google.com/sdk/gcloud/reference/pubsub/subscriptions/modify-message-ack-deadline).

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