Google Cloud Spanner CDC Source (Debezium) Connector for Confluent Cloud

The fully-managed Google Cloud Spanner Change Data Capture (CDC) Source (Debezium) connector for Confluent Cloud monitors a Google Cloud Spanner change stream and streams row-level inserts, updates, and deletes from Spanner tables to Apache Kafka® topics in near-real time.

If you require private networking for fully-managed connectors, make sure to set up the proper networking beforehand. For more information, see Manage Networking for Confluent Cloud Connectors.

Features

The Google Cloud Spanner CDC Source (Debezium) connector provides the following features:

Topics created automatically: Automatically creates Kafka topics using the naming convention <connectorID>.<tableName>. Per-row ordering is preserved regardless of topic partition count, because records are partitioned by the table’s primary key. For global ordering across all rows, use a single topic partition.
Tables included and Tables excluded: Sets whether a table is or is not monitored for changes. By default, the connector monitors every table tracked by the change stream.
Tombstones on delete: Configures whether a tombstone event should be generated after a delete event. Default is true.
Low watermark: A timestamp that guarantees all events with an earlier commit timestamp have been streamed to Kafka. Enable it when downstream consumers need to confirm event completeness.
Database role support: Supports Spanner fine-grained access control using database roles.
Output data formats: Supports Avro, JSON_SR (JSON Schema), and Protobuf output record value formats. Schema Registry must be enabled. For more information, see Limits for Fully-Managed Connectors for Confluent Cloud.
Client-side encryption (CSFLE and CSPE) support: Supports CSFLE and CSPE for sensitive data. For more information about CSFLE or CSPE setup, see Manage CSFLE or CSPE for connectors.

For more information and examples to use with the Confluent Cloud API for Connect, see the Confluent Cloud API for Connect Usage Examples section.

Limitations

For connector limitations, see Google Cloud Spanner CDC Source (Debezium) Connector limits.
If you plan to use one or more single message transformations (SMTs), see SMT Limitations.

If you plan to use one or more Custom SMTs, see Custom SMT limitations.

Quick start

Select the connector and configure it to stream change events from a Spanner change stream to Kafka.

Prerequisites

Authorized access to a Confluent Cloud cluster on Amazon Web Services (AWS), Microsoft Azure (Azure), or Google Cloud.
The Confluent CLI installed and configured for the cluster. See Install the Confluent CLI.
Schema Registry must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). For more information, see Limits for Fully-Managed Connectors for Confluent Cloud.
A Google Cloud project with a Spanner instance and database.
A Spanner change stream configured on the database. For details, see Create a change stream.
A Google Cloud service account with the following IAM roles on the Spanner instance:
- roles/spanner.databaseReader
- roles/spanner.viewer
The Google Cloud service account key in JSON format.
For networking considerations, see Networking and DNS. To use a set of public egress IP addresses, see Public Egress IP Addresses for Confluent Cloud Connectors.

Kafka cluster credentials. The following lists the different ways you can provide credentials.
- Enter an existing service account resource ID.
- Create a Confluent Cloud service account for the connector. Make sure to review the ACL entries required in the service account documentation. Some connectors have specific ACL requirements.
- Create a Confluent Cloud API key and secret. To create a key and secret, you can use confluent api-key create or you can autogenerate the API key and secret directly in the Cloud Console when setting up the connector.

Using the Confluent Cloud Console

Step 1: Launch your Confluent Cloud cluster

To create and launch a Kafka cluster in Confluent Cloud, see Create a kafka cluster in Confluent Cloud.

Step 2: Add a connector

In the left navigation menu, click Connectors. If you already have connectors in your cluster, click + Add connector.

Step 3: Select your connector

Click the Google Cloud Spanner CDC Source connector card.

Step 4: Enter the connector details

Before configuring the connector settings, ensure that you complete all prerequisites.

Note

An asterisk ( * ) designates a required entry.

At the Google Cloud Spanner CDC Source (Debezium) Connector screen, complete the following:

Kafka credentials

Select the way you want to provide Kafka Cluster credentials. You can choose one of the following options:
- My account: This setting allows your connector to globally access everything that you have access to. With a user account, the connector uses an API key and secret to access the Kafka cluster. This option is not recommended for production.
- Service account: This setting limits the access for your connector by using a service account. This option is recommended for production.
- Use an existing API key: This setting allows you to specify an API key and a secret pair. You can use an existing pair or create a new one. This method is not recommended for production environments.
Note
Freight clusters support only service accounts for Kafka authentication.
Click Continue.

