<a id="connect-spanner"></a>

# Google Cloud Spanner Sink Connector for Confluent Platform

The Google Cloud Spanner Sink connector moves data from Apache Kafka® to a Google
Cloud Spanner database. It writes data from a topic in Kafka to a table in the
specified Spanner database. Table auto-creation and limited auto-evolution are
also supported.

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

- [At least once delivery](#spanner-sink-at-least-once-delivery)
- [Dead Letter Queue](#spanner-sink-dead-letter-queue)
- [Multiple tasks](#spanner-sink-multiple-tasks)
- [Data mapping](#spanner-sink-data-mapping)
- [Data types](#spanner-sink-data-types)
- [Auto-creation and auto-evolution](#spanner-sink-auto-creation)
- [PostgreSQL dialect support](#spanner-sink-postgres-dialect)
- [Client-side encryption](#spanner-sink-csfle)

<a id="spanner-sink-at-least-once-delivery"></a>

### At least once delivery

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

<a id="spanner-sink-dead-letter-queue"></a>

### Dead Letter Queue

This connector supports the Dead Letter Queue (DLQ) functionality. For
information about accessing and using the DLQ, see [Confluent Platform
Dead Letter Queue](/platform/current/connect/concepts.html#dead-letter-queue).

<a id="spanner-sink-multiple-tasks"></a>

### Multiple tasks

The Google Cloud Spanner Sink 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="spanner-sink-data-mapping"></a>

### Data mapping

The sink connector requires knowledge of schemas, so you should use a suitable
converter, like the Avro converter that comes with Schema Registry or the JSON converter
with schemas enabled. Kafka record keys, if present, can be primitive types or a
Connect struct, and the record value must be a Connect struct. You may need to
implement a custom `Converter` if the data in the topic is not in a compatible
format.

<a id="spanner-sink-data-types"></a>

### Data types

Mapping from Kafka record types to [Spanner data types](https://cloud.google.com/spanner/docs/data-types) is detailed below.

| Kafka Record Type                                     | Spanner Type       |
|-------------------------------------------------------|--------------------|
| INT8, INT16, INT32, INT64                             | INT64              |
| FLOAT32, FLOAT64                                      | FLOAT64            |
| BOOLEAN                                               | BOOL               |
| STRING                                                | STRING(MAX)        |
| BYTES                                                 | BYTES(MAX)         |
| DATE                                                  | DATE               |
| TIMESTAMP                                             | TIMESTAMP          |
| ARRAY<INT8>, ARRAY<INT16>, ARRAY<INT32>, ARRAY<INT64> | ARRAY<INT64>       |
| ARRAY<FLOAT32>, ARRAY<FLOAT64>                        | ARRAY<FLOAT64>     |
| ARRAY<BOOLEAN>                                        | ARRAY<BOOL>        |
| ARRAY<STRING>                                         | ARRAY<STRING(MAX)> |
| NOT SUPPORTED                                         | ARRAY<BYTES(MAX)>  |
| NOT SUPPORTED                                         | ARRAY<DATE>        |
| NOT SUPPORTED                                         | ARRAY<TIMESTAMP>   |
| MAP                                                   | NOT SUPPORTED      |
| STRUCT                                                | NOT SUPPORTED      |

<a id="spanner-sink-auto-creation"></a>

### Auto-creation and auto-evolution

If `auto.create` is enabled, the connector can create the destination table if
it is found to be missing. The connector uses the record schema as a basis for
the table definition, so the creation takes place online with records consumed
from the topic.

If `auto.evolve` is enabled, the connector can perform limited auto-evolution
when it encounters a record for which a column is found to be missing. Since
data-type changes and removal of columns can be dangerous, the connector does
not attempt to perform such evolutions on the table.

#### IMPORTANT
For backward-compatible table schema evolution, new fields in record schemas
must be optional or have a default value.

<a id="spanner-sink-postgres-dialect"></a>

### PostgreSQL dialect support

The connector supports Google Cloud Spanner databases that are created with
the [PostgreSQL dialect](https://cloud.google.com/spanner/docs/reference/postgresql/overview). To
enable this feature, the connector includes a new configuration property, `gcp.spanner.database.dialect`.
For more information, see [PostgreSQL Dialect](configuration_options.md#spanner-connector-postgres-dialect-configs)
configuration property.

<a id="spanner-sink-csfle"></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

- Cloud Spanner has [table size](https://cloud.google.com/spanner/quotas#tables) and [query limitations](https://cloud.google.com/spanner/quotas#query_limits) that apply to the
  connector.
- The [performance limitations](https://cloud.google.com/spanner/docs/instances#regional-performance) of
  instances apply to the connector.
- The connector does not support creation of [interleaved tables](https://cloud.google.com/spanner/docs/schema-and-data-model#creating-interleaved-tables).
- When `auto.evolve` is enabled, if a new column with a default value is
  added, that default value is only used for new records. Existing records will
  have `<NULL>` as the value for the new column.
- 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`

## 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#cloud-spanner-sink-connector-license-config) for license properties
and [Confluent License Properties](configuration_options.md#spanner-connector-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 Spanner Sink Connector for Confluent Platform](configuration_options.md#spanner-connector-configuration-options).

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 Cloud Spanner 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.
- Confluent Platform 4.0.0 or later, or Kafka 1.0.0 or later. For installation help, see
  [Confluent Platform](/platform/current/installation/index.html).
- [Confluent CLI](https://docs.confluent.io/confluent-cli/current/installing.html) (requires separate installation).
- [Google Cloud Platform (GCP) Account](https://cloud.google.com/).
- A GCP project and billing enabled. For instructions, see the [Cloud Spanner
  Quickstart](https://cloud.google.com/spanner/docs/quickstart-console#before-you-begin).
- Java 1.8.
- At minimum, `roles/spanner.databaseUser` is required for this connector.
  For details, see [Access control for Cloud Spanner](https://cloud.google.com/spanner/docs/iam).
- 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-spanner: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-spanner:1.2.0
  ```

### Install the connector manually

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

<a id="spanner-connector-quickstart"></a>

## Quick Start

In this quick start, the Spanner sink connector is used to export data produced
by the Avro console producer to a database in a Spanner instance.

### Setup 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 nav, 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 `Cloud Spanner Database Admin` under `Cloud Spanner`.
7. On the next page click **Create Key** and download the JSON file.
8. For this quick start, save the file under your `$home` directory and name it `spanner-test-credentials.json`.

For more information on service account keys, see [Creating and managing service
account keys](https://cloud.google.com/iam/docs/creating-managing-service-account-keys#iam-service-account-keys-create-console).

### Create a Spanner Instance and Database

1. Create a test instance named `test-instance` in Spanner using the console.
   See the [Cloud Spanner Quickstart](https://cloud.google.com/spanner/docs/quickstart-console#create_an_instance)
   for instructions.
2. Create a database named `example-db` under the `test-instance`.
   See the [Cloud Spanner Quickstart](https://cloud.google.com/spanner/docs/quickstart-console#create_a_database)
   for instructions.

### Install and load the connector

1. Install the connector using the following [CLI
   command](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html):
   ```bash
   # run from your CP installation directory
   confluent connect plugin install confluentinc/kafka-connect-gcp-spanner:latest
   ```
2. Adding a new connector plugin requires restarting Kafka Connect. Use the
   following command to restart Connect.
   ```bash
   confluent local services connect stop && confluent local services connect start
   ```
3. Configure your connector by adding the file `spanner-sink.properties`,
   with the following properties:
   ```text
   name=SpannerSinkConnector
   topics=products
   tasks.max=1
   connector.class=io.confluent.connect.gcp.spanner.SpannerSinkConnector

   gcp.spanner.credentials.path=$home/spanner-test-credentials.json
   gcp.spanner.instance.id=test-instance
   gcp.spanner.database.id=example-db
   auto.create=true
   table.name.format=kafka_${topic}

   # The following define the Confluent license stored in Kafka, so we need the Kafka bootstrap addresses.
   # `replication.factor` may not be larger than the number of Kafka brokers in the destination cluster,
   # so here we set this to '1' for demonstration purposes. Always use at least '3' in production configurations.
   confluent.license=
   confluent.topic.bootstrap.servers=localhost:9092
   confluent.topic.replication.factor=1
   ```

   #### NOTE
   Ensure to replace the `$home` with your home directory path, or any
   other path where the credentials file was saved.
4. Start the Spanner Sink connector by loading the connector’s configuration
   with the following command:
   ```bash
   confluent local load spanner --config spanner-sink.properties
   ```

   Your output should resemble the following:
   ```json
   {
     "name": "spanner",
     "config": {
       "topics": "products",
       "tasks.max": "1",
       "connector.class": "io.confluent.connect.gcp.spanner.SpannerSinkConnector",
       "gcp.spanner.credentials.path": "$home/spanner-test-credentials.json",
       "gcp.spanner.instance.id": "test-instance",
       "gcp.spanner.database.id": "example-db",
       "auto.create": "true",
       "table.name.format": "kafka_${topic}",
       "confluent.license": "",
       "confluent.topic.bootstrap.servers": "localhost:9092",
       "confluent.topic.replication.factor": "1",
       "name": "spanner"
     },
     "tasks": [
       {
         "connector": "spanner",
         "task": 0
       }
     ],
     "type": "sink"
   }
   ```
5. Check the status of the connector to confirm that it is in a `RUNNING`
   state.
   ```bash
   confluent local status spanner
   ```

   Your output should resemble the following:
   ```bash
   {
     "name": "spanner",
     "connector": {
       "state": "RUNNING",
       "worker_id": "10.200.7.192:8083"
     },
     "tasks": [
       {
         "id": 0,
         "state": "RUNNING",
         "worker_id": "10.200.7.192:8083"
       }
     ],
     "type": "sink"
   }
   ```

### Send data to Kafka

1. To produce some records into the `products` topic, first start a Kafka
   producer.
   ```bash
   kafka-avro-console-producer \
   --broker-list localhost:9092 --topic products \
   --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"name","type":"string"},
   {"name":"price", "type": "float"}, {"name":"quantity", "type": "int"}]}'
   ```
2. The console producer is now waiting for input, so you can go ahead and
   insert some records into the topic.
   ```json
   {"name": "scissors", "price": 2.75, "quantity": 3}
   {"name": "tape", "price": 0.99, "quantity": 10}
   {"name": "notebooks", "price": 1.99, "quantity": 5}
   ```

### Check Spanner database for data

To verify that the data has been written to Spanner, you can use the GCP console
([https://console.cloud.google.com/spanner/instances/test-instance/databases](https://console.cloud.google.com/spanner/instances/test-instance/databases)).

Under databases you should see the `kafka_products` table. Clicking on the
table and then `DATA` will show the three rows the have just been inserted.

| `connect_topic__`   |   `connect_partition__` |   `connect_offset__` |   quantity |   price | name      |
|---------------------|-------------------------|----------------------|------------|---------|-----------|
| products            |                       0 |                    0 |          3 |    2.75 | scissors  |
| products            |                       0 |                    1 |         10 |    0.99 | tape      |
| products            |                       0 |                    2 |          5 |    1.99 | notebooks |

### Deleting unnecessary resources

Delete the test instance.

1. Click `test-instance` on the left sidebar.
2. Click **Delete Instance** on the top toolbar and type the instance name to
   verify deletion.

Delete the service account credentials used for the test.

1. Open the **IAM & Admin** page in the GCP Console.
2. Select your project and click **Continue**.
3. In the left nav, click **Service accounts**.
4. Locate the `test-service-account` and click the **More** button under **Actions**.
5. Click **Delete** and confirm deletion.

## Troubleshooting Connector and Task Failures

You can use the Connect [REST API](/platform/current/connect/references/restapi.html)
to check the status of the connectors and tasks. If a task or connector has
failed, the `trace` field will include a reason and a stack trace.

### Authorization failures

The Cloud Spanner connector must authenticate with a Spanner instance and
establish a connection. If a connection fails because of authentication, the
connector will stop immediately. These errors may require changes in your Google
Cloud account which may include creating service account keys. Try to rerun your
connector after you make the account changes. More information on service
account keys, see [Creating and managing service account credentials](https://cloud.google.com/iam/docs/creating-managing-service-account-keys#iam-service-account-keys-create-console).

### Spanner error codes

Whenever the connector or task fails, it captures a message that includes the
Spanner error code and error message. The message may also include a failure
reason or suggested fix. See the [Spanner error codes](https://cloud.google.com/spanner/docs/reference/rest/v1/Code) for error code
details.

### Repeated connection failures

If the connector stops frequently because of connection timeouts, consider
changing the following connector configuration properties and restarting the
connector:

* The `​request.timeout.ms` defaults to 6 seconds (6000 milliseconds) and
  dictates the maximum amount of time that the connector should wait for a
  single spanner operation to succeed. If this time is exceeded the connector
  will attempt to retry.
* The `​retry.timeout.ms` defaults to 1 minutes (60000 milliseconds) and
  specifies the maximum amount of time that the connector should continue to retry
  a failed request. The actual duration of the delay is random and it grows
  exponentially with each retry. If all retries take longer than this value, the
  connector task will fail. This does not apply to initial connection attempts,
  but it does apply to subsequent requests to reconnect. Only retry-able errors
  from spanner will be retried.

### Quota failures

If the connector fails due to write limits being exceeded, consider changing the
`max.batch.size` configuration property and restarting the connector. The
`max.batch.size` defaults to 1000 and specifies the maximum amount of rows to
batch together for a Spanner operation (INSERT, UPDATE, UPSERT). If the row
sizes are large, consider lowering the max batch size.

### Connector error mode

By default, the `error.mode` for the connector is `FAIL`. This means if
there is an error when writing a batch of records to Spanner the connector
fails. It may be convenient in some cases to set the `error.mode` to `WARN`
or `INFO` instead so that the connector keeps running even after a particular
batch fails.

### Proxy settings

When the `gcp.spanner.proxy.url` proxy settings are configured, the system
property variables (`https.proxyHost` and `https.proxyPort`) are set
globally for the entire JVM.

### Enabling debug logging

The Connect worker log configuration controls how much detail is included in
the logs. By default, the worker logs include enough detail to identify basic
functionality. Enable DEBUG logs in the Connect worker’s log configuration to
include more details. This change must be made on each worker and only takes
affect upon worker startup. After you change the log configuration as outlined
below on each Connect worker, restart all of the Connect workers. A
rolling restart can be used if necessary.

#### NOTE
Trace level logging is verbose and contains many more details, and may be
useful to solve certain failures. Trace level logging is enabled like debug
level logging is enabled, except `TRACE` is used instead of `DEBUG`.

### On-premises installation

For local or on-premises installations of Confluent Platform, the
`etc/kafka/connect-log4j.properties` file defines the logging configuration of
the Connect worker process. To enable DEBUG on just the Spanner connector,
modify the `etc/kafka/connect-log4j.properties` file to include the following
line:

```text
log4j.logger.io.confluent.gcp.spanner=DEBUG
```

To enable DEBUG on all of the Connect worker’s code, including all
connectors, change the `log4j.rootLogger=` line to use `DEBUG` instead of
`INFO`. For example, the default log configuration for Connect includes
this line:

```text
log4j.rootLogger=INFO, stdout
```

Change this line to the following to enable DEBUG on all of the Connect
worker code:

```text
log4j.rootLogger=DEBUG, stdout
```

#### NOTE
This setting causes may generate a large amount of logs from
`org.apache.kafka.clients` packages, which can be suppressed by setting
`log4j.logger.org.apache.kafka.clients=ERROR`.