Changefeed Sinks

On this page

Enterprise changefeeds emit messages to configurable downstream sinks. This page details the URIs, parameters, and configurations available for each changefeed sink.

CockroachDB supports the following sinks:

• Kafka
• New in v23.2: Confluent Cloud
• Google Cloud Pub/Sub
• Cloud Storage / HTTP
• Webhook

The CREATE CHANGEFEED page provides detail on using the SQL statement and a complete list of the query parameters and options available when setting up a changefeed.

For a step-by-step example connecting a changefeed to a sink, see the Changefeed Examples page.

Sink URI

The sink URI follows the basic format of:

'{scheme}://{host}:{port}?{query_parameters}'

You can create an external connection to represent a changefeed sink URI. This allows you to specify the external connection's name in statements rather than the provider-specific URI. For detail on using external connections, see the CREATE EXTERNAL CONNECTION page.

To set a different sink URI to an existing changefeed, use the sink option with ALTER CHANGEFEED.

Kafka

SET CLUSTER SETTING changefeed.new_kafka_sink.enabled = true;

Kafka sink connection

Example of a Kafka sink URI using SCRAM-SHA-256 authentication:

'kafka://broker.address.com:9092?topic_prefix=bar_&tls_enabled=true&ca_cert=LS0tLS1CRUdJTiBDRVJUSUZ&sasl_enabled=true&sasl_user={sasl user}&sasl_password={url-encoded password}&sasl_mechanism=SCRAM-SHA-256'

Example of a Kafka sink URI using OAUTHBEARER authentication:

'kafka://{kafka cluster address}:9093?topic_name={vehicles}&sasl_client_id={your client ID}&sasl_client_secret={your base64-encoded client secret}&sasl_enabled=true&sasl_mechanism=OAUTHBEARER&sasl_token_url={your token URL}'

OAuth 2.0 authentication uses credentials managed by a third-party provider (IdP) to authenticate with Kafka instead of requiring you to provide your Kafka cluster credentials directly in a CREATE CHANGEFEED statement. Your provider's authentication server will issue a temporary token, giving you flexibility to apply access rules on the credentials that your IdP provides.

To authenticate to Kafka with OAuth using Okta, see the Connect to a Changefeed Kafka sink with OAuth Using Okta tutorial.

The following table lists the available parameters for Kafka URIs:

This table shows the parameters for changefeeds to a specific sink. The CREATE CHANGEFEED page provides a list of all the available options.

Topic naming

By default, a Kafka topic has the same name as the table on which a changefeed was created. If you create a changefeed on multiple tables, the changefeed will write to multiple topics corresponding to those table names. When you run CREATE CHANGEFEED to a Kafka sink, the output will display the job ID as well as the topic name(s) that the changefeed will emit to.

To modify the default topic naming, you can specify a topic prefix, an arbitrary topic name, or use the full_table_name option. Using the topic_name parameter, you can specify an arbitrary topic name and feed all tables into that topic.

You can either manually create a topic in your Kafka cluster before starting the changefeed, or the topic will be automatically created when the changefeed connects to your Kafka cluster.

Kafka has the following topic limitations:

• Legal characters are numbers, letters, and [._-].
• The maximum character length of a topic name is 249.
• Topics with a period (.) and underscore (_) can collide on internal Kafka data structures, so you should use either but not both.
• Characters not accepted by Kafka will be automatically encoded as unicode characters by CockroachDB.

Kafka sink configuration

The kafka_sink_config option allows configuration of a changefeed's message delivery, Kafka server version, and batching parameters.

kafka_sink_config='{"Flush": {"MaxMessages": 1, "Frequency": "1s"}, "Version": "0.8.2.0", "RequiredAcks": "ONE", "Compression": "GZIP" }'

"Flush"."MaxMessages" and "Flush"."Frequency" are configurable batching parameters depending on latency and throughput needs. For example, if "MaxMessages" is set to 1000 and "Frequency" to 1 second, it will flush to Kafka either after 1 second or after 1000 messages are batched, whichever comes first. It's important to consider that if there are not many messages, then a "1s" frequency will add 1 second latency. However, if there is a larger influx of messages these will be flushed quicker.

Using the default values or not setting fields in kafka_sink_config will mean that changefeed messages emit immediately.

The configurable fields are as follows:

Kafka sink messages

