Debezium connector for CockroachDB

Table of Contents

• Overview

• How the connector works

  • Architecture
  • Snapshots
  • Streaming changes
  • Reusing an existing changefeed
  • Heartbeats
  • Schema evolution
  • Incremental snapshots
  • Topic names

• Data change events

  • Change event keys
  • Change event values
  • create events
  • update events
  • delete events
  • Tombstone events

• Data type mappings

  • CockroachDB types
  • Default values

• Setting up CockroachDB

  • Granting changefeed permissions

• Securing the changefeed sink

• Deployment

  • Connector configuration example
  • Adding the connector configuration
  • Monitoring
  • Customized MBean names
  • Connector configuration properties

• Behavior when things go wrong

  • Configuration and startup errors
  • CockroachDB becomes unavailable
  • Kafka Connect process stops gracefully
  • Kafka Connect process crashes
  • Kafka becomes unavailable
  • Connector is stopped for an extended period
  • Enabling debug logging

• Limitations

The Debezium CockroachDB connector captures row-level changes from a CockroachDB enriched changefeed and streams them into Kafka topics.

CockroachDB changefeeds detects data changes in the database — inserts, updates, and deletes — in near-real-time. The connector creates a changefeed that publishes events to an intermediate Kafka cluster, then consumes those events, transforms them into the standard Debezium envelope format, and produces them to the output Kafka topics that downstream consumers subscribe to.

Overview

CockroachDB is a distributed SQL database that provides strong consistency, horizontal scalability, and survivability. Unlike traditional databases that rely on a write-ahead log (WAL) for change data capture, CockroachDB provides a native changefeed mechanism that streams row-level changes directly from the storage layer.

The Debezium CockroachDB connector leverages this native changefeed capability. The connector produces a change event for every row-level insert, update, and delete operation that it captures and sends change event records for each table to a separate Kafka topic. Client applications read the Kafka topics that correspond to the database tables of interest and can react to each row-level event that they receive from those topics.

The connector is tolerant of failures. As the connector reads changes and produces events, it records the last resolved timestamp it processes. If the connector stops for any reason (including communication failures, network problems, or crashes), after the connector restarts, it resumes streaming from the last record that it processed.

How the connector works

To optimally configure and run a Debezium CockroachDB connector, it is helpful to understand how the connector performs snapshots, streams change events, determines Kafka topic names, and uses metadata.

Architecture

The CockroachDB connector uses a two-stage Kafka architecture:

1. CockroachDB changefeed to Intermediate Kafka: The connector creates a single CockroachDB changefeed that uses the enriched envelope format for all configured tables (CREATE CHANGEFEED FOR table1, table2, …​) CockroachDB automatically routes events to per-table Kafka topics in the intermediate cluster. The enriched format includes both schema metadata and the full before and after row states.
2. Intermediate Kafka to Debezium to Output Kafka: The connector subscribes to all per-table Kafka topics in a single KafkaConsumer; routes each event to the correct table based on the topic name; transforms the enriched changefeed events into the standard Debezium envelope format (with before, after, source, and op fields); and produces them to the final output Kafka topics.

By using this architecture the connector can leverage the native, highly-available changefeed infrastructure of CockroachDB to reliably capture data and provide it to downstream consumers in the standard Debezium event format. By default, the connector consolidates all configured tables into a single changefeed. By consolidating data in a single changefeed job, the connector conforms to the recommended limit on the number of changefeeds per CockroachDB cluster (approximately 80 at the time of writing). To improve throughput and prevent performance coupling, CockroachDB best practices advise against using a single changefeed to monitor a large number of tables. In environments with large table counts, consider setting cockroachdb.changefeed.max.tables.per.changefeed to split the tables across several changefeeds, or running multiple connector instances so that each connector captures a related subset of the tables.

Snapshots

CockroachDB changefeeds natively support an initial_scan option that backfills all existing rows before streaming ongoing changes. The Debezium CockroachDB connector maps its snapshot.mode configuration to the CockroachDB initial_scan changefeed option, so the connector does not require a separate JDBC-based snapshot phase.

During the initial scan phase, captured events are marked op=r to designate a read operation, which distinguishes snapshot events from ongoing change events. After the initial scan completes and the changefeed transitions to streaming, events are marked with the appropriate operation type (c for create, u for update, d for delete).

The following table shows how each snapshot.mode maps to the CockroachDB initial_scan option:

Table 1. Snapshot mode to initial_scan mapping

Streaming changes