Authentication

Configure the authentication properties:
Google Cloud Spanner connection
- GCP credentials file: The Google Cloud service account key in JSON format.
- GCP project ID: The Google Cloud project ID that contains the Spanner instance.
- Spanner instance ID: The Spanner instance ID.
- Spanner database ID: The Spanner database ID.
- Spanner host: The custom Spanner host endpoint. Leave empty to use the default Google Cloud Spanner endpoint.
- Database role: The Spanner database role to use for fine-grained access control. Leave empty if not using database roles.
Click Continue.

Configuration

Configuration properties that are not shown in the Cloud Console use the default values. For all property values and definitions, see Configuration properties.

Select output record value format: Sets the output Kafka record value format. Valid entries are AVRO, JSON_SR, or PROTOBUF.
Note
You must have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, or PROTOBUF.

Connector configuration

Change stream name: The name of the Spanner change stream to monitor.
Tables included: A comma-separated list of regular expressions matching fully-qualified table names to include. If not specified, all tables are included.
Tables excluded: A comma-separated list of regular expressions matching fully-qualified table names to exclude from monitoring.
Start time: The timestamp to start reading changes from, in ISO 8601 Instant format (for example, 2024-01-15T10:00:00Z). If not specified, starts from the current time. The start time must fall within the change stream’s retention window.
End time: The timestamp to stop reading changes, in ISO 8601 Instant format (for example, 2024-01-15T18:00:00Z). If not specified, the connector runs indefinitely. Must be after the start time.

Data encryption

Enable Client-Side Field Level Encryption for data encryption. Specify a Service Account to access the Schema Registry and associated encryption rules or keys with that schema. For more information on CSFLE or CSPE setup, see Manage encryption for connectors.

Show advanced configurations

Schema context: Select a schema context to use for this connector, if using a schema-based data format. This property defaults to the Default context, which configures the connector to use the default schema set up for Schema Registry in your Confluent Cloud environment. A schema context allows you to use separate schemas (like schema sub-registries) tied to topics in different Kafka clusters that share the same Schema Registry environment. For example, if you select a non-default context, a Source connector uses only that schema context to register a schema and a Sink connector uses only that schema context to read from. For more information about setting up a schema context, see What are schema contexts and when should you use them?.

Auto-restart policy

Enable Connector Auto-restart: Enables the auto-restart behavior of the connector and its task in the event of user-actionable errors. Defaults to true, enabling the connector to automatically restart in case of user-actionable errors. Set this property to false to disable auto-restart for failed connectors. If disabled, you must manually restart the connector.

Additional Configs

Value Converter Replace Null With Default: Specifies whether to replace fields that have a default value and that are null to the default value. When set to true, the connector uses the default value; otherwise, it uses null. Applies to the JSON converter.
Value Converter Reference Subject Name Strategy: Sets the subject reference name strategy for values. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. You can use this strategy only with PROTOBUF format; the default strategy is DefaultReferenceSubjectNameStrategy.
Value Converter Schemas Enable: Includes schema within each of the serialized values. Input messages must contain schema and payload fields and must not contain additional fields. For plain JSON data, set this to false. Applies to the JSON converter.
Errors Tolerance: Use this property to configure the connector’s error handling behavior.
Warning
Use this property with caution for sink connectors, as it can lead to data loss. If you set this property to all, the connector does not fail on errant records, but logs them (and sends to DLQ for sink connectors) and continues processing. If you set this property to none, the connector task fails on errant records.
Value Converter Ignore Default For Nullables: When set to true, this property ensures that the corresponding record in Kafka is null, instead of showing the default column value. Applies to the AVRO, PROTOBUF, and JSON_SR converters.
Value Converter Decimal Format: Specifies the JSON or JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals: BASE64 to serialize DECIMAL logical types as base64 encoded binary data, and NUMERIC to serialize DECIMAL logical type values in JSON or JSON_SR as a number representing the decimal value.
Key Converter Schema ID Serializer: The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.
Value Converter Connect Meta Data: Enables the Connect converter to add its metadata to the output schema. Applies to Avro converters.
Value Converter Value Subject Name Strategy: Determines how to construct the subject name under which the value schema is registered with Schema Registry.
Key Converter Key Subject Name Strategy: Determines how to construct the subject name for key schema registration.
Value Converter Schema ID Serializer: The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.

