<a id="connect-gcp-functions-connector"></a>

# Google Cloud Functions Sink Connector for Confluent Platform

The Kafka Connect Google Cloud Functions Sink connector integrates Apache Kafka®
with Google Cloud Functions. The connector consumes records from Kafka topic(s)
and executes a Google Cloud Function. Each request sent to Google Cloud
Functions can contain up to `max.batch.size` records. The connector can also
send many requests concurrently, set by `max.pending.requests`.

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

The Connector also supports invoking authenticated functions–that is, functions
with private access. The user must provide service user credentials using either
the `gcf.credentials.json` or `gcf.credentials.path` configuration options
to use this functionality. The service user, whose credentials are provided,
must also have a minimum of `cloudfunctions.functions.get` and
`cloudfunctions.functions.invoke` IAM roles assigned to it. If the user
doesn’t provide the required credentials, it’s assumed that the Google Cloud
function can be invoked without any authentication.

The target function must be configured and ready to accept requests with the
following JSON format:

```bash
[
  {
    "key": ...,
    "value": ...,
    "topic": string,
    "partition": <number>,
    "offset": <number>,
    "timestamp": <number>
  },
  ...,
]
```

#### NOTE
The key and value are encoded as follows:

* `String`, `int`, `long`, `float`, `double`, `boolean`, `null` are encoded as-is into JSON.
* `Structs` are converted to JSON and exported *without* the schema.
* `byte[]` is encoded as a base64 `String` and sent as a JSON string.
* Any other Java objects are converted to `String` using `toString()`, and then sent as JSON strings.

The connector receives the responses from the Google Cloud Function and writes
it to a result or error topic (set by configurations) depending on the HTTP
Response code. Response code `400` and above are considered errors and
anything below is a success.

The connector attempts to map each response to a single record before producing
it to the corresponding topic. It can receive the responses from the Google
Cloud Function in the following three formats.

1. The first format is JSON:
   ```bash
   [
     {
       "payload": {
         "result": ...,
         "topic": string,
         "partition": <number>,
         "offset": <number>,
       }
     },
     ...
   ]
   ```

   This list can be out of order relative to the order that the records were
   provided. The connector will correctly match the record to the result based
   off its Kafka coordinates. However the list must be one-to-one to the list of
   records that were sent in the request.
2. The second format is a JSON list:
   ```bash
   [
     ...,
     ...,
     ...
   ]
   ```

   As long as the list is one-to-one to the list of records, the list will be
   assumed to be ordered and matched with the corresponding records.
3. The third format can be any format that does not satisfy either of the
   above formats. The connector will report the entire response for each individual
   record (one-to-many correlation).

## Features

### At least once delivery

This connector guarantees that records from the Kafka topic are delivered at
least once.

<a id="gcp-functions-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).

## Limitations