After an initial scan completes, the CockroachDB connector streams changes continuously. When a row-level change occurs in CockroachDB, the changefeed writes a corresponding event to the intermediate Kafka topic. The connector consumes these events, transforms them into Debezium change events, and forwards them to the output Kafka topics.

CockroachDB changefeeds emit resolved timestamp messages at a configurable interval. These messages indicate that all changes up to that timestamp have been emitted. The connector uses resolved timestamps for offset tracking, so that on restart it can create a new changefeed with a cursor pointing to the last resolved timestamp, ensuring that it does not miss any events.

When the connector receives changes it transforms the events into Debezium read, create, update, or delete event records. The connector forwards these change records to the Kafka Connect framework, which is running in the same process. The Kafka Connect process asynchronously writes the change event records in the same order in which they were generated to the appropriate Kafka topic.

Periodically, Kafka Connect writes the most recent offset to another Kafka topic. The offset indicates source-specific position information that Debezium includes with each event. For the CockroachDB connector, the last resolved timestamp is the offset.

When Kafka Connect gracefully shuts down, it stops the connectors, flushes all event records to Kafka, and records the last offset received from each connector. When Kafka Connect restarts, it reads the last recorded offset for each connector, and starts each connector at its last recorded offset.

Reusing an existing changefeed

By default, the connector creates and manages its own CockroachDB changefeed. Before it creates a changefeed, the connector checks whether a running changefeed already covers the configured tables. If a matching changefeed is found, the connector skips creation and consumes from the existing changefeed instead. This behavior keeps connector restarts idempotent and avoids creating duplicate changefeed jobs on the cluster. It also lets an operator pre-provision a changefeed that the connector then attaches to.

A changefeed is considered a match only when both of the following are true for the running changefeed job:

• Its description includes the fully-qualified name of each configured table.
• Its description includes the topic_prefix that the connector expects, which is the value of cockroachdb.changefeed.sink.topic.prefix, or the value of topic.prefix when no sink topic prefix is set.

For the connector to consume the existing changefeed correctly, the changefeed must publish to Kafka topics whose names match the topics that the connector subscribes to, in the format topicPrefix.database.schema.table. To produce that topic layout, and to emit events in the structure that the connector expects, create the changefeed with the following options:

full_table_name

Required so that CockroachDB names topics in database.schema.table form.

topic_prefix='prefix.'

Required, and must equal the connector’s topic prefix described in the preceding list. The specified prefix must include a trailing dot (.).

envelope='enriched'

Required. The connector consumes the enriched envelope.

enriched_properties

Optional. The connector reads the base enriched fields (op, ts_ns, after, and before when diff is set), so it does not require any enriched property to consume the feed. The default value, source ensures that intermediate topics include origin and commit-time metadata.

format='json'

Required. The connector consumes JSON-encoded changefeed messages.

diff, updated, resolved

Set these to match the connector configuration so that before-images, update markers, and resolved timestamps are available.

If any of these conditions is not met, the connector does not recognize the existing changefeed and creates its own. A changefeed created with a different topic_prefix or without full_table_name is not recognized as a match, and the connector creates its own. If a running changefeed matches the connector’s topic prefix and tables, but it was created with an envelope other than enriched, the connector cannot consume it. The connector then fails to start and returns an error stating that you must recreate that changefeed with envelope='enriched' or else use a different topic prefix.

Heartbeats

Resolved timestamps in the CockroachDB changefeed serve as natural heartbeats. When the connector receives a resolved timestamp, it performs the following operations:

1. Updates the stored offset cursor to the resolved timestamp value.
2. Dispatches a Debezium heartbeat event to advance Kafka Connect offsets.

By emitting heartbeat messages the connector helps to ensure that offsets advance even during idle periods in which no data changes occur. When offsets fail to advance, monitoring systems might report that the connector is unresponsive, which can cause CockroachDB protected timestamps to accumulate, preventing garbage collection.

To enable Debezium heartbeat records on the __debezium-heartbeat.<topic.prefix> Kafka topic, set heartbeat.interval.ms to a value greater than 0 (in milliseconds). The cockroachdb.changefeed.resolved.interval property controls how frequently CockroachDB emits resolved timestamps (default: 10s).

Schema evolution

The connector automatically detects DDL changes such as ALTER TABLE ADD COLUMN, DROP COLUMN, and RENAME COLUMN without requiring a restart. When an incoming changefeed event contains fields that do not match the registered table schema, the connector adapts the schema by performing the following tasks:

1. Detects the mismatch by comparing event field names against registered column names.
2. Re-queries information_schema to retrieve the updated table definition.
3. Refreshes the internal schema and continues processing the table based on the adjusted column layout.