The following shows the Avro messages for a changefeed emitting to Kafka:

{
    "after":{
       "users":{
          "name":{
             "string":"Michael Clark"
          },
          "address":{
             "string":"85957 Ashley Junctions"
          },
          "credit_card":{
             "string":"4144089313"
          },
          "id":{
             "string":"d84cf3b6-7029-4d4d-aa81-e5caa9cce09e"
          },
          "city":{
             "string":"seattle"
          }
       }
    },
    "updated":{
       "string":"1659643584586630201.0000000000"
    }
 }
 {
    "after":{
       "users":{
          "address":{
             "string":"17068 Christopher Isle"
          },
          "credit_card":{
             "string":"6664835435"
          },
          "id":{
             "string":"11b99275-92ce-4244-be61-4dae21973f87"
          },
          "city":{
             "string":"amsterdam"
          },
          "name":{
             "string":"John Soto"
          }
       }
    },
    "updated":{
       "string":"1659643585384406152.0000000000"
    }
 }

See the Changefeed Examples page and the Stream a Changefeed to a Confluent Cloud Kafka Cluster tutorial for examples to set up a Kafka sink.

For an overview of the messages emitted from changefeeds, see the Changefeed Messages page.

Confluent Cloud

New in v23.2: Changefeeds can deliver messages to Kafka clusters hosted on Confluent Cloud.

A Confluent Cloud sink connection URI must include the following:

'confluent-cloud://{bootstrap server}:9092?api_key={key}&api_secret={secret}'

The api_key and api_secret are the required parameters for the Confluent Cloud sink connection URI.

Changefeeds emitting to a Confluent Cloud Kafka cluster support the standard Kafka parameters, such as topic_name and topic_prefix. Confluent Cloud sinks also support the standard Kafka changefeed options and the Kafka sink configuration option.

For a Confluent Cloud setup example, refer to the Changefeed Examples page.

The following parameters are also needed, but are set by default in CockroachDB:

• tls_enabled=true
• sasl_enabled=true
• sasl_handshake=true
• sasl_mechanism=PLAIN

Google Cloud Pub/Sub

Changefeeds can deliver messages to a Google Cloud Pub/Sub sink, which is integrated with Google Cloud Platform.

A Pub/Sub sink URI follows this example:

'gcpubsub://{project name}?region={region}&topic_name={topic name}&AUTH=specified&CREDENTIALS={base64-encoded credentials}'

This table shows the parameters for changefeeds to a specific sink. The CREATE CHANGEFEED page provides a list of all the available options.

When using Pub/Sub as your downstream sink, consider the following:

• Pub/Sub sinks support JSON message format. You can use the format=csv option in combination with initial_scan='only' for CSV-formatted messages.
• Use the unordered option for multi-region Pub/Sub. Google Cloud's multi-region Pub/Sub will have lower latency when emitting from multiple regions, but Google Cloud Pub/Sub does not support message ordering for multi-region topics.
• Changefeeds connecting to a Pub/Sub sink do not support the topic_prefix option.

Ensure one of the following Pub/Sub roles are set in your Google Service Account at the project level:

• To create topics on changefeed creation, you must use the Pub/Sub Editor role, which contains the permissions to create a topic.
• If the topic the changefeed is writing to already exists, then you can use the more limited Pub/Sub Publisher role, which can only write to existing topics.

For more information, read about compatible changefeed options and the Create a changefeed connected to a Google Cloud Pub/Sub sink example.

Pub/Sub topic naming

When running a CREATE CHANGEFEED statement to a Pub/Sub sink, consider the following regarding topic names:

• Changefeeds will try to create a topic automatically. When you do not specify the topic in the URI with the topic_name parameter, the changefeed will use the table name to create the topic name.
• If the topic already exists in your Pub/Sub sink, the changefeed will write to it.
• Changefeeds watching multiple tables will write to multiple topics corresponding to those table names.
• The full_table_name option will create a topic using the fully qualified table name for each table the changefeed is watching.
• The output from CREATE CHANGEFEED will display the job ID as well as the topic name(s) to which the changefeed will emit.

You can manually create a topic in your Pub/Sub sink before starting the changefeed. Refer to the Creating a changefeed to Google Cloud Pub/Sub example for more detail. To understand restrictions on user-specified topic names, refer to Google's documentation on Guidelines to name a topic or subscription.

