Changefeed Sinks

On this page

Enterprise changefeeds emit messages to configurable downstream sinks. CockroachDB supports the following sinks:

• Kafka
• Cloud Storage / HTTP
• Webhook

See CREATE CHANGEFEED for more detail on the query parameters available when setting up a changefeed.

Sink URI

The sink URI follows the basic format of:

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

Kafka

Example of a Kafka sink URI:

'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=SASL-SCRAM-SHA-256'

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 a changefeed was created on multiple tables, the changefeed will write to multiple topics corresponding to those table names.

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

New in v21.2: 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" }'

"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:

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 only work with JSON and emit newline-delimited JSON files.
• 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.

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 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.

Webhook sink

New in v21.2: 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.
• Only supports JSON output format.
• There is no concurrency configurability.

Webhook sink configuration

New in v21.2: The webhook_sink_config option allows the changefeed flushing and retry behavior of your webhook sink to be configured.

The following details the configurable fields:

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." 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 defaults to 0 in this case, so a batch will never trigger due to a configured byte size:

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

See also

• Use Cloud Storage for Bulk Operations
• CREATE CHANGEFEED

Was this helpful?