CockroachDB changefeeds handle schema changes natively by performing a backfill operation in which the Schema Change Manager re-emits all existing rows with the updated schema. A backfill ensures that no events are lost during the schema transition. After a backfill completes, the refreshed columns are immediately available to downstream consumers.

Incremental snapshots

The CockroachDB connector supports Debezium signal-based incremental snapshots. An incremental snapshot re-reads existing rows from one or more tables and emits them as op=r (read) events, while the connector continues to capture ongoing changes without interruption.

You can use incremental snapshots to perform the following tasks:

• Repopulate a downstream consumer that lost data.
• Backfill a newly added sink or topic.
• Verify source-target consistency by re-snapshotting and comparing.

Prepare to use incremental snapshots

To enable the connector to use incremental snapshots, you must create a signaling table and then update the connector configuration to recognize the signaling table.

Procedure

1. Create a signaling table in CockroachDB by issuing the following SQL command:

   CREATE TABLE debezium_signal (
       id STRING PRIMARY KEY,
       type STRING NOT NULL,
       data STRING
   );

2. Add the following properties to the connector configuration to reference the signaling table:

   {
       "signal.data.collection": "mydb.public.debezium_signal",
       "table.include.list": "public.my_table,public.debezium_signal"
   }

   You must include the signaling table in the table.include.list so that the changefeed can deliver signal events to the connector.

Trigger an incremental snapshot

You trigger an incremental snapshot by inserting a row in the connector’s signaling table.

Procedure

• Run a SQL command that uses the following format to add a row to the signaling table to trigger an incremental snapshot:

  INSERT INTO debezium_signal (id, type, data) VALUES
      ('snap-1', 'execute-snapshot',
       '{"data-collections": ["mydb.public.my_table"]}');

The connector rereads all rows from the specified tables and emits them as op=r events.

For more information about signaling, see Sending signals to a Debezium connector.

Topic names

The CockroachDB connector emits all data change events for a table (insert, update, and delete) to a Kafka topic that is dedicated to that table. By default, the Kafka topic name is topicPrefix.schemaName.tableName where:

• topicPrefix is the topic prefix as specified by the topic.prefix connector configuration property.
• schemaName is the name of the database schema (default: public).
• tableName is the name of the database table in which the operation occurred.

For example, suppose that cockroachdb is the topic prefix for a connector that captures changes from a database that contains two tables: orders and customers. The connector would stream records to the following two Kafka topics:

• cockroachdb.public.orders
• cockroachdb.public.customers

The connector applies naming conventions that are similar to those used by other Debezium connectors, and it supports the standard topic naming strategies.

Data change events

The Debezium CockroachDB connector generates a data change event for each row-level INSERT, UPDATE, and DELETE operation. Each event contains a key and a value. The key and value are separate documents. The structure of the key and the value depends on the table that was changed.

Debezium and Kafka Connect are designed around continuous streams of event messages. However, the structure of these events may change over time, which can be difficult for consumers to handle. To help consumers adapt to structural volatility, the connector emits self-contained events. That is, each event message either contains the schema for its content , or in environments that use a schema registry, each message contains a schema ID that a consumer can use to obtain the schema from the registry.

The following skeleton JSON documents show the basic structure of the key and value fields in an event message. However, the exact representation of the key and value documents depends on how you configure the Kafka Connect converter that you use in your application. A schema field is present a change event key or change event value only if you configure the converter to produce it. Likewise, the event key and event payload are present only if you configure a converter to produce it. If you use the JSON converter and you configure it to produce schemas, change events have the following structure:

// Key
{
 "schema": { (1)
   ...
  },
 "payload": { (2)
   ...
 }
}

// Value
{
 "schema": { (3)
   ...
 },
 "payload": { (4)
   ...
 }
}

Table 2. Overview of change event basic content

The default xref: topic naming behavior results in the connector streaming change event records to a topic whose name matches the name of the table that the event describes.

Change event keys

The change event key for a table contains fields for each column that was present in the table’s primary key at the time the event was created.

For example, consider the following definition for an orders table in the defaultdb database:

Example table

CREATE TABLE orders (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  customer_name STRING NOT NULL,
  amount DECIMAL NOT NULL,
  status STRING DEFAULT 'pending',
  created_at TIMESTAMP DEFAULT now()
);

Example change event key

Based on the preceding definition of the orders table, every change event for the table uses the same key structure. The following JSON represents this key structure:

{
  "schema": {
    "type": "struct",
    "name": "cockroachdb.public.orders.Key",
    "optional": false,
    "fields": [
      {
        "type": "string",
        "optional": false,
        "field": "id"
      }
    ]
  },
  "payload": {
    "id": "5f8a1c2e-3b4d-4e6f-8a9b-1c2d3e4f5a6b"
  }
}

