Changefeed Examples

Create a changefeed connected to Kafka

In this example, you'll set up a changefeed for a single-node cluster that is connected to a Kafka sink. The changefeed will watch two tables.

1. If you do not already have one, request a trial Enterprise license.

2. Use the cockroach start-single-node command to start a single-node cluster:

   $ cockroach start-single-node --insecure --listen-addr=localhost --background
   

3. Download and extract the Confluent Open Source platform (which includes Kafka).

4. Move into the extracted confluent-<version> directory and start Confluent:

   $ ./bin/confluent local services start
   

   Only zookeeper and kafka are needed. To troubleshoot Confluent, see their docs and the Quick Start Guide.

5. Create two Kafka topics:

   $ ./bin/kafka-topics \
   --create \
   --zookeeper localhost:2181 \
   --replication-factor 1 \
   --partitions 1 \
   --topic office_dogs
   

   $ ./bin/kafka-topics \
   --create \
   --zookeeper localhost:2181 \
   --replication-factor 1 \
   --partitions 1 \
   --topic employees
   

   Note:

   You are expected to create any Kafka topics with the necessary number of replications and partitions. Topics can be created manually or Kafka brokers can be configured to automatically create topics with a default partition count and replication factor.

6. As the root user, open the built-in SQL client:

   $ cockroach sql --insecure
   

7. Set your organization name and Enterprise license key that you received via email:

   > SET CLUSTER SETTING cluster.organization = '<organization name>';
   

   > SET CLUSTER SETTING enterprise.license = '<secret>';
   

8. Enable the kv.rangefeed.enabled cluster setting:

   > SET CLUSTER SETTING kv.rangefeed.enabled = true;
   

9. Create a database called cdc_demo:

   > CREATE DATABASE cdc_demo;
   

10. Set the database as the default:

    > SET DATABASE = cdc_demo;
    

11. Create a table and add data:

    > CREATE TABLE office_dogs (
         id INT PRIMARY KEY,
         name STRING);
    

    > INSERT INTO office_dogs VALUES
       (1, 'Petee'),
       (2, 'Carl');
    

    > UPDATE office_dogs SET name = 'Petee H' WHERE id = 1;
    

12. Create another table and add data:

    > CREATE TABLE employees (
         dog_id INT REFERENCES office_dogs (id),
         employee_name STRING);
    

    > INSERT INTO employees VALUES
       (1, 'Lauren'),
       (2, 'Spencer');
    

13. Start the changefeed:

    > CREATE CHANGEFEED FOR TABLE office_dogs, employees INTO 'kafka://localhost:9092';
    

    
            job_id       
    +--------------------+
      360645287206223873
    (1 row)
    

    This will start up the changefeed in the background and return the job_id. The changefeed writes to Kafka.

14. In a new terminal, move into the extracted confluent-<version> directory and start watching the Kafka topics:

    $ ./bin/kafka-console-consumer \
    --bootstrap-server=localhost:9092 \
    --from-beginning \
    --whitelist 'office_dogs|employees'
    

    {"after": {"id": 1, "name": "Petee H"}}
    {"after": {"id": 2, "name": "Carl"}}
    {"after": {"id": 1, "name": "Lauren", "rowid": 528514320239329281}}
    {"after": {"id": 2, "name": "Spencer", "rowid": 528514320239362049}}
    

    The initial scan displays the state of the tables as of when the changefeed started (therefore, the initial value of "Petee" is omitted).

    Note:

    This example only prints the value. To print both the key and value of each message in the changefeed (e.g., to observe what happens with DELETEs), use the --property print.key=true flag.

15. Back in the SQL client, insert more data:

    > INSERT INTO office_dogs VALUES (3, 'Ernie');
    

16. Back in the terminal where you're watching the Kafka topics, the following output has appeared:

    {"after": {"id": 3, "name": "Ernie"}}
    

17. When you are done, exit the SQL shell (\q).

18. To stop cockroach:

    Get the process ID of the node:

    ps -ef | grep cockroach | grep -v grep
    

      501 21766     1   0  6:21PM ttys001    0:00.89 cockroach start-single-node --insecure --listen-addr=localhost
    

    Gracefully shut down the node, specifying its process ID:

    kill -TERM 21766
    

    initiating graceful shutdown of server
    server drained and shutdown completed
    

19. To stop Kafka, move into the extracted confluent-<version> directory and stop Confluent:

    $ ./bin/confluent local services stop
    

Create a changefeed connected to Kafka using Avro