Connector configuration

Topic auto-creation partitions: The number of partitions for automatically created data change topics. Per-row ordering is preserved regardless of partition count, because records are partitioned by primary key. Increasing this value allows higher consumer parallelism but loses global ordering across different rows.
Tombstones on delete: Controls whether a tombstone event should be generated after a delete event.
- true: a delete operation is represented by a delete event and a subsequent tombstone event.
- false: only a delete event is emitted.
Emitting a tombstone event, which is the default behavior, allows Kafka to completely delete all events associated with the key of the deleted row when log compaction is enabled for the topic.
Heartbeat interval: The interval at which heartbeat messages are sent, in milliseconds. The value must be between 100 and 300000.
Max missed heartbeats: The maximum number of missed heartbeats before the connector identifies that a partition is stuck.
Schema name adjustment mode: Specifies how schema names should be adjusted for compatibility with the message converter used by the connector. Use one of the following:
- none does not apply any adjustment.
- avro replaces the characters that cannot be used in the Avro type name with an underscore.
- avro_unicode replaces the underscore or characters that cannot be used in the Avro type name with corresponding unicode like _uxxxx.
Note
_ is an escape sequence like backslash in Java.
Field name adjustment mode: Specifies how field names should be adjusted for compatibility with the message converter used by the connector. Use one of the following:
- none does not apply any adjustment.
- avro replaces the characters that cannot be used in the AVRO type name with an underscore.
- avro_unicode replaces the underscore or characters that cannot be used in the AVRO type name with corresponding unicode like _uxxxx.
Note that _ is an escape sequence like backslash in Java.
Enable low watermark: Enables low-watermark tracking for monitoring connector progress.
Low watermark update period: The low-watermark update period for each task, in milliseconds.
Low watermark stamp interval: The interval, in milliseconds, at which the connector periodically sends low-watermark stamp messages to all table topics and Kafka partitions.

Sync topic configuration

Sync topic cleanup policy: The sync topic cleanup policy.
Sync topic retention: The sync topic retention, in milliseconds.
Sync topic segment time: The sync topic segment time, in milliseconds.

Transforms

Single Message Transformations: To add a new SMT, see Add transforms. For more information about unsupported SMTs, see Unsupported transformations.

For more information about the Debezium SMTs ExtractNewRecordState and EventRouter (Debezium), see Debezium transformations.

For all property values and definitions, see Configuration properties.

Click Continue.

Sizing

Based on the number of topic partitions you select, you are provided with a recommended number of tasks.

To change the number of tasks, use the Range Slider to select the desired number of tasks.
Click Continue.

Review and Launch

Verify the connection details by previewing the running configuration.
After you’ve validated that the properties are configured to your satisfaction, click Launch.
The status for the connector should go from Provisioning to Running.

Step 5: Check the Kafka topic

After the connector is running, verify that messages are populating your Kafka topics. The connector creates one topic per monitored Spanner table, using the naming convention <connectorID>.<tableName>.

For more information and examples to use with the Confluent Cloud API for Connect, see the Confluent Cloud API for Connect Usage Examples section.

Using the Confluent CLI

Before configuring the connector settings, ensure that you complete all prerequisites. Complete the following steps to set up and run the connector using the Confluent CLI.

Step 1: List the available connectors

Enter the following command to list available connectors:

confluent connect plugin list

Step 2: List the connector configuration properties

Enter the following command to show the connector configuration properties:

confluent connect plugin describe <connector-plugin-name>

The command output shows the required and optional configuration properties.

Step 3: Create the connector configuration file

Create a JSON file that contains the connector configuration properties. The following example shows required and optional connector properties:

{
  "connector.class": "SpannerCdcSource",
  "name": "SpannerCdcSourceConnector_0",
  "kafka.auth.mode": "KAFKA_API_KEY",
  "kafka.api.key": "<my-kafka-api-key>",
  "kafka.api.secret": "<my-kafka-api-secret>",
  "gcp.spanner.credentials.json": "<my-gcp-service-account-key-json>",
  "gcp.spanner.project.id": "<my-gcp-project-id>",
  "gcp.spanner.instance.id": "<my-spanner-instance-id>",
  "gcp.spanner.database.id": "<my-spanner-database-id>",
  "gcp.spanner.change.stream": "<my-change-stream-name>",
  "output.data.format": "AVRO",
  "tasks.max": "1"
}