Change event values

Consider the same sample orders table that was used in the change event key example:

CREATE TABLE orders (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  customer_name STRING NOT NULL,
  amount DECIMAL NOT NULL,
  status STRING DEFAULT 'pending',
  created_at TIMESTAMP DEFAULT now()
);

Based on the preceding orders table definition, the following examples show representative structures for the following types of change event values:

• create event values
• update event values
• delete event values

create events

The following example shows the value portion of a change event that the connector generates for an operation that inserts data in the orders table:

{
    "schema": { ... },
    "payload": {
        "before": null, (1)
        "after": { (2)
            "id": "5f8a1c2e-3b4d-4e6f-8a9b-1c2d3e4f5a6b",
            "customer_name": "Alice",
            "amount": "99.99",
            "status": "pending",
            "created_at": "2026-01-15T10:30:00"
        },
        "source": { (3)
            "version": "3.6.3.Final",
            "connector": "cockroachdb",
            "name": "cockroachdb_connector",
            "ts_ms": 1705312200000,
            "ts_us": 1705312200000000,
            "ts_ns": 1705312200000000000,
            "snapshot": "false",
            "db": "defaultdb",
            "sequence": "[]",
            "schema": "public",
            "table": "orders",
            "cluster": "cockroachdb_connector",
            "resolved_ts": null,
            "ts_hlc": null
        },
        "op": "c", (4)
        "ts_ms": 1705312200123, (5)
        "ts_us": 1705312200123456,
        "ts_ns": 1705312200123456789
    }
}

Table 3. Descriptions of create event value fields

Item Field name Description

1

before

An optional field that specifies the state of the row before the event occurred. When the op field is c for create, the before field is null since this change event is for new content.

2

after

An optional field that specifies the state of the row after the event occurred. In this example, the after field contains the values of the new row’s id, customer_name, amount, status, and created_at columns.

3

source

Mandatory field that describes the source metadata for the event. This field contains information that you can use to compare this event with other events, with regard to the origin of the events, the order in which the events occurred, and whether events were part of the same transaction. The source metadata includes:

• Debezium version
• Connector type and name
• Database and table that contains the new row
• Whether the event was part of a snapshot (initial scan)
• The schema name
• The logical cluster name
• The resolved timestamp (for consistency tracking)
• The Hybrid Logical Clock (HLC) timestamp. HLC is the internal timestamp format for CockroachDB.

4

op

Mandatory string that describes the type of operation that caused the connector to generate the event. In this example, c indicates that the operation created a row. Valid values are:

• r = read (initial scan / snapshot)
• c = create
• u = update
• d = delete

5

ts_ms, ts_us, ts_ns

Optional field that displays the time at which the connector processed the event. The time is based on the system clock in the JVM running the Kafka Connect task.

In the source object, ts_ms indicates the time that the change was made in the database. By comparing the value for payload.source.ts_ms with the value for payload.ts_ms, you can determine the lag between the source database update and Debezium.

update events

The value portion of a an event record for an update event in the sample orders table has the same schema as a create event for that table. Likewise, the event value’s payload has the same structure. However, the event value payload contains different values in an update event. The following example shows the JSON for the value portion of a change event that the connector generates for an update in the orders table:

{
    "schema": { ... },
    "payload": {
        "before": null, (1)
        "after": { (2)
            "id": "5f8a1c2e-3b4d-4e6f-8a9b-1c2d3e4f5a6b",
            "customer_name": "Alice",
            "amount": "99.99",
            "status": "shipped",
            "created_at": "2026-01-15T10:30:00"
        },
        "source": { (3)
            "version": "3.6.3.Final",
            "connector": "cockroachdb",
            "name": "cockroachdb_connector",
            "ts_ms": 1705312500000,
            "ts_us": 1705312500000000,
            "ts_ns": 1705312500000000000,
            "snapshot": "false",
            "db": "defaultdb",
            "sequence": "[]",
            "schema": "public",
            "table": "orders",
            "cluster": "cockroachdb_connector",
            "resolved_ts": null,
            "ts_hlc": null
        },
        "op": "u", (4)
        "ts_ms": 1705312500123,
        "ts_us": 1705312500123456,
        "ts_ns": 1705312500123456789
    }
}

Table 4. Descriptions of update event value fields

delete events

The value in a delete change event has the same schema portion as create and update events for the same table. A delete change event record provides a consumer with the information that it needs to process the removal of a row.