In this example, you'll set up a changefeed for a single-node cluster that is connected to a Kafka sink and emits Avro records. The changefeed will watch two tables.

1. If you do not already have one, request a trial Enterprise license.

2. Use the cockroach start-single-node command to start a single-node cluster:

   $ cockroach start-single-node --insecure --listen-addr=localhost --background
   

3. Download and extract the Confluent Open Source platform (which includes Kafka).

4. Move into the extracted confluent-<version> directory and start Confluent:

   $ ./bin/confluent local services start
   

   Only zookeeper, kafka, and schema-registry are needed. To troubleshoot Confluent, see their docs and the Quick Start Guide.

5. Create two Kafka topics:

   $ ./bin/kafka-topics \
   --create \
   --zookeeper localhost:2181 \
   --replication-factor 1 \
   --partitions 1 \
   --topic office_dogs
   

   $ ./bin/kafka-topics \
   --create \
   --zookeeper localhost:2181 \
   --replication-factor 1 \
   --partitions 1 \
   --topic employees
   

   Note:

   You are expected to create any Kafka topics with the necessary number of replications and partitions. Topics can be created manually or Kafka brokers can be configured to automatically create topics with a default partition count and replication factor.

6. As the root user, open the built-in SQL client:

   $ cockroach sql --insecure
   

7. Set your organization name and Enterprise license key that you received via email:

   > SET CLUSTER SETTING cluster.organization = '<organization name>';
   

   > SET CLUSTER SETTING enterprise.license = '<secret>';
   

8. Enable the kv.rangefeed.enabled cluster setting:

   > SET CLUSTER SETTING kv.rangefeed.enabled = true;
   

9. Create a database called cdc_demo:

   > CREATE DATABASE cdc_demo;
   

10. Set the database as the default:

    > SET DATABASE = cdc_demo;
    

11. Create a table and add data:

    > CREATE TABLE office_dogs (
         id INT PRIMARY KEY,
         name STRING);
    

    > INSERT INTO office_dogs VALUES
       (1, 'Petee'),
       (2, 'Carl');
    

    > UPDATE office_dogs SET name = 'Petee H' WHERE id = 1;
    

12. Create another table and add data:

    > CREATE TABLE employees (
         dog_id INT REFERENCES office_dogs (id),
         employee_name STRING);
    

    > INSERT INTO employees VALUES
       (1, 'Lauren'),
       (2, 'Spencer');
    

13. Start the changefeed:

    > CREATE CHANGEFEED FOR TABLE office_dogs, employees INTO 'kafka://localhost:9092' WITH format = avro, confluent_schema_registry = 'http://localhost:8081';
    

    To connect to Confluent Cloud, use the following URL structure: 'https://{API_KEY_ID}:{API_SECRET_URL_ENCODED}@{CONFLUENT_REGISTRY_URL}:443'. See the Confluent Cloud Schema Registry Tutorial for further detail.

            job_id       
    +--------------------+
      360645287206223873
    (1 row)
    

    This will start up the changefeed in the background and return the job_id. The changefeed writes to Kafka.

14. In a new terminal, move into the extracted confluent-<version> directory and start watching the Kafka topics:

    $ ./bin/kafka-avro-console-consumer \
    --bootstrap-server=localhost:9092 \
    --from-beginning \
    --whitelist 'office_dogs|employees'
    

    {"after":{"office_dogs":{"id":{"long":1},"name":{"string":"Petee H"}}}}
    {"after":{"office_dogs":{"id":{"long":2},"name":{"string":"Carl"}}}}
    {"after":{"employees":{"dog_id":{"long":1},"employee_name":{"string":"Lauren"},"rowid":{"long":528537452042682369}}}}
    {"after":{"employees":{"dog_id":{"long":2},"employee_name":{"string":"Spencer"},"rowid":{"long":528537452042747905}}}}
    

    The initial scan displays the state of the table as of when the changefeed started (therefore, the initial value of "Petee" is omitted).

    Note:

    This example only prints the value. To print both the key and value of each message in the changefeed (e.g., to observe what happens with DELETEs), use the --property print.key=true flag.

15. Back in the SQL client, insert more data:

    > INSERT INTO office_dogs VALUES (3, 'Ernie');
    

16. Back in the terminal where you're watching the Kafka topics, the following output has appeared:

    {"after":{"office_dogs":{"id":{"long":3},"name":{"string":"Ernie"}}}}
    

17. When you are done, exit the SQL shell (\q).