The connector does not currently support [Google Cloud Functions (2nd gen)](https://cloud.google.com/functions/docs/concepts/version-comparison).

The connector does not currently support [Single Message Transformations (SMTs)](https://docs.confluent.io/platform/current/connect/transforms/overview.html)
that modify the topic name. Additionally, the following transformations are not
allowed:

- `io.debezium.transforms.ByLogicalTableRouter`
- `io.debezium.transforms.outbox.EventRouter`
- `org.apache.kafka.connect.transforms.RegexRouter`
- `org.apache.kafka.connect.transforms.TimestampRouter`
- `io.confluent.connect.transforms.MessageTimestampRouter`
- `io.confluent.connect.transforms.ExtractTopic$Key`
- `io.confluent.connect.transforms.ExtractTopic$Value`

## Install the Google Cloud Functions 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.
- 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-functions: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-functions:1.0.0-preview
  ```

### Install the connector manually

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

## 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](connector_config.md#gcp-functions-sink-connector-license-config) for license properties
and [Confluent License Properties](connector_config.md#gcp-functions-sink-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 Functions Sink Connector for Confluent Platform](connector_config.md#gcp-functions-sink-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).

## Quick Start

This quick start uses the Google Cloud Functions Sink connector to consume
records and send them to a Google Cloud Functions function.

Prerequisites
: - [Confluent Platform](/platform/current/installation/index.html)
  - [Confluent CLI](https://docs.confluent.io/confluent-cli/current/installing.html) (requires separate installation)

1. Before starting the connector, create and deploy a basic Google Cloud Functions instance.
   - Navigate to the [Google Cloud Console](https://console.cloud.google.com).
   - Go to the [Cloud Functions](https://console.cloud.google.com/functions) tab.
   - Create a new function.
   - For creating an unauthenticated function select **Allow unauthenticated
     invocations** and go ahead.
   - For authenticated functions select **Require Authentication** and then
     click Variables, Networking and Advanced Settings to display additional
     settings. Click the Service account drop down and select the desired service
     account.
   - Note down the project id, the region, and the function name as they will
     be used later.
   - Further, to add an invoker account for an already deployed function, click
     **Add members** in the Permission tab of the functions home page. In the
     popup, select add member and select *Cloud Functions Invoker* Role.
2. Install the connector by running the following command from your Confluent Platform
   installation directory:
   ```bash
   confluent connect plugin install confluentinc/kafka-connect-gcp-functions:latest
   ```
3. Start Confluent Platform.
   ```bash
   confluent local start
   ```
4. Produce test data to the `functions-messages` topic in Kafka using the CLI
   command below.
   ```bash
   echo key1,value1 | confluent local produce functions-messages --property parse.key=true --property key.separator=,
   echo key2,value2 | confluent local produce functions-messages --property parse.key=true --property key.separator=,
   echo key3,value3 | confluent local produce functions-messages --property parse.key=true --property key.separator=,
   ```
5. Create a `gcp-functions.json` file with the following contents:
   ```json
   {
     "name": "gcp-functions",
     "config": {
       "topics": "functions-messages",
       "tasks.max": "1",
       "connector.class": "io.confluent.connect.gcp.functions.GoogleCloudFunctionsSinkConnector",
       "key.converter":"org.apache.kafka.connect.storage.StringConverter",
       "value.converter":"org.apache.kafka.connect.storage.StringConverter",
       "confluent.topic.bootstrap.servers": "localhost:9092",
       "confluent.topic.replication.factor":1,
       "function.name": "<insert function name here>",
       "project.id": "<insert project id here>",
       "region": "<insert region here>",
       "gcf.credentials.path": "<insert path to service credentials JSON here>",
       "reporter.bootstrap.servers": "localhost:9092",
       "reporter.error.topic.name": "test-error",
       "reporter.error.topic.replication.factor": 1,
       "reporter.error.topic.key.format": "string",
       "reporter.error.topic.value.format": "string",
       "reporter.result.topic.name": "test-result",
       "reporter.result.topic.key.format": "string",
       "reporter.result.topic.value.format": "string",
       "reporter.result.topic.replication.factor": 1
     }
   }
   ```

   #### NOTE
   For details about using this connector with Kafka Connect Reporter, see
   [Connect
   Reporter](/kafka-connectors/self-managed/userguide.html#userguide-connect-reporter).
6. Load the Google Cloud Functions Sink connector.
   ```bash
   confluent local load gcp-functions --config gcp-functions.json
   ```

   #### IMPORTANT
   Don’t use the CLI commands in production environments.
7. Confirm that the connector is in a `RUNNING` state.
   ```bash
   confluent local status gcp-functions
   ```
8. Confirm that the messages were delivered to the result topic in Kafka
   ```bash
   confluent local consume test-result --from-beginning
   ```
9. Cleanup resources
   * Delete the connector
     ```bash
     confluent local unload gcp-functions
     ```
   * Stop Confluent Platform
     ```bash
     confluent local stop
     ```
   * Delete the created Google Cloud Function in the Google Cloud Platform
     portal.