Note the following property definitions:

"connector.class": Identifies the connector plugin name.
"name": Sets a name for your new connector.

"kafka.auth.mode": Identifies the connector authentication mode you want to use. There are two options: SERVICE_ACCOUNT or KAFKA_API_KEY (the default). To use an API key and secret, specify the configuration properties kafka.api.key and kafka.api.secret, as shown in the example configuration (above). To use a service account, specify the Resource ID in the property kafka.service.account.id=<service-account-resource-ID>. To list the available service account resource IDs, use the following command:
```
confluent iam service-account list
```
For example:
```
confluent iam service-account list

   Id     | Resource ID |       Name        |    Description
+---------+-------------+-------------------+-------------------
   123456 | sa-l1r23m   | sa-1              | Service account 1
   789101 | sa-l4d56p   | sa-2              | Service account 2
```

"gcp.spanner.credentials.json": The Google Cloud service account key in JSON format. The service account must have read access to the Spanner instance.
"gcp.spanner.project.id": The Google Cloud project ID that contains the Spanner instance.
"gcp.spanner.instance.id": The Spanner instance ID.
"gcp.spanner.database.id": The Spanner database ID.
"gcp.spanner.change.stream": The name of the Spanner change stream to monitor.
"output.data.format": Sets the output Kafka record value format. Valid entries are AVRO, JSON_SR, or PROTOBUF. You must have Confluent Cloud Schema Registry configured.
"tasks.max": The maximum number of tasks for the connector to use.

Note

To enable CSFLE or CSPE for data encryption, specify the following properties:

csfle.enabled: Flag to indicate whether the connector honors CSFLE or CSPE rules.
sr.service.account.id: A Service Account to access the Schema Registry and associated encryption rules or keys with that schema.

For more information on CSFLE or CSPE setup, see Manage encryption for connectors.

Single Message Transformations: For details about adding SMTs using the Confluent CLI, see Single Message Transformations (SMT). For more information about the Debezium SMTs ExtractNewRecordState and EventRouter (Debezium), see Debezium transformations.

For all properties and definitions, see Configuration properties.

Step 4: Load the properties file and create the connector

Enter the following command to load the configuration and start the connector:

confluent connect cluster create --config-file <file-name>.json

For example:

confluent connect cluster create --config-file google-spanner-cdc-source.json

Example output:

Created connector SpannerCdcSourceConnector_0 lcc-ix4dl

Step 5: Check the connector status

Enter the following command to check the connector status:

confluent connect cluster list

Example output:

ID          |             Name                  | Status  |  Type
+-----------+-----------------------------------+---------+-------+
lcc-ix4dl   | SpannerCdcSourceConnector_0       | RUNNING | source

Step 6: Check the Kafka topic

After the connector is running, verify that messages are populating your Kafka topics. The connector creates one topic per monitored Spanner table.

For more information and examples to use with the Confluent Cloud API for Connect, see the Confluent Cloud API for Connect Usage Examples section.

Configuration properties

Use the following configuration properties with the fully-managed connector.

How should we connect to your data?

name

Sets a name for your connector.

Type: string
Valid Values: A string at most 64 characters long
Importance: high

Kafka Cluster credentials

kafka.auth.mode

Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode, whenever possible.

Type: string
Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY
Importance: high

kafka.api.key

Kafka API Key. Required when kafka.auth.mode==KAFKA_API_KEY.

Type: password
Importance: high

kafka.service.account.id

The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.

Type: string
Importance: high

kafka.api.secret

Secret associated with Kafka API key. Required when kafka.auth.mode==KAFKA_API_KEY.

Type: password
Importance: high

Output messages

output.data.format

Sets the output Kafka record value format. Valid entries are AVRO, JSON_SR, or PROTOBUF. Note that you must have Confluent Cloud Schema Registry configured to use a schema-based message format such as AVRO, JSON_SR, or PROTOBUF.

Type: string
Default: JSON_SR
Importance: high

Google Cloud Spanner connection

gcp.spanner.credentials.json

The Google Cloud service account key in JSON format.

Type: password
Importance: high