18. To stop cockroach:

    Get the process ID of the node:

    ps -ef | grep cockroach | grep -v grep
    

      501 21766     1   0  6:21PM ttys001    0:00.89 cockroach start-single-node --insecure --listen-addr=localhost
    

    Gracefully shut down the node, specifying its process ID:

    kill -TERM 21766
    

    initiating graceful shutdown of server
    server drained and shutdown completed
    

19. To stop Kafka, move into the extracted confluent-<version> directory and stop Confluent:

    $ ./bin/confluent local services stop
    

Create a changefeed connected to a cloud storage sink

In this example, you'll set up a changefeed for a single-node cluster that is connected to an AWS S3 sink. The changefeed watches two tables. Note that you can set up changefeeds for any of these cloud storage providers.

1. If you do not already have one, request a trial Enterprise license.

2. Use the cockroach start-single-node command to start a single-node cluster:

   $ cockroach start-single-node --insecure --listen-addr=localhost --background
   

3. As the root user, open the built-in SQL client:

   $ cockroach sql --insecure
   

4. Set your organization name and Enterprise license key that you received via email:

   > SET CLUSTER SETTING cluster.organization = '<organization name>';
   

   > SET CLUSTER SETTING enterprise.license = '<secret>';
   

5. Enable the kv.rangefeed.enabled cluster setting:

   > SET CLUSTER SETTING kv.rangefeed.enabled = true;
   

6. Create a database called cdc_demo:

   > CREATE DATABASE cdc_demo;
   

7. Set the database as the default:

   > SET DATABASE = cdc_demo;
   

8. Create a table and add data:

   > CREATE TABLE office_dogs (
        id INT PRIMARY KEY,
        name STRING);
   

   > INSERT INTO office_dogs VALUES
      (1, 'Petee'),
      (2, 'Carl');
   

   > UPDATE office_dogs SET name = 'Petee H' WHERE id = 1;
   

9. Create another table and add data:

   > CREATE TABLE employees (
        dog_id INT REFERENCES office_dogs (id),
        employee_name STRING);
   

   > INSERT INTO employees VALUES
      (1, 'Lauren'),
      (2, 'Spencer');
   

10. Start the changefeed:

    > CREATE CHANGEFEED FOR TABLE office_dogs, employees INTO 's3://example-bucket-name/test?AWS_ACCESS_KEY_ID=enter_key-here&AWS_SECRET_ACCESS_KEY=enter_key_here' with updated, resolved='10s';
    

            job_id       
    +--------------------+
      360645287206223873
    (1 row)
    

    This will start up the changefeed in the background and return the job_id. The changefeed writes to AWS.

11. Monitor your changefeed on the DB Console. For more information, see Changefeeds Dashboard.

12. When you are done, exit the SQL shell (\q).

13. To stop cockroach:

    Get the process ID of the node:

    ps -ef | grep cockroach | grep -v grep
    

      501 21766     1   0  6:21PM ttys001    0:00.89 cockroach start-single-node --insecure --listen-addr=localhost
    

    Gracefully shut down the node, specifying its process ID:

    kill -TERM 21766
    

    initiating graceful shutdown of server
    server drained and shutdown completed
    

Create a changefeed connected to a webhook sink

New in v21.2: In this example, you'll set up a changefeed for a single-node cluster that is connected to a local HTTP server via a webhook. For this example, you'll use an example HTTP server to test out the webhook sink.

1. If you do not already have one, request a trial Enterprise license.

2. Use the cockroach start-single-node command to start a single-node cluster:

   $ cockroach start-single-node --insecure --listen-addr=localhost --background
   

3. In this example, you'll run CockroachDB's Movr application workload to set up some data for your changefeed.

   First create the schema for the workload:

    cockroach workload init movr "postgresql://root@127.0.0.1:26257?sslmode=disable"
   

   Then run the workload:

    cockroach workload run movr --duration=1m "postgresql://root@127.0.0.1:26257?sslmode=disable"
   

4. As the root user, open the built-in SQL client:

   $ cockroach sql --insecure
   

5. Set your organization name and Enterprise license key that you received via email:

   > SET CLUSTER SETTING cluster.organization = '<organization name>';
   

   > SET CLUSTER SETTING enterprise.license = '<secret>';
   

6. Enable the kv.rangefeed.enabled cluster setting:

   > SET CLUSTER SETTING kv.rangefeed.enabled = true;
   