For a list of compatible parameters and options, refer to Parameters on the CREATE CHANGEFEED page.

Pub/Sub sink configuration

You can configure flushing, retry, and concurrency behavior of changefeeds running to a Pub/Sub sink with the following:

• Set the changefeed.sink_io_workers cluster setting to configure the number of concurrent workers used by changefeeds in the cluster when sending requests to a Pub/Sub sink. When you set changefeed.sink_io_workers, it will not affect running changefeeds; pause the changefeed, set changefeed.sink_io_workers, and then resume the changefeed. Note that this cluster setting will also affect changefeeds running to webhook sinks.
• Set the pubsub_sink_config option to configure the changefeed flushing and retry behavior to your webhook sink. For details on the pubsub_sink_config option's configurable fields, refer to the following table and examples.

For example:

pubsub_sink_config = '{ "Flush": {"Messages": 100, "Frequency": "5s"}, "Retry": { "Max": 4, "Backoff": "10s"} }'

Some complexities to consider when setting Flush fields for batching:

• When all batching parameters are zero ("Messages", "Bytes", and "Frequency") the sink will interpret this configuration as "send batch every time a message is available." This would be the same as not providing any configuration at all:

  {
    "Flush": {
      "Messages": 0,
      "Bytes": 0,
      "Frequency": "0s"
    }
  }
  

• If one or more fields are set as non-zero values, any fields with a zero value the sink will interpret as infinity. For example, in the following configuration, the sink will send a batch whenever the size reaches 100 messages, or, when 5 seconds has passed since the batch was populated with its first message. Bytes is unset, so the batch size is unlimited. No flush will be triggered due to batch size:

  {
    "Flush": {
      "Messages": 100,
      "Frequency": "5s"
    }
  }
  

Pub/Sub sink messages

New in v23.2: The changefeed.new_pubsub_sink_enabled cluster setting is enabled by default, which provides improved changefeed throughput peformance. With changefeed.new_pubsub_sink_enabled enabled, the changefeed JSON-encoded message format have top-level fields that are capitalized:

{Key: ..., Value: ..., Topic: ...}

With changefeed.new_pubsub_sink_enabled set to false, changefeeds emit JSON messages with the top-level fields all lowercase:

{key: ..., value: ..., topic: ...}

If changefeed.new_pubsub_sink_enabled is set to false, changefeeds will not benefit from the improved throughput performance that this setting enables.

The following shows the default JSON messages for a changefeed emitting to Pub/Sub. These changefeed messages were emitted as part of the Create a changefeed connected to a Google Cloud Pub/Sub sink example:

