> ## Documentation Index
> Fetch the complete documentation index at: https://risingwavelabs-wyx-add-timestamp-dedicated-page.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Sink data from RisingWave to Apache Pulsar

> This guide describes how to sink data from RisingWave to Apache Pulsar.

[Apache Pulsar](https://pulsar.apache.org) is an open-source distributed pub-sub messaging system and event streaming platform that is scalable and designed to support geo-replication.

## Prerequisites

Before sinking data from RisingWave to Pulsar, please ensure the following:

* A Pulsar cluster is running and accessible from RisingWave.
* You have the permission to access the Pulsar topics you want to sink data to.

## Syntax[](#syntax "Direct link to Syntax")

```sql theme={null}
CREATE SINK [ IF NOT EXISTS ] sink_name
[FROM sink_from | AS select_query]
WITH (
   connector='pulsar',
   connector_parameter = 'value', ...
)
FORMAT data_format ENCODE data_encode [ (
    key = 'value' ) ]
[KEY ENCODE key_encode [(...)]]
;
```

## Parameters

| Parameter names                     | Description                                                                                                                                                                          |
| :---------------------------------- | :----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| topic                               | Required. The address of the Pulsar topic. One source can only correspond to one topic.                                                                                              |
| service.url                         | Required. The address of the Pulsar service.                                                                                                                                         |
| auth.token                          | Optional. A token for auth. If both auth.token and oauth are set, only oauth authorization is considered.                                                                            |
| oauth.issuer.url                    | Optional. The issuer URL for OAuth2. This field must be filled if other oauth fields are specified.                                                                                  |
| oauth.credentials.url               | Optional. The path for credential files, which starts with `file://`. This field must be filled if other oauth fields are specified.                                                 |
| oauth.audience                      | Optional. The audience for OAuth2. This field must be filled if other oauth fields are specified.                                                                                    |
| oauth.scope                         | Optional. The scope for OAuth2.                                                                                                                                                      |
| aws.credentials.access\_key\_id     | Optional. The AWS access key for loading from S3. This field does not need to be filled if oauth.credentials.url is specified to a local path.                                       |
| aws.credentials.secret\_access\_key | Optional. The AWS secret access key for loading from S3. This field does not need to be filled if oauth.credentials.url is specified to a local path.                                |
| max\_retry\_num                     | Optional. The maximum number of times to retry sending a batch to Pulsar. This allows retrying in case of transient errors. The default value is 3.                                  |
| retry\_interval                     | Optional. The time in milliseconds to wait after a failure before retrying to send a batch. The default value is 100ms.                                                              |
| primary\_key                        | Optional. The primary keys of the sink. Use `,` to delimit the primary key columns. Primary keys are optional when creating a PLAIN sink but required for UPSERT and DEBEZIUM sinks. |

## FORMAT and ENCODE options

<Note>
  These options should be set in `FORMAT data_format ENCODE data_encode (key = 'value')`, instead of the `WITH` clause
</Note>

| Field                     | Notes                                                                                                                                                                                                                                                                                                                                                                                            |
| :------------------------ | :----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| data\_format              | Data format. Allowed formats:<ul><li> `PLAIN`: Output data with insert operations.</li><li> `DEBEZIUM`: Output change data capture (CDC) log in Debezium format.</li><li> `UPSERT`: Output data as a changelog stream. `primary_key` must be specified in this case. </li></ul>To learn about when to define the primary key if creating an UPSERT sink, see the [Overview](/delivery/overview). |
| data\_encode              | Data encode. Supported encode: JSON.                                                                                                                                                                                                                                                                                                                                                             |
| 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. <ul><li>If omitted, the output format of timestamptz is `2023-11-11T18:30:09.453000Z` which includes the UTC suffix `Z`.</li><li>When `utc_without_suffix` is specified, the format is changed to `2023-11-11 18:30:09.453000`.</li></ul>                        |
| key\_encode               | Optional. When specified, the key encode can only be TEXT, and the primary key should be one and only one of the following types: `varchar`, `bool`, `smallint`, `int`, and `bigint`; When absent, both key and value will use the same setting of `ENCODE data_encode ( ... )`.                                                                                                                 |

## Example

The following SQL query in RisingWave creates a Pulsar sink.

```sql theme={null}
CREATE SINK IF NOT EXISTS pulsar_sink
FROM mv_name
WITH (
  connector = 'pulsar',
  topic = 'test-topic',
  service.url = 'pulsar://broker:6650',

  -- OAuth
  oauth.issuer.url = 'https://issuer.com',
  oauth.credentials.url = 'https://provider.com',
  oauth.audience = 'test-aud',
  oauth.scope = 'consume',

  -- S3 credential for oauth file
  aws.credentials.access_key_id = 'xxx',
  aws.credentials.secret_access_key = 'xxx'
)
FORMAT DEBEZIUM ENCODE JSON;
```