7. In a separate terminal window, set up your HTTP server. Clone the test repository:

   git clone https://github.com/cockroachlabs/cdc-webhook-sink-test-server.git
   

   cd cdc-webhook-sink-test-server/go-https-server
   

8. Next make the script executable and then run the server (passing a specific port if preferred, otherwise it will default to :3000):

   chmod +x ./server.sh
   

   ./server.sh <port>
   

9. Back in your SQL shell, run the following statement to create a changefeed that emits to your webhook sink:

   CREATE CHANGEFEED FOR TABLE movr.vehicles INTO 'webhook-https://localhost:3000?insecure_tls_skip_verify=true' WITH updated;
   

   You set up a changefeed on the vehicles table, which emits changefeed messages to the local HTTP server.

   See the options table for more information on the options available for creating your changefeed to a webhook sink.

         job_id
   ----------------------
   687842491801632769
   (1 row)
   

   In the terminal where your HTTP server is running, you'll receive output similar to:

   2021/08/24 14:00:21 {"payload":[{"after":{"city":"rome","creation_time":"2019-01-02T03:04:05","current_location":"39141 Travis Curve Suite 87","ext":{"brand":"Schwinn","color":"red"},"id":"d7b18299-c0c4-4304-9ef7-05ae46fd5ee1","owner_id":"5d0c85b5-8866-47cf-a6bc-d032f198e48f","status":"in_use","type":"bike"},"key":["rome","d7b18299-c0c4-4304-9ef7-05ae46fd5ee1"],"topic":"vehicles","updated":"1629813621680097993.0000000000"}],"length":1}
   2021/08/24 14:00:22 {"payload":[{"after":{"city":"san francisco","creation_time":"2019-01-02T03:04:05","current_location":"84888 Wallace Wall","ext":{"color":"black"},"id":"020cf7f4-6324-48a0-9f74-6c9010fb1ab4","owner_id":"b74ea421-fcaf-4d80-9dcc-d222d49bdc17","status":"available","type":"scooter"},"key":["san francisco","020cf7f4-6324-48a0-9f74-6c9010fb1ab4"],"topic":"vehicles","updated":"1629813621680097993.0000000000"}],"length":1}
   2021/08/24 14:00:22 {"payload":[{"after":{"city":"san francisco","creation_time":"2019-01-02T03:04:05","current_location":"3893 Dunn Fall Apt. 11","ext":{"color":"black"},"id":"21b2ec54-81ad-4af7-a76d-6087b9c7f0f8","owner_id":"8924c3af-ea6e-4e7e-b2c8-2e318f973393","status":"lost","type":"scooter"},"key":["san francisco","21b2ec54-81ad-4af7-a76d-6087b9c7f0f8"],"topic":"vehicles","updated":"1629813621680097993.0000000000"}],"length":1}
   

   For more detail on emitted changefeed messages, see responses.

Core changefeeds stream row-level changes to a client until the underlying SQL connection is closed.

Create a Core changefeed

In this example, you'll set up a core changefeed for a single-node cluster.

1. In a terminal window, start cockroach:

   $ cockroach start-single-node \
   --insecure \
   --listen-addr=localhost \
   --background
   

2. As the root user, open the built-in SQL client:

   $ cockroach sql \
   --url="postgresql://root@127.0.0.1:26257?sslmode=disable" \
   --format=csv
   

   Note:

   Because core changefeeds return results differently than other SQL statements, they require a dedicated database connection with specific settings around result buffering. In normal operation, CockroachDB improves performance by buffering results server-side before returning them to a client; however, result buffering is automatically turned off for core changefeeds. Core changefeeds also have different cancellation behavior than other queries: they can only be canceled by closing the underlying connection or issuing a CANCEL QUERY statement on a separate connection. Combined, these attributes of changefeeds mean that applications should explicitly create dedicated connections to consume changefeed data, instead of using a connection pool as most client drivers do by default.

   Note:

   To determine how wide the columns need to be, the default table display format in cockroach sql buffers the results it receives from the server before printing them to the console. When consuming core changefeed data using cockroach sql, it's important to use a display format like csv that does not buffer its results. To set the display format, use the --format=csv flag when starting the built-in SQL client, or set the \set display_format=csv option once the SQL client is open.

3. Enable the kv.rangefeed.enabled cluster setting:

   > SET CLUSTER SETTING kv.rangefeed.enabled = true;
   

4. Create table foo:

   > CREATE TABLE foo (a INT PRIMARY KEY);
   

5. Insert a row into the table:

   > INSERT INTO foo VALUES (0);
   