┌─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┬───────────────────┬──────────────┬────────────┬──────────────────┬────────────┐
│                                                                                                                                     DATA                                                                                                                                    │     MESSAGE_ID    │ ORDERING_KEY │ ATTRIBUTES │ DELIVERY_ATTEMPT │ ACK_STATUS │
├─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼───────────────────┼──────────────┼────────────┼──────────────────┼────────────┤
│ {"Key":["amsterdam", "09ee2856-5856-40c4-85d3-7d65bed978f0"],"Value":{"after": {"address": "84579 Peter Divide Apt. 47", "city": "amsterdam", "credit_card": "0100007510", "id": "09ee2856-5856-40c4-85d3-7d65bed978f0", "name": "Timothy Jackson"}},"Topic":"users"}       │ 11249015757941393 │              │            │                  │ SUCCESS    │
│ {"Key":["new york", "8803ab9e-5001-4994-a2e6-68d587f95f1d"],"Value":{"after": {"address": "37546 Andrew Roads Apt. 68", "city": "new york", "credit_card": "4731676650", "id": "8803ab9e-5001-4994-a2e6-68d587f95f1d", "name": "Susan Harrington"}},"Topic":"users"}        │ 11249015757941394 │              │            │                  │ SUCCESS    │
│ {"Key":["seattle", "32e27201-ca0d-4a0c-ada2-fbf47f6a4711"],"Value":{"after": {"address": "86725 Stephen Gardens", "city": "seattle", "credit_card": "3639690115", "id": "32e27201-ca0d-4a0c-ada2-fbf47f6a4711", "name": "Brad Hill"}},"Topic":"users"}                      │ 11249015757941395 │              │            │                  │ SUCCESS    │
│ {"Key":["san francisco", "27b03637-ef9f-49a0-9b58-b16d7a9e34f4"],"Value":{"after": {"address": "85467 Tiffany Field", "city": "san francisco", "credit_card": "0016125921", "id": "27b03637-ef9f-49a0-9b58-b16d7a9e34f4", "name": "Mark Garcia"}},"Topic":"users"}          │ 11249015757941396 │              │            │                  │ SUCCESS    │
│ {"Key":["rome", "982e1863-88d4-49cb-adee-0a35baae7e0b"],"Value":{"after": {"address": "54918 Sutton Isle Suite 74", "city": "rome", "credit_card": "6015706174", "id": "982e1863-88d4-49cb-adee-0a35baae7e0b", "name": "Kimberly Nichols"}},"Topic":"users"}                │ 11249015757941397 │              │            │                  │ SUCCESS    │
│ {"Key":["washington dc", "7b298994-7b12-414c-90ef-353c7105f012"],"Value":{"after": {"address": "45205 Romero Ford Apt. 86", "city": "washington dc", "credit_card": "3519400314", "id": "7b298994-7b12-414c-90ef-353c7105f012", "name": "Taylor Bullock"}},"Topic":"users"} │ 11249015757941398 │              │            │                  │ SUCCESS    │
│ {"Key":["boston", "4f012f57-577b-4853-b5ab-0d79d0df1369"],"Value":{"after": {"address": "15765 Vang Ramp", "city": "boston", "credit_card": "6747715133", "id": "4f012f57-577b-4853-b5ab-0d79d0df1369", "name": "Ryan Garcia"}},"Topic":"users"}                            │ 11249015757941399 │              │            │                  │ SUCCESS    │
│ {"Key":["seattle", "9ba85917-5545-4674-8ab2-497fa47ac00f"],"Value":{"after": {"address": "24354 Whitney Lodge", "city": "seattle", "credit_card": "8642661685", "id": "9ba85917-5545-4674-8ab2-497fa47ac00f", "name": "Donald Walsh"}},"Topic":"users"}                     │ 11249015757941400 │              │            │                  │ SUCCESS    │
│ {"Key":["seattle", "98312fb3-230e-412d-9b22-074ec97329ff"],"Value":{"after": {"address": "72777 Carol Shoal", "city": "seattle", "credit_card": "7789799678", "id": "98312fb3-230e-412d-9b22-074ec97329ff", "name": "Christopher Davis"}},"Topic":"users"}                  │ 11249015757941401 │              │            │                  │ SUCCESS    │
└─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┴───────────────────┴──────────────┴────────────┴──────────────────┴────────────┘

For an overview of the messages emitted from changefeeds, see the Changefeed Messages page.

Cloud storage sink

Use a cloud storage sink to deliver changefeed data to OLAP or big data systems without requiring transport via Kafka.

Some considerations when using cloud storage sinks:

• Cloud storage sinks work with JSON and emit newline-delimited JSON files. You can use the format=csv option in combination with initial_scan='only' for CSV-formatted messages.
• Cloud storage sinks can be configured to store emitted changefeed messages in one or more subdirectories organized by date. See file partitioning and the General file format examples.
• The supported cloud schemes are: s3, gs, azure, http, and https.
• Both http:// and https:// are cloud storage sinks, not webhook sinks. It is necessary to prefix the scheme with webhook- for webhook sinks.

You can authenticate to cloud storage sinks using specified or implicit authentication. CockroachDB also supports assume role authentication for Amazon S3 and Google Cloud Storage, which allows you to limit the control specific users have over your storage buckets. For detail and instructions on authenticating to cloud storage sinks, see Cloud Storage Authentication.

Examples of supported cloud storage sink URIs:

Amazon S3

's3://{BUCKET NAME}/{PATH}?AWS_ACCESS_KEY_ID={KEY ID}&AWS_SECRET_ACCESS_KEY={SECRET ACCESS KEY}'

Azure Blob Storage

'azure://{CONTAINER NAME}/{PATH}?AZURE_ACCOUNT_NAME={ACCOUNT NAME}&AZURE_ACCOUNT_KEY={URL-ENCODED KEY}'

Google Cloud Storage

'gs://{BUCKET NAME}/{PATH}?AUTH=specified&CREDENTIALS={ENCODED KEY}'

HTTP

'http://localhost:8080/{PATH}'