Debezium CockroachDB connector events are designed to work with Kafka log compaction. Log compaction enables removal of some older messages as long as at least the most recent message for every key is kept. Removing older messages lets Kafka reclaim storage space while ensuring that the topic contains a complete data set and can be used for reloading key-based state.

The following example shows the JSON for the payload portion of a change event that the connector generates for a delete event in the orders table:

{
    "schema": { ... },
    "payload": {
        "before": null,
        "after": null, (1)
        "source": { (2)
            "version": "3.6.3.Final",
            "connector": "cockroachdb",
            "name": "cockroachdb_connector",
            "ts_ms": 1705312800000,
            "ts_us": 1705312800000000,
            "ts_ns": 1705312800000000000,
            "snapshot": "false",
            "db": "defaultdb",
            "sequence": "[]",
            "schema": "public",
            "table": "orders",
            "cluster": "cockroachdb_connector",
            "resolved_ts": null,
            "ts_hlc": null
        },
        "op": "d", (3)
        "ts_ms": 1705312800123,
        "ts_us": 1705312800123456,
        "ts_ns": 1705312800123456789
    }
}

Table 5. Descriptions of delete event value fields

A delete change event record provides a consumer with the information it needs to process the removal of this row.

Tombstone events

When you delete a row, the delete event value still works with log compaction, because Kafka can remove all earlier messages that have that same key. However, for Kafka to remove all messages that have that same key, the message value must be null. To make this possible, the connector follows a delete event with a special tombstone event that has the same key but a null value.

Data type mappings

The CockroachDB connector represents changes to rows with events that are structured like the table in which the row exists. The event contains a field for each column value. The way in which Debezium represents values in the event record depends on the data type of the CockroachDB column. The following data type mappings are applied for CockroachDB data:

• Mappings for CockroachDB data types

CockroachDB types

Table 6. Mappings for CockroachDB data types

Default values

When a CockroachDB column specifies a DEFAULT clause, the connector parses the JDBC-reported default expression into the Java type that matches the column’s Kafka Connect schema, and sets it as the schema default for that field. CockroachDB annotates default expressions with a CockroachDB-specific :::TYPE suffix (for example, 0:::INT8, 'PENDING':::STRING, '[1.0,2.0,3.0]':::VECTOR). The connector strips this annotation, unwraps quoted literals, and converts the value to the column’s Java type.

Function-generated defaults such as current_timestamp(), gen_random_uuid(), now(), and unique_rowid() are not evaluated by the connector. For these columns, the connector omits the default from the schema so that CockroachDB computes the value at insert time.

Setting up CockroachDB

Before you deploy the Debezium CockroachDB connector, prepare your CockroachDB environment so that the connector can create and read changefeeds. At a high level, ensure that the required components and connectivity are in place, enable rangefeeds on the cluster, and grant the database user that the connector uses the privileges that changefeeds require. For information about enabling rangefeeds and granting the required privileges, see Granting changefeed permissions.

Prerequisites

CockroachDB cluster

A CockroachDB cluster that supports sink-based changefeeds. All changefeed features are available in every CockroachDB edition without an enterprise license.

Intermediate Kafka cluster

A Kafka cluster where CockroachDB publishes changefeed events. This can be the same Kafka cluster that Kafka Connect uses, or a separate one.

Network connectivity

The CockroachDB cluster must be able to connect to the intermediate Kafka cluster to publish changefeed events. The Kafka Connect worker must be able to read from both the CockroachDB cluster (for JDBC) and the intermediate Kafka cluster (for consuming changefeed events).

Changefeed user account

A CockroachDB user that has one of the following privileges on the target tables:

• The CHANGEFEED privilege (CockroachDB v22.2+)
• The ALL privilege
• Membership in the admin role

Granting changefeed permissions

Enable cluster-wide rangefeeds and grant the necessary privileges to the database user that the connector uses. CockroachDB validates changefeed privileges at CREATE CHANGEFEED time and returns accurate, version-specific error messages if the user lacks sufficient permissions.

Procedure

• As a CockroachDB admin user, run the following SQL commands:

  -- Enable rangefeeds (required cluster-wide setting)
  SET CLUSTER SETTING kv.rangefeed.enabled = true;
  
  -- Grant changefeed privilege to a specific user
  GRANT CHANGEFEED ON TABLE orders TO myuser;
  GRANT VIEWCLUSTERSETTING TO myuser;
  
  -- Or grant on all tables in a database
  GRANT CHANGEFEED ON ALL TABLES IN DATABASE defaultdb TO myuser;

  Granting VIEWCLUSTERSETTING lets the connector verify that kv.rangefeed.enabled is set to true, which is a prerequisite for changefeeds.