gcp.spanner.project.id

The Google Cloud project ID that contains the Spanner instance.

Type: string
Importance: high

gcp.spanner.instance.id

The Spanner instance ID.

Type: string
Importance: high

gcp.spanner.database.id

The Spanner database ID.

Type: string
Importance: high

gcp.spanner.host

The custom Spanner host endpoint. Leave empty to use the default Google Cloud Spanner endpoint.

Type: string
Default: “”
Importance: low

gcp.spanner.database.role

The Spanner database role to use for fine-grained access control. Leave empty if not using database roles.

Type: string
Default: “”
Importance: medium

Connector configuration

gcp.spanner.change.stream

The name of the Spanner change stream to monitor.

Type: string
Importance: high

table.include.list

A comma-separated list of regular expressions matching fully-qualified table names to include. If not specified, all tables are included.

Type: list
Importance: medium

table.exclude.list

A comma-separated list of regular expressions matching fully-qualified table names to exclude from monitoring.

Type: list
Importance: medium

gcp.spanner.start.time

The timestamp to start reading changes from, in ISO 8601 Instant format (e.g., 2024-01-15T10:00:00Z). If not specified, starts from the current time. The start time must fall within the change stream’s retention window.

Type: string
Importance: medium

gcp.spanner.end.time

The timestamp to stop reading changes, in ISO 8601 Instant format (e.g., 2024-01-15T18:00:00Z). If not specified, the connector runs indefinitely. Must be after the start time.

Type: string
Importance: low

topic.creation.default.partitions

The number of partitions for automatically created data change topics. Per-row ordering is preserved regardless of partition count, because records are partitioned by primary key. Increasing this value allows higher consumer parallelism but loses global ordering across different rows.

Type: int
Default: 1
Importance: low

tombstones.on.delete

Controls whether a tombstone event should be generated after a delete event.

true - a delete operation is represented by a delete event and a subsequent tombstone event.

false - only a delete event is emitted.

Emitting tombstone events (the default behavior) allows Kafka to completely delete all events pertaining to the key of the deleted row when log compaction is enabled for the topic.

Type: boolean
Default: true
Importance: medium

heartbeat.interval.ms

The interval at which heartbeat messages are sent, in milliseconds. Must be between 100 and 300000.

Type: int
Default: 300000 (5 minutes)
Valid Values: [100,…,300000]
Importance: low

connector.spanner.max.missed.heartbeats

The maximum number of missed heartbeats before the connector identifies that a partition is stuck.

Type: int
Default: 10
Valid Values: [1,…]
Importance: low

schema.name.adjustment.mode

Specifies how schema names should be adjusted for compatibility with the message converter used by the connector. Possible settings are none, avro, and avro_unicode.

none does not apply any adjustment.

avro replaces the characters that cannot be used in the AVRO type name with an underscore.

avro_unicode replaces the underscore or characters that cannot be used in the AVRO type name with corresponding unicode like _uxxxx.

Note that _ is an escape sequence like backslash in Java.

Type: string
Default: none
Valid Values: avro, avro_unicode, none
Importance: medium

field.name.adjustment.mode

Specifies how field names should be adjusted for compatibility with the message converter used by the connector. Possible settings are none, avro, and avro_unicode.

none does not apply any adjustment.

avro replaces the characters that cannot be used in the AVRO type name with an underscore.

avro_unicode replaces the underscore or characters that cannot be used in the AVRO type name with corresponding unicode like _uxxxx.

Note that _ is an escape sequence like backslash in Java.

Type: string
Default: none
Valid Values: avro, avro_unicode, none
Importance: medium

gcp.spanner.low-watermark.enabled

Enables low-watermark tracking for monitoring connector progress.

Type: boolean
Default: false
Importance: low

gcp.spanner.low-watermark.update-period.ms

The low-watermark update period for each task, in milliseconds.

Type: long
Default: 1000 (1 second)
Valid Values: [1000,…]
Importance: low

gcp.spanner.low-watermark.stamp.interval

The interval, in milliseconds, at which the connector periodically sends low-watermark stamp messages to all table topics and Kafka partitions.

Type: int
Default: 10000
Valid Values: [5000,…]
Importance: low

Sync topic configuration

connector.spanner.sync.cleanup.policy

The sync topic cleanup policy.

Type: string
Default: delete
Valid Values: compact, delete
Importance: low