Cloud storage parameters

The following table lists the available parameters for cloud storage sink URIs:

This table shows the parameters for changefeeds to a specific sink. The CREATE CHANGEFEED page provides a list of all the available options.

Use Cloud Storage for Bulk Operations provides more detail on authentication to cloud storage sinks.

Cloud storage sink messages

The following shows the default JSON messages for a changefeed emitting to a cloud storage sink:

{
    "after":{
       "address":"51438 Janet Valleys",
       "city":"boston",
       "credit_card":"0904722368",
       "id":"33333333-3333-4400-8000-00000000000a",
       "name":"Daniel Hernandez MD"
    },
    "key":[
       "boston",
       "33333333-3333-4400-8000-00000000000a"
    ]
 }
 {
    "after":{
       "address":"15074 Richard Falls",
       "city":"boston",
       "credit_card":"0866384459",
       "id":"370117cf-d77d-4778-b0b9-01ac17c15a06",
       "name":"Cheyenne Morales"
    },
    "key":[
       "boston",
       "370117cf-d77d-4778-b0b9-01ac17c15a06"
    ]
 }
 {
    "after":{
       "address":"69687 Jessica Islands Apt. 68",
       "city":"boston",
       "credit_card":"6837062320",
       "id":"3851eb85-1eb8-4200-8000-00000000000b",
       "name":"Sarah Wang DDS"
    },
    "key":[
       "boston",
       "3851eb85-1eb8-4200-8000-00000000000b"
    ]
 }
. . .

For an overview of the messages emitted from changefeeds, see the Changefeed Messages page.

Webhook sink

Use a webhook sink to deliver changefeed messages to an arbitrary HTTP endpoint.

Example of a webhook sink URL:

'webhook-https://{your-webhook-endpoint}?insecure_tls_skip_verify=true'

The following table lists the parameters you can use in your webhook URI:

This table shows the parameters for changefeeds to a specific sink. The CREATE CHANGEFEED page provides a list of all the available options.

The following are considerations when using the webhook sink:

• Only supports HTTPS. Use the insecure_tls_skip_verify parameter when testing to disable certificate verification; however, this still requires HTTPS and certificates.
• Supports JSON output format. You can use the format=csv option in combination with initial_scan='only' for CSV-formatted messages.

Webhook sink configuration

You can configure flushing, retry, and concurrency behavior of changefeeds running to a webhook sink with the following:

• Set the changefeed.sink_io_workers cluster setting to configure the number of concurrent workers used by changefeeds in the cluster when sending requests to a webhook sink. When you set changefeed.sink_io_workers, it will not affect running changefeeds; pause the changefeed, set changefeed.sink_io_workers, and then resume the changefeed. Note that this cluster setting will also affect changefeeds running to Google Cloud Pub/Sub sinks.
• Set the webhook_sink_config option to configure the changefeed flushing and retry behavior to your webhook sink. For details on the webhook_sink_config option's configurable fields, refer to the following table and examples.

For example:

webhook_sink_config = '{ "Flush": {"Messages": 100, "Frequency": "5s"}, "Retry": { "Max": 4, "Backoff": "10s"} }'

Some complexities to consider when setting Flush fields for batching:

• When all batching parameters are zero ("Messages", "Bytes", and "Frequency") the sink will interpret this configuration as "send batch every time a message is available." This would be the same as not providing any configuration at all:

  {
    "Flush": {
      "Messages": 0,
      "Bytes": 0,
      "Frequency": "0s"
    }
  }
  

• If one or more fields are set as non-zero values, any fields with a zero value the sink will interpret as infinity. For example, in the following configuration, the sink will send a batch whenever the size reaches 100 messages, or, when 5 seconds has passed since the batch was populated with its first message. Bytes is unset, so the batch size is unlimited. No flush will be triggered due to batch size:

  {
    "Flush": {
      "Messages": 100,
      "Frequency": "5s"
    }
  }
  

Webhook sink messages

The following shows the default JSON messages for a changefeed emitting to a webhook sink. These changefeed messages were emitted as part of the Create a changefeed connected to a Webhook sink example:

"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",
             "dog_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",
             "dog_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
 }

For an overview of the messages emitted from changefeeds, see the Changefeed Messages page.

See also

• Use Cloud Storage
• CREATE CHANGEFEED

Was this helpful?