> ## Documentation Index > Fetch the complete documentation index at: https://docs.risingwave.com/llms.txt > Use this file to discover all available pages before exploring further. # Sink to Kafka > This topic describes how to sink data from RisingWave to a Kafka broker and how to specify security (encryption and authentication) settings. A sink is an external target that you can send data to. To stream data out of RisingWave, you need to create a sink. Use the `CREATE SINK` statement to create a sink. You can create a sink with data from a materialized view or a table. RisingWave only supports writing messages in non-transactional mode. ## Delivery semantics The Kafka sink provides **at-least-once delivery** guarantees. RisingWave will retry failed sends, which may result in duplicate messages being written to Kafka topics during failures or retries. * For upsert sinks, duplicates are naturally handled by Kafka consumers through primary key deduplication. * For append-only sinks, downstream consumers should handle potential duplicates. To enable idempotent producer behavior at the Kafka level (preventing duplicates due to broker acknowledgment issues), you can set `properties.enable.idempotence = true`. Note that this provides idempotence for individual sends but does not guarantee end-to-end exactly-once when combined with RisingWave's checkpoint mechanism. ## Syntax ```sql theme={null} CREATE SINK [ IF NOT EXISTS ] sink_name [FROM sink_from | AS select_query] WITH ( connector='kafka', connector_parameter = 'value', ... ) FORMAT data_format ENCODE data_encode [ ( key = 'value' ) ] [KEY ENCODE key_encode [(...)]] ; ``` Names and unquoted identifiers are case-insensitive. Therefore, you must double-quote any of these fields for them to be case-sensitive. See also [Identifiers](/sql/identifiers). ## Basic parameters All `WITH` options are required unless explicitly mentioned as optional. | Parameter or clause | Description | | :-------------------------- | :--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | sink\_name | Name of the sink to be created. | | sink\_from | A clause that specifies the direct source from which data will be output. `sink_from` can be a materialized view or a table. Either this clause or a SELECT query must be specified. | | AS select\_query | A SELECT query that specifies the data to be output to the sink. Either this query or a FROM clause must be specified. See [SELECT](/sql/commands/sql-select) for the syntax and examples of the SELECT command. | | connector | Sink connector type must be `kafka` for Kafka sink. | | properties.bootstrap.server | Address of the Kafka broker. Format: `ip:port`. If there are multiple brokers, separate them with commas. | | topic | Address of the Kafka topic. One sink can only correspond to one topic. | | primary\_key | **Conditional**. The primary keys of the sink. Use `,` to delimit the primary key columns. This field is optional if creating a `PLAIN` sink, but required if creating a `DEBEZIUM` or `UPSERT` sink. | ## Additional Kafka parameters When creating a Kafka sink in RisingWave, you can specify the following Kafka-specific parameters. To set the parameter, add the RisingWave equivalent of the Kafka parameter as a `WITH` option. For additional details on these parameters, see the [Configuration properties](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md). | Kafka parameter name | RisingWave parameter name | Type | | :------------------------------------ | :----------------------------------------------- | :----- | | allow\.auto.create.topics | properties.allow\.auto.create.topics | bool | | batch.num.messages | properties.batch.num.messages | int | | batch.size | properties.batch.size | int | | client.id | properties.client.id | string | | enable.idempotence | properties.enable.idempotence | bool | | enable.ssl.certificate.verification | properties.enable.ssl.certificate.verification | bool | | max.in.flight.requests.per.connection | properties.max.in.flight.requests.per.connection | int | | message.max.bytes | properties.message.max.bytes | int | | message.send.max.retries | properties.message.send.max.retries | int | | message.timeout.ms | properties.message.timeout.ms | int | | queue.buffering.max.kbytes | properties.queue.buffering.max.kbytes | int | | queue.buffering.max.messages | properties.queue.buffering.max.messages | int | | queue.buffering.max.ms | properties.queue.buffering.max.ms | float | | request.required.acks | properties.request.required.acks | int | | retry.backoff.ms | properties.retry.backoff.ms | int | | receive.message.max.bytes | properties.receive.message.max.bytes | int | | ssl.endpoint.identification.algorithm | properties.ssl.endpoint.identification.algorithm | str | | statistics.interval.ms | properties.statistics.interval.ms | int | Set `properties.ssl.endpoint.identification.algorithm` to `none` to bypass the verification of CA certificates and resolve SSL handshake failure. This parameter can be set to either `https` or `none`. By default, it is `https`. To monitor Kafka metrics in Grafana, set `properties.statistics.interval.ms` to a non-zero value. The granularity is 1000ms. Starting with version 2.0, the default value for `properties.message.timeout.ms` has changed from 5 seconds to **5 minutes**, aligning with the default setting in the [official Kafka library](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md). ## FORMAT and ENCODE options These options should be set in `FORMAT data_format ENCODE data_encode (key = 'value')`, instead of the `WITH` clause. | Field | Notes | | :------------------------ | :--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | data\_format | Data format. Allowed formats: To learn about when to define the primary key if creating an UPSERT sink, see the [Overview](/delivery/overview). | | data\_encode | Data encode. Allowed encodes: For `UPSERT PROTOBUF` sinks, you must specify `key encode text`, while it remains optional for other format/encode combinations. | | force\_append\_only | If true, forces the sink to be `PLAIN` (also known as append-only), even if it cannot be. | | timestamptz.handling.mode | Controls the timestamptz output format. This parameter specifically applies to append-only or upsert sinks using JSON encoding. | | schemas.enable | Only configurable for upsert JSON sinks. By default, this value is false for upsert JSON sinks and true for debezium JSON sinks. If true, RisingWave will sink the data with the schema to the Kafka sink. This is not referring to a schema registry containing a JSON schema, but rather schema formats defined using [Kafka Connect](https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained/#json-schemas). | | key\_encode | **Optional**. When specified, the key encode can only be TEXT or BYTES. If set to TEXT, the primary key should be one and only one of the following types: `varchar`, `bool`, `smallint`, `int`, and `bigint`; If set to BYTES, the primary key should be one and only one of type `bytea`; When absent, both key and value will use the same setting of `ENCODE data_encode ( ... )`. | ### Avro When creating an Avro sink, the following options can be used following `FORMAT UPSERT ENCODE AVRO` or `FORMAT PLAIN ENCODE AVRO`. | Field | Notes | | :---------------------------- | :--------------------------------------------------------------------------------------------------------------------------------- | | schema.registry | **Required**. The address of the schema registry. | | schema.registry.username | **Optional**. The user name used to access the schema registry. | | schema.registry.password | **Optional**. The password associated with the user name. | | schema.registry.name.strategy | **Optional**. Accepted options include topic\_name\_strategy (default), record\_name\_strategy, and topic\_record\_name\_strategy. | | key.message | Required if schema.registry.name.strategy is set to record\_name\_strategy or topic\_record\_name\_strategy. | | message | Required if schema.registry.name.strategy is set to record\_name\_strategy or topic\_record\_name\_strategy. | ```sql Syntax theme={null} FORMAT [ UPSERT | PLAIN ] ENCODE AVRO ( schema.registry = 'schema_registry_url', [schema.registry.username = 'username'], [schema.registry.password = 'password'], [schema.registry.name.strategy = 'topic_name_strategy'], [key.message = 'test_key'], [message = 'main_message',] ) ``` For data type mapping, the serial type is supported. We map the serial type to the 64-bit signed integer. #### Use AWS Glue Schema Registry **PREMIUM FEATURE** This is a premium feature. For a comprehensive overview of all premium features and their usage, please see [RisingWave premium features](/get-started/premium-features). AWS Glue Schema Registry is a serverless feature of AWS Glue that allows you to centrally manage and enforce schemas for data streams, enabling data validation and compatibility checks. It helps in improving the quality of data streams by providing a central repository for managing and enforcing schemas across various AWS services and custom applications. You can specify the following configurations in the `FORMAT PLAIN ENCODE AVRO (...)` or `FORMAT UPSERT ENCODE AVRO (...)` clause. This allows RisingWave to load schemas from and encode metadata for AWS Glue Schema Registry, in addition to Confluent Schema Registry. **Auth-related configurations**: | Parameter | Description | | :---------------------------------- | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `aws.region` | The region of the AWS Glue Schema Registry. For example, `us-west-2`. | | `aws.credentials.access_key_id` | Your AWS access key ID. | | `aws.credentials.secret_access_key` | Your AWS secret access key. | | `aws.credentials.role.arn` | The Amazon Resource Name (ARN) of the role to assume. For example, `arn:aws:iam::123456123456:role/MyGlueRole`. This IAM role shall be granted permissions for the action `glue:GetSchemaVersion`. | In RisingWave Cloud, AWS credentials (`aws.credentials.access_key_id`, `aws.credentials.secret_access_key`) are required and cannot be omitted. In self-hosted deployments, you may omit them to rely on the AWS SDK default credential chain (for example, EC2 instance profile or environment variables). **ARN to the schema**: | Parameter | Description | | :-------------------- | :-------------------------------------------------------------------------------------------------------------------------------------------- | | `aws.glue.schema_arn` | The ARN of the schema in AWS Glue Schema Registry. For example, `'arn:aws:glue:ap-southeast-1:123456123456:schema/default-registry/MyEvent'`. | ```sql Example theme={null} ENCODE AVRO ( aws.glue.schema_arn = 'arn:aws:glue:ap-southeast-1:123456123456:schema/default-registry/MyEvent', aws.region = 'US-WEST-2', aws.credentials.access_key_id = 'your_access_key_id', aws.credentials.secret_access_key = 'your_secret_access_key', aws.credentials.role.arn = 'arn:aws:iam::123456123456:role/MyGlueRole' ... ) [KEY ENCODE [TEXT | BYTES]] -- when specifying `primary_key` as Kafka key ``` ### Protobuf When creating an append-only Protobuf sink, the following options can be used following `FORMAT PLAIN ENCODE PROTOBUF` or `FORMAT UPSERT ENCODE PROTOBUF`. | Field | Notes | | :---------------------------- | :----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | message | **Required**. Package qualified message name of the main Message in the schema definition. | | schema.location | Required if schema.registry is not specified. Only one of schema.location or schema.registry can be defined. The schema location. This can be in either `file://`, `http://`, `https://` format. | | schema.registry | Required if schema.location is not specified. Only one of schema.location or schema.registry can be defined. The address of the schema registry. | | schema.registry.username | **Optional**. The user name used to access the schema registry. | | schema.registry.password | **Optional**. The password associated with the user name. | | schema.registry.name.strategy | **Optional**. Accepted options include topic\_name\_strategy (default), record\_name\_strategy, and topic\_record\_name\_strategy. | The `file://` format is not recommended for production use. If it is used, it needs to be available for both Meta and Compute Nodes. When using Protobuf with Confluent Schema Registry, RisingWave bundles only protobuf [well-known types](https://protobuf.dev/reference/protobuf/google.protobuf/), such as `google/protobuf/timestamp.proto`, `duration.proto`, `any.proto`, `empty.proto`, `struct.proto`, and wrappers. Other imports, including Google common types such as `google/type/date.proto`, `google/type/latlng.proto`, `google/rpc/*`, `google/api/*`, and `confluent/meta.proto`, must also be registered as schema references in Schema Registry so that RisingWave can encode them properly. ```sql FORMAT as PLAIN theme={null} FORMAT PLAIN ENCODE PROTOBUF ( message = 'com.example.MyMessage', schema.location = 'location' ) ``` ```sql FORMAT as UPSERT theme={null} FORMAT UPSERT ENCODE PROTOBUF ( message = 'com.example.MyMessage', schema.location = 'location' ) KEY ENCODE TEXT ``` For data type mapping, the serial type is supported. We map the serial type to the 64-bit signed integer. ### JSON The `jsonb.handling.mode` determines how `jsonb` data types are encoded. This parameter has two possible values: * `string`: Encodes the `jsonb` type to a string. For example, if you set this parameter, `{"k": 2}` will be converted to `"{\"k\": 2}"`. * `dynamic`: Dynamically encodes a `jsonb` type value to a JSON type value. For example, if you set this parameter, `{"k": 2}` will be converted to `{"k": 2}`. Here the `jsonb` value is encoded to a JSON object type value. You can set this parameter in the `WITH` clause of `ENCODE JSON`. For data mapping, the serial type is supported. However, note that it is mapped into a JSON string like `"0x05fb93d677c4e000"` instead of a JSON number `431100738685689856`. This string form avoids JSON number precision issues with large int64 values, and you can still order by the fixed-length hexadecimal string to obtain the same order as the serial number (whereas variable-length string `"12"` sorts before `"7"`). ### BYTES The BYTES encoding allows you to sink raw binary data directly to Kafka topics. This is useful for scenarios where you need to output unencoded binary payloads. **Requirements:** * Only supported with `FORMAT PLAIN` * The source table or materialized view must have exactly one column of type `bytea` * `force_append_only='true'` must be specified ```sql Example theme={null} CREATE TABLE t_binary_data ( data bytea ); CREATE SINK sink_bytes FROM t_binary_data WITH ( connector = 'kafka', properties.bootstrap.server = 'localhost:9092', topic = 'binary-topic' ) FORMAT PLAIN ENCODE BYTES ( force_append_only = 'true' ); ``` ## Examples Create a sink by selecting an entire materialized view. ```sql theme={null} CREATE SINK sink1 FROM mv1 WITH ( connector='kafka', properties.bootstrap.server='localhost:9092', topic='test' ) FORMAT PLAIN ENCODE JSON; ``` Create a sink with the Kafka configuration `message.max.bytes` set at 2000 by setting `properties.message.max.bytes` to 2000. ```sql theme={null} CREATE SINK sink1 FROM mv1 WITH ( connector='kafka', properties.bootstrap.server='localhost:9092', topic='test', properties.message.max.bytes = 2000 ) FORMAT PLAIN ENCODE JSON; ``` Create a sink by selecting the average `distance` and `duration` from `taxi_trips`. The schema of `taxi_trips` is like this: ```yaml theme={null} { "id": VARCHAR, "distance": DOUBLE PRECISION, "duration": DOUBLE PRECISION, "fare": DOUBLE PRECISION } ``` The table may look like this: ```sql theme={null} id | distance | duration | city ----+----------+----------+---------- 1 | 16 | 23 | Dallas 2 | 23 | 9 | New York 3 | 6 | 15 | Chicago 4 | 9 | 35 | New York ``` ```sql theme={null} CREATE SINK sink2 AS SELECT avg(distance) as avg_distance, avg(duration) as avg_duration FROM taxi_trips WITH ( connector='kafka', properties.bootstrap.server='localhost:9092', topic='test' ) FORMAT PLAIN ENCODE JSON; ``` ## Create sink with PrivateLink connection If your Kafka sink service is located in a different VPC from RisingWave, use AWS PrivateLink or GCP Private Service Connect to establish a secure and direct connection. For details on how to set up an AWS PrivateLink connection, see [Create an AWS PrivateLink connection](/sql/commands/sql-create-connection#create-an-aws-privatelink-connection). To create a Kafka sink with a PrivateLink connection, in the WITH section of your `CREATE SINK` statement, specify the following parameters. | Parameter | Notes | | :------------------- | :---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | privatelink.targets | The PrivateLink targets that correspond to the Kafka brokers. The targets should be in JSON format. Note that each target listed corresponds to each broker specified in the properties.bootstrap.server field. If the order is incorrect, there will be connectivity issues. | | privatelink.endpoint | The DNS name of the VPC endpoint. If you're using RisingWave Cloud, you can find the auto-generated endpoint after you created a connection. See details in [Create a VPC connection](/cloud/create-a-connection/#whats-next). | Here is an example of creating a Kafka sink using a PrivateLink connection. Notice that `{"port": 8001}` corresponds to the broker `ip1:9092`, and `{"port": 8002}` corresponds to the broker `ip2:9092`. ```sql theme={null} CREATE SINK sink2 FROM mv2 WITH ( connector='kafka', properties.bootstrap.server='b-1.xxx.amazonaws.com:9092,b-2.test.xxx.amazonaws.com:9092', topic='msk_topic', privatelink.endpoint='10.148.0.4', privatelink.targets = '[{"port": 8001}, {"port": 8002}]' ) FORMAT PLAIN ENCODE JSON ( force_append_only='true' ); ``` ## TLS/SSL encryption and SASL authentication RisingWave can sink data to Kafka that is encrypted with [Transport Layer Security (TLS)](https://en.wikipedia.org/wiki/Transport%5FLayer%5FSecurity) and/or authenticated with SASL. Secure Sockets Layer (SSL) was the predecessor of Transport Layer Security (TLS), and has been deprecated since June 2015. For historical reasons, `SSL` is used in configuration and code instead of `TLS`. Simple Authentication and Security Layer (SASL) is a framework for authentication and data security in Internet protocols. RisingWave supports these SASL authentication mechanisms: * `SASL/PLAIN` * `SASL/SCRAM` * `SASL/GSSAPI` * `SASL/OAUTHBEARER` SSL encryption can be used concurrently with SASL authentication mechanisms. To learn about how to enable SSL encryption and SASL authentication in Kafka, including how to generate the keys and certificates, see the [Security Tutorial](https://docs.confluent.io/platform/current/security/security%5Ftutorial.html#overview) from Confluent. You need to specify encryption and authentication parameters in the WITH section of a `CREATE SINK` statement. To sink data encrypted with SSL without SASL authentication, specify these parameters in the WITH section of your `CREATE SINK` statement. | Parameter | Notes | | :---------------------------------- | :---------- | | properties.security.protocol | Set to SSL. | | properties.ssl.ca.location | | | properties.ssl.certificate.location | | | properties.ssl.key.location | | | properties.ssl.key.password | | For the definitions of the parameters, see the [librdkafka properties list](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). Note that the parameters in the list assumes all parameters start with `properties.` and therefore do not include this prefix. Here is an example of creating a sink encrypted with SSL without using SASL authentication. ```sql theme={null} CREATE SINK sink1 FROM mv1 WITH ( connector='kafka', topic='quickstart-events', properties.bootstrap.server='localhost:9093', properties.security.protocol='SSL', properties.ssl.ca.location='/home/ubuntu/kafka/secrets/ca-cert', properties.ssl.certificate.location='/home/ubuntu/kafka/secrets/client_risingwave_client.pem', properties.ssl.key.location='/home/ubuntu/kafka/secrets/client_risingwave_client.key', properties.ssl.key.password='abcdefgh' ) FORMAT PLAIN ENCODE JSON; ``` | Parameter | Notes | | :--------------------------- | :--------------------------------------------------------------------------------------------- | | properties.security.protocol | For SASL/PLAIN without SSL, set to SASL\_PLAINTEXT. For SASL/PLAIN with SSL, set to SASL\_SSL. | | properties.sasl.mechanism | Set to PLAIN. | | properties.sasl.username | | | properties.sasl.password | | For the definitions of the parameters, see the [librdkafka properties list](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). Note that the parameters in the list assumes all parameters start with `properties.` and therefore do not include this prefix. For SASL/PLAIN with SSL, you need to include these SSL parameters: * `properties.ssl.ca.location` * `properties.ssl.certificate.location` * `properties.ssl.key.location` * `properties.ssl.key.password` Here is an example of creating a sink authenticated with SASL/PLAIN without SSL encryption. ```sql theme={null} CREATE SINK sink1 FROM mv1 WITH ( connector='kafka', topic='quickstart-events', properties.bootstrap.server='localhost:9093', properties.sasl.mechanism='PLAIN', properties.security.protocol='SASL_PLAINTEXT', properties.sasl.username='admin', properties.sasl.password='admin-secret' ) FORMAT PLAIN ENCODE JSON; ``` This is an example of creating a sink authenticated with SASL/PLAIN with SSL encryption. ```sql theme={null} CREATE SINK sink1 FROM mv1 WITH ( connector='kafka', topic='quickstart-events', properties.bootstrap.server='localhost:9093', properties.sasl.mechanism='PLAIN', properties.security.protocol='SASL_SSL', properties.sasl.username='admin', properties.sasl.password='admin-secret', properties.ssl.ca.location='/home/ubuntu/kafka/secrets/ca-cert', properties.ssl.certificate.location='/home/ubuntu/kafka/secrets/client_risingwave_client.pem', properties.ssl.key.location='/home/ubuntu/kafka/secrets/client_risingwave_client.key', properties.ssl.key.password='abcdefgh' ) FORMAT PLAIN ENCODE JSON; ``` | Parameter | Notes | | :--------------------------- | :--------------------------------------------------------------------------------------------- | | properties.security.protocol | For SASL/SCRAM without SSL, set to SASL\_PLAINTEXT. For SASL/SCRAM with SSL, set to SASL\_SSL. | | properties.sasl.mechanism | Set to SCRAM-SHA-256 or SCRAM-SHA-512 depending on the encryption method used. | | properties.sasl.username | | | properties.sasl.password | | For the definitions of the parameters, see the [librdkafka properties list](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). Note that the parameters in the list assumes all parameters start with `properties.` and therefore do not include this prefix. For SASL/SCRAM with SSL, you also need to include these SSL parameters: * properties.ssl.ca.location * properties.ssl.certificate.location * properties.ssl.key.location * properties.ssl.key.password Here is an example of creating a sink authenticated with SASL/SCRAM without SSL encryption. ```sql theme={null} CREATE SINK sink1 FROM mv1 WITH ( connector='kafka', topic='quickstart-events', properties.bootstrap.server='localhost:9093', properties.sasl.mechanism='SCRAM-SHA-256', properties.security.protocol='SASL_PLAINTEXT', properties.sasl.username='admin', properties.sasl.password='admin-secret' ) FORMAT PLAIN ENCODE JSON; ``` | Parameter | Notes | | :----------------------------------------------- | :--------------------------------------------------------------------------------- | | properties.security.protocol | Set to SASL\_PLAINTEXT, as RisingWave does not support using SASL/GSSAPI with SSL. | | properties.sasl.mechanism | Set to GSSAPI. | | properties.sasl.kerberos.service.name | | | properties.sasl.kerberos.keytab | | | properties.sasl.kerberos.principal | | | properties.sasl.kerberos.kinit.cmd | | | properties.sasl.kerberos.min.time.before.relogin | | For the definitions of the parameters, see the [librdkafka properties list](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). Note that the parameters in the list assumes all parameters start with `properties.` and therefore do not include this prefix. Here is an example of creating a sink authenticated with SASL/GSSAPI without SSL encryption. ```sql theme={null} CREATE SINK sink1 FROM mv1 WITH ( connector='kafka', type = 'append-only', topic='quickstart-events', properties.bootstrap.server='localhost:9093', properties.sasl.mechanism='GSSAPI', properties.security.protocol='SASL_PLAINTEXT', properties.sasl.kerberos.service.name='kafka', properties.sasl.kerberos.keytab='/etc/krb5kdc/kafka.client.keytab', properties.sasl.kerberos.principal='kafkaclient4@AP-SOUTHEAST-1.COMPUTE.INTERNAL', properties.sasl.kerberos.kinit.cmd='sudo kinit -R -kt "%{sasl.kerberos.keytab}" %{sasl.kerberos.principal} || sudo kinit -kt "%{sasl.kerberos.keytab}" %{sasl.kerberos.principal}', properties.sasl.kerberos.min.time.before.relogin='10000' ) FORMAT PLAIN ENCODE JSON; ``` The implementation of SASL/OAUTHBEARER in RisingWave validates only [unsecured client side tokens](https://docs.confluent.io/platform/current/kafka/authentication%5Fsasl/authentication%5Fsasl%5Foauth.html#unsecured-client-side-token-creation-options-for-sasl-oauthbearer), and does not support OpenID Connect (OIDC) authentication. Therefore, it should not be used in production environments. | Parameter | Notes | | :--------------------------------- | :--------------------------------------------------------------------------------------------------------- | | properties.security.protocol | For SASL/OAUTHBEARER without SSL, set to SASL\_PLAINTEXT. For SASL/OAUTHBEARER with SSL, set to SASL\_SSL. | | properties.sasl.mechanism | Set to OAUTHBEARER. | | properties.sasl.oauthbearer.config | | For the definitions of the parameters, see the [librdkafka properties list](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). Note that the parameters in the list assumes all parameters start with `properties.` and therefore do not include this prefix. Also, due to the limitation of the SASL/OAUTHBEARER implementation, you only need to specify one OAUTHBEARER parameter: `properties.sasl.oauthbearer.config`. Other OAUTHBEARER parameters are not applicable. For SASL/OAUTHBEARER with SSL, you also need to include these SSL parameters: * `properties.ssl.ca.location` * `properties.ssl.certificate.location` * `properties.ssl.key.location` * `properties.ssl.key.password` This is an example of creating a sink authenticated with SASL/OAUTHBEARER without SSL encryption. ```sql theme={null} CREATE SINK sink1 FROM mv1 WITH ( connector='kafka', type = 'append-only', topic='quickstart-events', properties.bootstrap.server='localhost:9093', properties.sasl.mechanism='OAUTHBEARER', properties.security.protocol='SASL_PLAINTEXT', properties.sasl.oauthbearer.config='principal=bob' ) FORMAT PLAIN ENCODE JSON; ``` ## Data type mapping - RisingWave and Debezium JSON | RisingWave Data Type | Schema Type in JSON | Schema Name in JSON | | :--------------------- | :------------------ | :-------------------------------------- | | boolean | boolean | n/a | | smallint | int16 | n/a | | integer | int32 | n/a | | bigint | int64 | n/a | | real | float | n/a | | double precision | double | n/a | | character varying | string | n/a | | bytea | bytes | n/a | | numeric | string | n/a | | date | int32 | org.apache.kafka.connect.data.Date | | time without time zone | int64 | org.apache.kafka.connect.data.Time | | timestamp | int64 | org.apache.kafka.connect.data.Timestamp | | timestamptz | string | io.debezium.time.ZonedTimestamp | | interval | string | io.debezium.time.Interval | | JSONB | string | io.debezium.data.Json | | struct | string | n/a | | array | string | n/a |