Securing the changefeed sink

When a changefeed sink uses TLS or mutual TLS (mTLS), the connector can load PEM files from disk and automatically add their base64-encoded contents to the sink URI at startup. This automatic injection removes the need to manually copy large, encoded certificate strings into the connector configuration.

Set one or more of the following properties to specify the mechanism that the connector uses to encrypt sink connections:

cockroachdb.changefeed.sink.tls.ca.cert.file

Path to a PEM-encoded CA certificate that verifies the sink broker’s certificate.

cockroachdb.changefeed.sink.tls.client.cert.file

Path to a PEM-encoded client certificate used for mutual TLS to the sink.

cockroachdb.changefeed.sink.tls.client.key.file

Path to a PEM-encoded client private key used for mutual TLS to the sink.

For each property that is set, the connector reads the file, validates that it is readable and non-empty, base64-encodes the contents, URL-encodes the result, and appends a corresponding query parameter to cockroachdb.changefeed.sink.uri in the form expected by the configured sink type. For the Kafka sink, the appended parameters are ca_cert=…​, client_cert=…​, client_key=…​, and tls_enabled=true (added automatically whenever any of the three TLS file options is set). File-based values overwrite any same-named query parameter that is already present inline in the sink URI.

Example configuration for a Kafka sink secured with mTLS

{
  "cockroachdb.changefeed.sink.type": "kafka",
  "cockroachdb.changefeed.sink.uri": "kafka://kafka.example.com:9093",
  "cockroachdb.changefeed.sink.tls.ca.cert.file": "/etc/kafka/secrets/ca.pem",
  "cockroachdb.changefeed.sink.tls.client.cert.file": "/etc/kafka/secrets/client.pem",
  "cockroachdb.changefeed.sink.tls.client.key.file": "/etc/kafka/secrets/client-key.pem"
}

The connector validates each configured file path at startup and fails fast if the file does not exist, is unreadable, or is empty. Currently only the Kafka sink consumes these TLS parameters; for other sink types the sink URI passes through unchanged.

Deployment

To deploy a Debezium CockroachDB connector you obtain the connector plug-in files archive, add the JAR files to your Kafka Connect environment, and then edit the plugin.path configuration to reference the location of the files.

If you are working with immutable containers, see the Debezium container images for Kafka and Kafka Connect.

You can also run Debezium on Kubernetes and OpenShift.

Prerequisites

• Kafka and Kafka Connect are deployed and running.

Procedure

1. Download the archive file to a temporary directory.
2. Copy the downloaded file to a directory in your Kafka Connect environment, for example, /usr/local/share/kafka/plugins.
3. Extract the JAR file into the directory and then remove the downloaded archive file.
4. Add an entry for the directory with the JAR files to the Kafka Connect plugin.path.
5. Configure the connector and add the configuration to your Kafka Connect cluster.
6. Restart the Kafka Connect process so that it picks up the new JAR files.

Connector configuration example

The following example shows a possible configuration for a CockroachDB connector that connects to a CockroachDB cluster and captures changes from all tables in the public schema of the defaultdb database.

{
  "name": "cockroachdb-connector",  (1)
  "config": {
    "connector.class": "io.debezium.connector.cockroachdb.CockroachDBConnector", (2)
    "database.hostname": "cockroachdb-host", (3)
    "database.port": "26257", (4)
    "database.user": "myuser", (5)
    "database.password": "mypassword", (6)
    "database.dbname": "defaultdb", (7)
    "topic.prefix": "cockroachdb", (8)
    "cockroachdb.changefeed.sink.uri": "kafka://kafka-broker:9092", (9)
    "snapshot.mode": "initial", (10)
    "tasks.max": "1" (11)
  }
}

For more information about the properties that you can specify in the connector configuration, see the complete list of CockroachDB connector properties .

Adding the connector configuration

Edit the CockroachDB connector configuration to specify how you want the connector to behave. After you finalize the configuration, use the Kafka Connect API to apply the configuration to the cluster.

Prerequisites

• CockroachDB is running and accessible.
• The intermediate Kafka cluster is running and accessible from the CockroachDB cluster.
• The CockroachDB connector is installed.

Procedure

1. Create the configuration for the CockroachDB connector.

2. Use the Kafka Connect REST API to submit a POST request that applies the JSON for the connector configuration to the Kafka Connect cluster.

   For example:

   curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
     http://localhost:8083/connectors/ -d @connector-config.json

Results

When the connector starts, it connects to the CockroachDB database, creates a changefeed, and starts generating data change events for row-level operations and streaming change event records to Kafka topics.