connector.spanner.sync.retention.ms

The sync topic retention, in milliseconds.

Type: int
Default: 86400000 (1 day)
Importance: low

connector.spanner.sync.segment.ms

The sync topic segment time, in milliseconds.

Type: int
Default: 43200000 (12 hours)
Importance: low

Number of tasks for this connector

tasks.max

Maximum number of tasks for the connector.

Type: int
Valid Values: [1,…]
Importance: high

Schema Config

schema.context.name

Add a schema context name. A schema context represents an independent scope in Schema Registry. It is a separate sub-schema tied to topics in different Kafka clusters that share the same Schema Registry instance. If not used, the connector uses the default schema configured for Schema Registry in your Confluent Cloud environment.

Type: string
Default: default
Importance: medium

Auto-restart policy

auto.restart.on.user.error

Enable connector to automatically restart on user-actionable errors.

Type: boolean
Default: true
Importance: medium

Additional Configs

header.converter

The converter class for the headers. This is used to serialize and deserialize the headers of the messages.

Type: string
Importance: low

producer.override.compression.type

The compression type for all data generated by the producer. Valid values are none, gzip, snappy, lz4, and zstd.

Type: string
Importance: low

producer.override.linger.ms

The producer groups together any records that arrive in between request transmissions into a single batched request. More details can be found in the documentation: https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms.

Type: long
Valid Values: [100,…,1000]
Importance: low

value.converter.allow.optional.map.keys

Allow optional string map key when converting from Connect Schema to Avro Schema. Applicable for Avro Converters.

Type: boolean
Importance: low

value.converter.auto.register.schemas

Specify if the Serializer should attempt to register the Schema.

Type: boolean
Importance: low

value.converter.connect.meta.data

Allow the Connect converter to add its metadata to the output schema. Applicable for Avro Converters.

Type: boolean
Importance: low

value.converter.enhanced.avro.schema.support

Enable enhanced schema support to preserve package information and Enums. Applicable for Avro Converters.

Type: boolean
Importance: low

value.converter.enhanced.protobuf.schema.support

Enable enhanced schema support to preserve package information. Applicable for Protobuf Converters.

Type: boolean
Importance: low

value.converter.flatten.unions

Whether to flatten unions (oneofs). Applicable for Protobuf Converters.

Type: boolean
Importance: low

value.converter.generate.index.for.unions

Whether to generate an index suffix for unions. Applicable for Protobuf Converters.

Type: boolean
Importance: low

value.converter.generate.struct.for.nulls

Whether to generate a struct variable for null values. Applicable for Protobuf Converters.

Type: boolean
Importance: low

value.converter.int.for.enums

Whether to represent enums as integers. Applicable for Protobuf Converters.

Type: boolean
Importance: low

value.converter.latest.compatibility.strict

Verify latest subject version is backward compatible when use.latest.version is true.

Type: boolean
Importance: low

value.converter.object.additional.properties

Whether to allow additional properties for object schemas. Applicable for JSON_SR Converters.

Type: boolean
Importance: low

value.converter.optional.for.nullables

Whether nullable fields should be specified with an optional label. Applicable for Protobuf Converters.

Type: boolean
Importance: low

value.converter.optional.for.proto2

Whether proto2 optionals are supported. Applicable for Protobuf Converters.

Type: boolean
Importance: low

value.converter.scrub.invalid.names

Whether to scrub invalid names by replacing invalid characters with valid characters. Applicable for Avro and Protobuf Converters.

Type: boolean
Importance: low

value.converter.use.latest.version

Use latest version of schema in subject for serialization when auto.register.schemas is false.

Type: boolean
Importance: low

value.converter.use.optional.for.nonrequired

Whether to set non-required properties to be optional. Applicable for JSON_SR Converters.

Type: boolean
Importance: low

value.converter.wrapper.for.nullables

Whether nullable fields should use primitive wrapper messages. Applicable for Protobuf Converters.

Type: boolean
Importance: low

value.converter.wrapper.for.raw.primitives

Whether a wrapper message should be interpreted as a raw primitive at root level. Applicable for Protobuf Converters.

Type: boolean
Importance: low

errors.tolerance