6. Start the core changefeed:

   > EXPERIMENTAL CHANGEFEED FOR foo;
   

   table,key,value
   foo,[0],"{""after"": {""a"": 0}}"
   

7. In a new terminal, add another row:

   $ cockroach sql --insecure -e "INSERT INTO foo VALUES (1)"
   

8. Back in the terminal where the core changefeed is streaming, the following output has appeared:

   foo,[1],"{""after"": {""a"": 1}}"
   

   Note that records may take a couple of seconds to display in the core changefeed.

9. To stop streaming the changefeed, enter CTRL+C into the terminal where the changefeed is running.

10. To stop cockroach:

    Get the process ID of the node:

    ps -ef | grep cockroach | grep -v grep
    

      501 21766     1   0  6:21PM ttys001    0:00.89 cockroach start-single-node --insecure --listen-addr=localhost
    

    Gracefully shut down the node, specifying its process ID:

    kill -TERM 21766
    

    initiating graceful shutdown of server
    server drained and shutdown completed
    

Create a Core changefeed using Avro

In this example, you'll set up a core changefeed for a single-node cluster that emits Avro records. CockroachDB's Avro binary encoding convention uses the Confluent Schema Registry to store Avro schemas.

1. Use the cockroach start-single-node command to start a single-node cluster:

   $ cockroach start-single-node \
   --insecure \
   --listen-addr=localhost \
   --background
   

2. Download and extract the Confluent Open Source platform.

3. Move into the extracted confluent-<version> directory and start Confluent:

   $ ./bin/confluent local services start
   

   Only zookeeper, kafka, and schema-registry are needed. To troubleshoot Confluent, see their docs and the Quick Start Guide.

4. As the root user, open the built-in SQL client:

   $ cockroach sql --url="postgresql://root@127.0.0.1:26257?sslmode=disable" --format=csv
   

   Note:

   Because core changefeeds return results differently than other SQL statements, they require a dedicated database connection with specific settings around result buffering. In normal operation, CockroachDB improves performance by buffering results server-side before returning them to a client; however, result buffering is automatically turned off for core changefeeds. Core changefeeds also have different cancellation behavior than other queries: they can only be canceled by closing the underlying connection or issuing a CANCEL QUERY statement on a separate connection. Combined, these attributes of changefeeds mean that applications should explicitly create dedicated connections to consume changefeed data, instead of using a connection pool as most client drivers do by default.

   Note:

   To determine how wide the columns need to be, the default table display format in cockroach sql buffers the results it receives from the server before printing them to the console. When consuming core changefeed data using cockroach sql, it's important to use a display format like csv that does not buffer its results. To set the display format, use the --format=csv flag when starting the built-in SQL client, or set the \set display_format=csv option once the SQL client is open.

5. Enable the kv.rangefeed.enabled cluster setting:

   > SET CLUSTER SETTING kv.rangefeed.enabled = true;
   

6. Create table bar:

   > CREATE TABLE bar (a INT PRIMARY KEY);
   

7. Insert a row into the table:

   > INSERT INTO bar VALUES (0);
   

8. Start the core changefeed:

   > EXPERIMENTAL CHANGEFEED FOR bar WITH format = avro, confluent_schema_registry = 'http://localhost:8081';
   

   table,key,value
   bar,\000\000\000\000\001\002\000,\000\000\000\000\002\002\002\000
   

9. In a new terminal, add another row:

   $ cockroach sql --insecure -e "INSERT INTO bar VALUES (1)"
   

10. Back in the terminal where the core changefeed is streaming, the output will appear:

    bar,\000\000\000\000\001\002\002,\000\000\000\000\002\002\002\002
    

    Note that records may take a couple of seconds to display in the core changefeed.

11. To stop streaming the changefeed, enter CTRL+C into the terminal where the changefeed is running.

12. To stop cockroach:

    Get the process ID of the node:

    ps -ef | grep cockroach | grep -v grep
    

      501 21766     1   0  6:21PM ttys001    0:00.89 cockroach start-single-node --insecure --listen-addr=localhost
    

    Gracefully shut down the node, specifying its process ID:

    kill -TERM 21766
    

    initiating graceful shutdown of server
    server drained and shutdown completed
    

13. To stop Confluent, move into the extracted confluent-<version> directory and stop Confluent:

    $ ./bin/confluent local services stop
    

    To terminate all Confluent processes, use:

    $ ./bin/confluent local destroy
    

For further information on Core changefeeds, see EXPERIMENTAL CHANGEFEED FOR.