Monitoring

The Debezium CockroachDB connector includes built-in support for both the JMX metrics that Kafka and Kafka Connect provide, as well as snapshot metrics and streaming metrics that provide insights into connector behavior during the initial scan and during the capture and streaming of change event records.

Additional resources

• Using JMX to expose metrics (Debezium monitoring documentation)^ .

Customized MBean names

Debezium connectors expose metrics via the MBean name for the connector. These metrics, which are specific to each connector instance, provide data about the behavior of the connector’s snapshot, streaming, and schema history processes.

By default, when you deploy a correctly configured connector, Debezium generates a unique MBean name for each of the different connector metrics. To view the metrics for a connector process, you configure your observability stack to monitor its MBean. But these default MBean names depend on the connector configuration; configuration changes can result in changes to the MBean names. A change to the MBean name breaks the linkage between the connector instance and the MBean, disrupting monitoring activity. In this scenario, you must reconfigure the observability stack to use the new MBean name if you want to resume monitoring.

To prevent monitoring disruptions that result from MBean name changes, you can configure custom metrics tags. You configure custom metrics by adding the custom.metric.tags property to the connector configuration. The property accepts key-value pairs in which each key represents a tag for the MBean object name, and the corresponding value represents the value of that tag. For example: k1=v1,k2=v2. Debezium appends the specified tags to the MBean name of the connector.

After you configure the custom.metric.tags property for a connector, you can configure the observability stack to retrieve metrics associated with the specified tags. The observability stack then uses the specified tags, rather than the mutable MBean names to uniquely identify connectors. Later, if Debezium redefines how it constructs MBean names, or if the topic.prefix in the connector configuration changes, metrics collection is uninterrupted, because the metrics scrape task uses the specified tag patterns to identify the connector.

A further benefit of using custom tags, is that you can use tags that reflect the architecture of your data pipeline, so that metrics are organized in a way that suits you operational needs. For example, you might specify tags with values that declare the type of connector activity, the application context, or the data source, for example, db1-streaming-for-application-abc. If you specify multiple key-value pairs, all of the specified pairs are appended to the connector’s MBean name.

The following example illustrates how tags modify the default MBean name.

Example 1. How custom tags modify the connector MBean name

By default, the CockroachDB connector uses the following MBean name for streaming metrics:

debezium.CockroachDB:type=connector-metrics,context=streaming,server=<topic.prefix>

If you set the value of custom.metric.tags to database=salesdb-streaming,table=inventory, Debezium generates the following custom MBean name:

debezium.CockroachDB:type=connector-metrics,context=streaming,server=<topic.prefix>,database=salesdb-streaming,table=inventory

Snapshot metrics

The connector backfills existing rows through the CockroachDB changefeed initial_scan and drives the standard snapshot lifecycle around it, so the snapshot metrics are populated during the initial scan.

The MBean is debezium.CockroachDB:type=connector-metrics,context=snapshot,server=<topic.prefix>.

The following table lists the JMX metrics that are available for monitoring Debezium snapshot operations, including row counts, table progress, duration, and queue capacity. Snapshot metrics are not exposed unless a snapshot operation is active, or a snapshot has occurred since the last connector start.

The following table lists the additional JMX metrics that are available when a connector runs an incremental snapshot, including chunk and table boundary identifiers that you can use to track snapshot progress.

Streaming metrics

The MBean is debezium.CockroachDB:type=connector-metrics,context=streaming,server=<topic.prefix>.

The following table lists the JMX metrics that are available for monitoring Debezium streaming operations, including event counts by type, lag behind the source, queue capacity, and connection status.

Connector configuration properties

The Debezium CockroachDB connector has many configuration properties that you can use to achieve the right connector behavior for your application. Many properties have default values. Information about the properties is organized as follows:

• Required configuration properties
• Connection configuration properties
• Changefeed configuration properties
• Connector configuration properties
• Advanced configuration properties

The following configuration properties are required unless a default value is available.

Table 7. Required connector configuration properties

The following properties configure the JDBC connection to the CockroachDB database.

Table 8. Connection configuration properties

The following properties configure the CockroachDB changefeed behavior.

Table 9. Changefeed configuration properties

Set the following properties to configure the connector’s snapshot and schema behavior.

Table 10. Connector configuration properties

The following advanced configuration properties have defaults that work in most situations and therefore rarely need to be specified in the connector’s configuration.

Table 11. Advanced connector configuration properties

Behavior when things go wrong

Debezium is a distributed system that captures all changes in multiple upstream databases; it never misses or loses an event. When the system is operating normally and is managed carefully, Debezium provides exactly once delivery of every change event record.