Use this property if you would like to configure the connector’s error handling behavior. WARNING: This property should be used with CAUTION for SOURCE CONNECTORS as it may lead to dataloss. If you set this property to ‘all’, the connector will not fail on errant records, but will instead log them (and send to DLQ for Sink Connectors) and continue processing. If you set this property to ‘none’, the connector task will fail on errant records.

Type: string
Default: none
Importance: low

key.converter.key.schema.id.serializer

The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.

Type: string
Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer
Importance: low

key.converter.key.subject.name.strategy

How to construct the subject name for key schema registration.

Type: string
Default: TopicNameStrategy
Importance: low

value.converter.decimal.format

Specify the JSON/JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals:

BASE64 to serialize DECIMAL logical types as base64 encoded binary data and

NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.

Type: string
Default: BASE64
Importance: low

value.converter.flatten.singleton.unions

Whether to flatten singleton unions. Applicable for Avro and JSON_SR Converters.

Type: boolean
Default: false
Importance: low

value.converter.ignore.default.for.nullables

When set to true, this property ensures that the corresponding record in Kafka is NULL, instead of showing the default column value. Applicable for AVRO,PROTOBUF and JSON_SR Converters.

Type: boolean
Default: false
Importance: low

value.converter.reference.subject.name.strategy

Set the subject reference name strategy for value. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.

Type: string
Default: DefaultReferenceSubjectNameStrategy
Importance: low

value.converter.replace.null.with.default

Whether to replace fields that have a default value and that are null to the default value. When set to true, the default value is used, otherwise null is used. Applicable for JSON Converter.

Type: boolean
Default: true
Importance: low

value.converter.schemas.enable

Include schemas within each of the serialized values. Input messages must contain schema and payload fields and may not contain additional fields. For plain JSON data, set this to false. Applicable for JSON Converter.

Type: boolean
Default: false
Importance: low

value.converter.value.schema.id.serializer

The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.

Type: string
Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer
Importance: low

value.converter.value.subject.name.strategy

Determines how to construct the subject name under which the value schema is registered with Schema Registry.

Type: string
Default: TopicNameStrategy
Importance: low

Frequently asked questions

What happens if I set a start time outside the change stream’s retention window?

The connector fails with an OUT_OF_RANGE error. The start time must fall within the change stream’s retention window, which defaults to one day. You can configure it up to 30 days using Spanner DDL.

Do I need to create the change stream before configuring the connector?

Yes. The connector does not create change streams automatically. Create the change stream using Spanner DDL before configuring the connector. For example:

CREATE CHANGE STREAM my_stream FOR ALL

What timestamp format does the connector accept for start and end time?

ISO 8601 Instant format with a Z suffix. For example: 2024-01-15T10:00:00Z. Timezone offsets like +05:30 are not supported.

Can I get duplicate records from this connector?

Yes. The connector provides at-least-once delivery semantics. If a task restarts between writing records to Kafka and committing offsets, the connector might replay records. Design your consumers to handle duplicates.

How does the connector detect stuck partitions?

The connector uses heartbeat messages. If a change stream partition misses more than Max missed heartbeats consecutive heartbeats, the connector considers the partition stuck and takes corrective action. The detection window is heartbeat.interval.ms × max.missed.heartbeats. With defaults (300,000 ms × 10), this is approximately 50 minutes.

What is the low watermark and when should I enable it?

The low watermark provides a completeness guarantee: a timestamp at which the connector has streamed all events with an earlier commit timestamp to Kafka. Enable it if your downstream consumers need to know all events up to time T have arrived, for example, when building consistent materialized views.

When enabled, the connector embeds a low_watermark field in every record’s source struct, and low-watermark stamp records are periodically published to all data topics.

What happens if the connector is stopped longer than the sync topic retention period?

The connector might lose its prior state. On restart, it might re-read changes from the change stream’s start, potentially causing duplicate records. Increase the Sync topic retention value if you expect extended downtime. The default is 24 hours.

How do I reset the connector to re-read from a specific point in time?

Delete the internal sync topic, then restart the connector with the desired Start time configured. Updating Kafka Connect offsets alone does not reset the connector’s read position.

Next steps

For an example that shows fully-managed Confluent Cloud connectors in action with Confluent Cloud for Apache Flink, see the Cloud ETL Demo. This example also shows how to use Confluent CLI to manage your resources in Confluent Cloud.

Topology diagram for the Confluent Cloud ETL demo showing connectors, Apache Kafka topics, and Apache Flink processing.