CockroachDB CDC: Confluent Kafka, Schema Registry - DZone Database

Previous Articles on CDC:

Motivation

I was working on a demo requiring data to be serialized in Avro format. CockroachDB supports Avro for change data capture when used in conjunction with Confluent Schema Registry. We have documentation on CDC with Avro using a local Schema Registry, but what is missing is hosted Schema Registry on Confluent Cloud. We are going to address that in the short term but in the meantime, this tutorial should suffice.

As of the time of this article, the hosted setup has nuances and I attempt to document it in the meantime.

This tutorial is using enterprise change feeds, you will need an enterprise license or access to a CockroachDB Dedicated environment.

High-Level Steps

• Deploy a Confluent Kafka cluster and Schema Registry
• Deploy a CockroachDB cluster with enterprise changefeeds
• Verify

Step-By-Step Instructions

Deploy a Confluent Kafka Cluster and Schema Registry

You will need a Confluent cloud account, you can sign up for a free account using the following link.

Once you're done, create a cluster, you can use the steps in my previous article ("SaaS Galore") linked at the beginning of this post.

export KAFKA_CLUSTER=<cluster ID>
confluent kafka cluster use $KAFKA_CLUSTER
confluent api-key create --resource $KAFKA_CLUSTER
confluent api-key store --resource $KAFKA_CLUSTER --force
confluent kafka cluster describe $KAFKA_CLUSTER

To capture the endpoint from the console, we will need it to set up a changefeed in CockroachDB.

SASL_SSL://<confluent cloud kafka endpoint url>:9092

To create topics for your tables, I'm going to select the following four tables: stock, history, warehouse, district.

confluent kafka topic create stock --partitions 6
confluent kafka topic create warehouse --partitions 6
confluent kafka topic create history --partitions 6
confluent kafka topic create district --partitions 6 

Create Confluent Schema Registry if it doesn't exist. Capture the endpoint, and generate an API key and secret for SR.

Finally, set up an Avro consumer with the above information in a new terminal window.

confluent api-key use <API Key> --resource $KAFKA_CLUSTER

confluent kafka topic consume stock \
 --value-format avro \
 --from-beginning \
 --sr-endpoint https://<Confluent Schema Registry url>.confluent.cloud \
 --sr-api-key <SR API Key> \
 --sr-api-secret <SR Secret>

Deploy a CockroachDB Cluster With Enterprise Changefeeds

You can spin up a Dedicated cluster using the following directions. My cluster is a 3-node cluster in GCP with AZ failure tolerance in us-east4.

Enable changefeeds.

SET CLUSTER SETTING kv.rangefeed.enabled = true;

Create a changefeed pointing to the Kafka cluster and SR.

CREATE CHANGEFEED FOR TABLE stock, warehouse, district, history INTO "kafka://<confluent kafka cluster endpoint>:9092?tls_enabled=true&sasl_enabled=true&sasl_user=<Confluent Kafka API Key>&sasl_password=<Confluent Kafka Secret in url-encoded form>&sasl_mechanism=PLAIN" WITH updated, format = avro, confluent_schema_registry = "https://<SR API Key>:<SR Secret in url-encoded form>@<Confluent SR url>:443";

The key part in the command is that the Confluent Schema Registry expects the Schema Registry API Key and Secret in the basic auth form, not SASL. Once you fill that out, you should be off to the races.

The only thing that remains is generating a workload. We are going to use the TPC-C workload bundled with the Cockroach binary. In a new terminal window, run the following two commands.

Generate sample data:

cockroach workload fixtures import tpcc --warehouses=10 "postgresql://<user>@<Cockroach Cloud Dedicated url>:26257/tpcc?sslmode=verify-full&sslrootcert=/path/certs/cluster-ca.crt"     

Execute the workload:

cockroach workload run tpcc --warehouses=10 --ramp=3m --duration=1h "postgresql://<user>@<Cockroach Cloud Dedicated url>:26257/tpcc?sslmode=verify-full&sslrootcert=/path/certs/cluster-ca.crt"

Verify

At this point, if you switch to the terminal window with the Avro consumer, you will see that it scrolls quickly because it's consuming change data from CockroachDB.

This completes this tutorial. I hope you found it useful.