If a fault occurs, the system is designed to prevent the loss of any events. However, while it is recovering from a fault, it might repeat some change events. In these abnormal situations, Debezium, like Kafka, provides at least once delivery of change events.

The rest of this section describes how Debezium handles various kinds of faults and problems.

Configuration and startup errors

In the following situations, the connector fails when trying to start, reports an error or exception in the log, and then stops running:

• The connector’s configuration is invalid.
• The connector cannot successfully connect to CockroachDB by using the specified connection parameters.
• The connector cannot create a changefeed because the user lacks sufficient privileges (CHANGEFEED, ALL, or admin role membership).

In these cases, the error message has details about the problem and possibly a suggested workaround. After you correct the configuration or address the CockroachDB problem, restart the connector.

CockroachDB becomes unavailable

When the connector is running, the CockroachDB cluster could become unavailable for any number of reasons. Because CockroachDB is a distributed database, the cluster automatically handles individual node failures. If the entire cluster becomes unavailable, the changefeed stops producing events. After the cluster recovers, the connector resumes processing from the last stored offset.

The connector includes retry logic with configurable parameters (connection.max.retries, connection.retry.delay.ms) for handling transient connection failures, including CockroachDB-specific errors like serialization failures (SQL state 40001) and connection errors (SQL state 08xxx).

Kafka Connect process stops gracefully

Suppose that Kafka Connect is being run in distributed mode and a Kafka Connect process is stopped gracefully. Prior to shutting down that process, Kafka Connect migrates the process’s connector tasks to another Kafka Connect process in that group. The new connector tasks start processing exactly where the prior tasks stopped. There is a short delay in processing while the connector tasks are stopped gracefully and restarted on the new processes.

Kafka Connect process crashes

If the Kafka Connector process stops unexpectedly, any connector tasks it was running terminates without recording the most recently processed offsets. When Kafka Connect is being run in distributed mode, Kafka Connect restarts those connector tasks on other processes. However, CockroachDB connectors resume from the last offset that was recorded by the earlier processes. This means that the new replacement tasks might generate some of the same change events that were processed just prior to the crash. The number of duplicate events depends on the offset flush period and the volume of data changes just before the crash.

Because there is a chance that some events might be duplicated during a recovery from failure, consumers should always anticipate some duplicate events.

Kafka becomes unavailable

As the connector generates change events, the Kafka Connect framework records those events in Kafka by using the Kafka producer API. Periodically, at a frequency that you specify in the Kafka Connect configuration, Kafka Connect records the latest offset that appears in those change events. If the Kafka brokers become unavailable, the Kafka Connect process that is running the connectors repeatedly tries to reconnect to the Kafka brokers. In other words, the connector tasks pause until a connection can be re-established, at which point the connectors resume exactly where they left off.

Connector is stopped for an extended period

If the connector is gracefully stopped, the database can continue to be used. When the connector restarts, it resumes streaming changes where it left off. That is, it generates change event records for all database changes that were made while the connector was stopped.

The CockroachDB garbage collection TTL (default: 4 hours) limits how far back in time a changefeed can be started. If the connector is stopped for longer than the configured TTL interval, the stored offset may no longer be valid. In this case, use snapshot.mode=when_needed to automatically re-snapshot when the offset has expired.

Enabling debug logging

To diagnose connector behavior, enable DEBUG (or TRACE) logging for one or more of the connector’s loggers in the Kafka Connect worker’s logging configuration. The connector uses SLF4J, so the standard connect-log4j.properties mechanism (or the runtime /admin/loggers REST endpoint) controls the level.

The following table list the most useful logger names for triage.

For example, to enable schema-discovery and streaming traces, add the following to connect-log4j.properties:

log4j.logger.io.debezium.connector.cockroachdb.CockroachDBSchema=DEBUG
log4j.logger.io.debezium.connector.cockroachdb.CockroachDBStreamingChangeEventSource=DEBUG

Or set the levels at runtime against a running Kafka Connect worker:

curl -X PUT -H "Content-Type: application/json" \
    --data '{"level":"DEBUG"}' \
    http://localhost:8083/admin/loggers/io.debezium.connector.cockroachdb

Limitations

Kafka sink only

The connector currently only supports Kafka as the intermediate changefeed sink. Support for webhook, Pub/Sub, and cloud storage sinks is planned (see debezium/dbz#1632).

Before state in updates

By default, CockroachDB changefeeds do not include the previous row state. To include the before field in update events, set cockroachdb.changefeed.include.diff=true to enable the diff changefeed option.