title: Pipeline configuration file

title: Overview

text:
The main configuration details for an RDI pipeline are in the `config.yaml` file.
This file specifies the connection details for the source and target databases,
and also the set of tables you want to capture. You can also add one or more
[job files]()
if you want to apply custom transformations to the captured data.

title: Example

text:
Below is an example of a `config.yaml` file. Note that the values of the
form "`${name}`" refer to secrets that you should set as described in
[Set secrets]().
In particular, you should normally use secrets as shown to set the source
and target username and password rather than storing them in plain text in this file.

[code example]

title: Sections

text:
The main sections of the file configure [`sources`](#sources), [`targets`](#targets),
and [`processors`](#processors).

title: Sources

text:
The `sources` section has a subsection for the source that
you need to configure. The source section starts with a unique name
to identify the source (in the example, there is a source
called `mysql` but you can choose any name you like). The example
configuration contains the following data:

- `type`: The type of collector to use for the pipeline.
  Currently, the only types RDI supports are `cdc` and `external`.
  If the source type is set to `external`, no collector resources will be created by the operator,
  and all other source sections should be empty or not specified at all.
- `connection`: The connection details for the source database: `type`, `host`, `port`,
  and credentials (`username` and `password`).
  - `type` is the source database type, one of `mariadb`, `mysql`, `oracle`, `postgresql`, or `sqlserver`.
  - If you use [TLS](https://en.wikipedia.org/wiki/Transport_Layer_Security)/
    or [mTLS](https://en.wikipedia.org/wiki/Mutual_authentication#mTLS) to connect
    to the source database, you may need to specify additional properties in the
    `advanced` section with references to the corresponding certificates depending
    on the source database type. Note that these properties **must** be references to
    secrets that you should set as described in [Set secrets]().
- `databases`: List of all databases to collect data from for source database types
  that support multiple databases, such as `mysql` and `mariadb`.
- `schemas`: List of all schemas to collect data from for source database types
  that support multiple schemas, such as `oracle`, `postgresql`, and `sqlserver`.
- `tables`: List of all tables to collect data from. Each table is identified by its
  full name, including a database or schema prefix. If there is a single
  database or schema, this prefix can be omitted.
  For each table, you can specify:
  - `columns`: A list of the columns you are interested in (the default is to
    include all columns)
  - `keys`: A list of columns to create a composite key if your table
    doesn't already have a [`PRIMARY KEY`](https://www.w3schools.com/sql/sql_primarykey.asp) or
    [`UNIQUE`](https://www.w3schools.com/sql/sql_unique.asp) constraint.
  - `snapshot_sql`: A query to be used when performing the initial snapshot.
    By default, a query that contains all listed columns of all listed tables will be used.
- `advanced`: These optional properties configure other Debezium-specific features.
  The available sub-sections are:
  - `source`: Properties for reading from the source database.
    See the Debezium [Source connectors](https://debezium.io/documentation/reference/stable/connectors/)
    pages for more information about the properties available for each database type.
  - `sink`: Properties for writing to Redis streams in the RDI database.
    See the Debezium [Redis stream properties](https://debezium.io/documentation/reference/stable/operations/debezium-server.html#_redis_stream)
    page for the full set of available properties.
  - `quarkus`: Properties for the Debezium server, such as the log level. See the
    Quarkus [Configuration options](https://quarkus.io/guides/all-config)
    docs for the full set of available properties.
  - `java_options`: controls the JAVA_OPTS environment variable (for RDI 1.15.1 and above). Use it to modify the default values for Java heap size and other Java options for the Debezium server.
    For example, set it to `"-Xmx2g -Xms512m"` to set the maximum heap size to 2 GB and the initial heap size to 512 MB.

title: Targets

text:
Use this section to provide the connection details for the target Redis
database(s). As with the sources, you should start each target section
with a unique name that you are free to choose (here, the example uses the name
`target`). In the `connection` section, you can specify the
`type` of the target database, which must be `redis`, along with
connection details such as `host`, `port`, and credentials (`username` and `password`).
If you use [TLS](https://en.wikipedia.org/wiki/Transport_Layer_Security)/
or [mTLS](https://en.wikipedia.org/wiki/Mutual_authentication#mTLS) to connect
to the target database, you must specify the CA certificate (for TLS),
and the client certificate and private key (for mTLS) in `cacert`, `cert`, and `key`.
Note that these certificates **must** be references to secrets
that you should set as described in [Set secrets]()
(it is not possible to include these certificates as plain text in the file).

If you specify `localhost` as the address of either the source or target server during
installation then the connection will fail if the actual IP address changes for the local
VM. For this reason, it is recommended that you don't use `localhost` for the address. However,
if you do encounter this problem, you can fix it using the following commands on the VM
that is running RDI itself:

[code example]

title: Processors

text:
The `processors` section configures the behavior of the pipeline. The [example](#example)
configuration above contains the following properties:

- `on_failed_retry_interval`: Number of seconds to wait before retrying a failed operation.
  The default is 5 seconds.
- `read_batch_size`: Maximum number of records to read from the source database. RDI will
  wait for the batch to fill up to `read_batch_size` or for `duration` to elapse,
  whichever happens first. The default is 2000.
- `target_data_type`: Data type to use in the target Redis database. The options are `hash`
  for Redis Hash (the default), or `json` for RedisJSON, which is available only if you have added the
  RedisJSON module to the target database. Note that this setting is mainly useful when you
  don't provide any custom jobs. When you do provide jobs, you can specify the
  target data type in each job individually and choose from a wider range of data types.
  See [Job files]()
  (which requires the RedisJSON module) for more information.
- `duration`: Time (in ms) after which data will be read from the stream even if
  `read_batch_size` was not reached. The default is 100 ms.
- `write_batch_size`: The batch size for writing data to the target Redis database. This should be
  less than or equal to the `read_batch_size`. The default is 200.
- `dedup`: Boolean value to enable the deduplication mechanism. The default is `false`.
- `dedup_max_size`: Maximum size of the deduplication set. The default is 1024.
- `error_handling`: The strategy to use when an invalid record is encountered. The available
  strategies are `ignore` and  `dlq` (store rejected messages in a dead letter queue).
  The default is `dlq`. See
  [What does RDI do if the data is corrupted or invalid?]()
  for more information about the dead letter queue.

See also the
[RDI configuration file reference]()
for full details of the other available properties.

code:
sources:
  mysql:
    type: cdc
    logging:
      level: info
    connection:
      type: mysql
      host: <DB_HOST> # e.g. localhost
      port: 3306
      # User and password are injected from the secrets.
      user: ${SOURCE_DB_USERNAME}
      password: ${SOURCE_DB_PASSWORD}
    # Additional properties for the source collector:
    # List of databases to include (optional).
    # databases:
    #   - database1
    #   - database2

    # List of tables to be synced (optional).
    # tables:
    #   If only one database is specified in the databases property above,
    #   then tables can be defined without the database prefix.
    #   <DATABASE_NAME>.<TABLE_NAME>:
    #     List of columns to be synced (optional).
    #     columns:
    #       - <COLUMN_NAME>
    #       - <COLUMN_NAME>
    #     List of columns to be used as keys (optional).
    #     keys:
    #       - <COLUMN_NAME>

    # Example: Sync specific tables.
    # tables:
    #   Sync a specific table with all its columns:
    #   redislabscdc.account: {}
    #   Sync a specific table with selected columns:
    #   redislabscdc.emp:
    #     columns:
    #       - empno
    #       - fname
    #       - lname

    # Advanced collector properties (optional):
    # advanced:
    #   Sink collector properties - see the full list at
    #     https://debezium.io/documentation/reference/stable/operations/debezium-server.html#_redis_stream
    #   sink:
    #     Optional hard limits on memory usage of RDI streams.
    #     redis.memory.limit.mb: 300
    #     redis.memory.threshold.percentage: 85

    #     Uncomment for production so RDI Collector will wait on replica
    #     when writing entries.
    #     redis.wait.enabled: true
    #     redis.wait.timeout.ms: 1000
    #     redis.wait.retry.enabled: true
    #     redis.wait.retry.delay.ms: 1000

    #   Source specific properties - see the full list at
    #     https://debezium.io/documentation/reference/stable/connectors/
    #   source:
    #     snapshot.mode: initial
    #     Uncomment if you want a snapshot to include only a subset of the rows
    #     in a table. This property affects snapshots only.
    #     snapshot.select.statement.overrides: <DATABASE_NAME>.<TABLE_NAME>
    #     The specified SELECT statement determines the subset of table rows to
    #     include in the snapshot.
    #     snapshot.select.statement.overrides.<DATABASE_NAME>.<TABLE_NAME>: <SELECT_STATEMENT>

    #     Example: Snapshot filtering by order status.
    #     To include only orders with non-pending status from customers.orders
    #     table:
    #     snapshot.select.statement.overrides: customer.orders
    #     snapshot.select.statement.overrides.customer.orders: SELECT * FROM customers.orders WHERE status != 'pending' ORDER BY order_id DESC

    #   Quarkus framework properties - see the full list at
    #     https://quarkus.io/guides/all-config
    #   quarkus:
    #     banner.enabled: "false"

    #   `java_options` (for RDI 1.15.1 and above) controls the JAVA_OPTS environment variable. Use it to modify the default values for
    #       Java heap size and other Java options for the Debezium server.
    #   java_options: "-Xmx2g -Xms512m"

targets:
  # Redis target database connections.
  # The default connection must be named 'target' and is used when no
  # connection is specified in jobs or no jobs
  # are deployed. However multiple connections can be defined here and used
  # in the job definition output blocks:
  # (e.g. target1, my-cloud-redis-db2, etc.)
  target:
    connection:
      type: redis
      # Host of the Redis database to which RDI will
      # write the processed data.
      host: <REDIS_TARGET_DB_HOST> # e.g. localhost
      # Port for the Redis database to which RDI will
      # write the processed data.
      port: <REDIS_TARGET_DB_PORT> # e.g. 12000
      # User of the Redis database to which RDI will write the processed data.
      # Uncomment if you are not using the default user.
      # user: ${TARGET_DB_USERNAME}
      # Password for Redis target database.
      password: ${TARGET_DB_PASSWORD}
      # SSL/TLS configuration: Uncomment to enable secure connections.
      # key: ${TARGET_DB_KEY}
      # key_password: ${TARGET_DB_KEY_PASSWORD}
      # cert: ${TARGET_DB_CERT}
      # cacert: ${TARGET_DB_CACERT}
processors:
  # Interval (in seconds) on which to perform retry on failure.
  # on_failed_retry_interval: 5
  # The batch size for reading data from the source database.
  # read_batch_size: 2000
  # Time (in ms) after which data will be read from stream even if
  # read_batch_size was not reached.
  # duration: 100
  # The batch size for writing data to the target Redis database. Should be
  # less than or equal to the read_batch_size.
  # write_batch_size: 200
  # Enable deduplication mechanism (default: false).
  # dedup: <DEDUP_ENABLED>
  # Max size of the deduplication set (default: 1024).
  # dedup_max_size: <DEDUP_MAX_SIZE>
  # Error handling strategy: ignore - skip, dlq - store rejected messages
  # in a dead letter queue.
  # error_handling: dlq
  # Dead letter queue max messages per stream.
  # dlq_max_messages: 1000
  # Data type to use in Redis target database: `hash` for Redis Hash,
  # `json` for JSON (which requires the RedisJSON module).
  # target_data_type: hash
  # Number of processes to use when syncing initial data.
  # initial_sync_processes: 4
  # Checks if the batch has been written to the replica shard.
  # wait_enabled: false
  # Timeout in milliseconds when checking write to the replica shard.
  # wait_timeout: 1000
  # Ensures that a batch has been written to the replica shard and keeps
  # retrying if not.
  # retry_on_replica_failure: true
  # Enable merge as the default strategy to writing JSON documents.
  # json_update_strategy: merge
  # Use native JSON merge if the target RedisJSON module supports it.
  # use_native_json_merge: true

section_id: example

code:
sudo k3s kubectl delete nodes --all
sudo service k3s restart

section_id: targets


title: Prepare AWS Aurora MySQL/AWS RDS MySQL for RDI

summary: Enable CDC features in your source databases

page_type: content

title: Overview

role: overview

text:
Follow the steps in the sections below to prepare an [AWS Aurora MySQL](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/CHAP_GettingStartedAurora.CreatingConnecting.Aurora.html) or [AWS RDS MySQL](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/CHAP_GettingStarted.CreatingConnecting.MySQL.html) database.
database to work with RDI.

Select the steps for your database type.

**AWS Aurora MySQL:**

[code example]

title: Add an Aurora reader node

role: content

text:
RDI requires that your Aurora MySQL database has at least one replica or reader node.

To add a reader node to an existing database, select **Add reader** from the **Actions** menu of the database and [add a reader node](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/aurora-replicas-adding.html).

You can also create one during database creation by selecting **Create an Aurora Replica or Reader node in a different AZ (recommended for scaled availability)** under **Availability & durability > Multi-AZ deployment**.

title: Create and apply parameter group

text:
RDI requires some changes to database parameters. On AWS Aurora, you change these parameters via a parameter group.

[code example]

1.
    In the [Relational Database Service (RDS) console](https://console.aws.amazon.com/rds/), navigate to **Parameter groups**.

    If you have no existing parameter group,
    [create a new parameter group](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_WorkingWithParamGroups.CreatingCluster.html)
    with the following settings:

    | Name | Value |
    | :-- | :-- |
    | **Parameter group name**  | Enter a suitable parameter group name, like `rdi-mysql` |
    | **Description**  | (Optional) Enter a description for the parameter group |
    | **Engine Type**  | Choose **Aurora MySQL**.  |
    | **Parameter group family**  | Choose **aurora-mysql8.0**. |
    | **Type**  | Select **DB Parameter Group**. |

    Select **Create** to create the parameter group.

    If you *do* have an existing parameter group, select it and then either:

    -   Select **Edit** from **Parameter group actions** to
        [modify the parameter group](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_WorkingWithParamGroups.ModifyingCluster.html)
        with the settings shown in the table above.
    -   Select **Copy** from **Parameter group actions** to
        [copy the existing parameter group](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_WorkingWithParamGroups.CopyingCluster.html)
        and then modify the copy with the settings shown in the table above.

1.
    Ensure that the parameter group you have just created or modified is selected
    and then select **Edit**. Change the following parameters:

    | Name | Value |
    | :-- | :-- |
    | `binlog_format`  | `ROW` |
    | `binlog_row_image`  | `FULL` |
    | `gtid_mode`  | `ON` |
    | `enforce_gtid_consistency`  | `ON` |

    Select **Save Changes** to apply the changes to the parameter group.

1.
    Go back to your target database on the RDS console, select **Modify** and then scroll down to **Additional Configuration**. Set the **DB Cluster Parameter Group** to the group you just created.

    Select **Save changes** to apply the parameter group to the new database.

1.
    Reboot your database instance. See [Rebooting a DB instance within an Aurora cluster](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/aurora-reboot-db-instance.html) for more information.

**AWS RDS MySQL:**

[code example]

title: Create and apply parameter group

text:
RDI requires some changes to database parameters. On AWS RDS, you change these parameters via a parameter group.

[code example]

1.
    In the [Relational Database Service (RDS) console](https://console.aws.amazon.com/rds/), navigate to **Parameter groups**.

    If you have no existing parameter group,
    [create a new parameter group](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_WorkingWithParamGroups.CreatingCluster.html)
    with the following settings:

    | Name | Value |
    | :-- | :-- |
    | **Parameter group name**  | Enter a suitable parameter group name, like `rdi-mysql` |
    | **Description**  | (Optional) Enter a description for the parameter group |
    | **Engine Type**  | Choose **MySQL Community**.  |
    | **Parameter group family**  | Choose **mysql8.0**. |

    Select **Create** to create the parameter group.

    If you *do* have an existing parameter group, select it and then either:

    -   Select **Edit** from **Parameter group actions** to
        [modify the parameter group](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_WorkingWithParamGroups.ModifyingCluster.html)
        with the settings shown in the table above.
    -   Select **Copy** from **Parameter group actions** to
        [copy the existing parameter group](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_WorkingWithParamGroups.CopyingCluster.html)
        and then modify the copy with the settings shown in the table above.

1.
    Ensure that the parameter group you have just created or modified is selected
    and then select **Edit**. Change the following parameters:

    | Name | Value |
    | :-- | :-- |
    | `binlog_format`  | `ROW` |
    | `binlog_row_image`  | `FULL` |

    Select **Save Changes** to apply the changes to the parameter group.

1.
    Go back to your target database on the RDS console, select **Modify** and then scroll down to **Additional Configuration**. Set the **DB Cluster Parameter Group** to the group you just created.

    Select **Save changes** to apply the parameter group to the new database.

1.
    Reboot your database instance. See [Rebooting a DB instance](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_RebootInstance.html) for more information.

language: checklist {id="auroramysql" nointeractive="true" }

code:
- [ ] [Add an Aurora reader node](#add-an-aurora-reader-node)
- [ ] [Create and apply parameter group](#aurora-create-and-apply-parameter-group)
- [ ] [Create Debezium user](#aurora-create-debezium-user)

section_id: overview

language: checklist {id="auroramysql-param-group" nointeractive="true" }

code:
- [ ] [Create/modify a parameter group](#aurora-create-a-parameter-group)
- [ ] [Apply the parameter group](#aurora-apply-the-parameter-group)
- [ ] [Apply the parameter group to the database](#aurora-apply-the-parameter-group-to-the-database)
- [ ] [Reboot the database instance](#aurora-reboot-the-database-instance)

language: checklist {id="rds-mysql-list" nointeractive="true" }

code:
- [ ] [Create and apply parameter group](#rds-create-and-apply-parameter-group)
- [ ] [Create Debezium user](#rds-create-debezium-user)

language: checklist {id="rds-mysql-param-group" nointeractive="true" }

code:
- [ ] [Create/modify a parameter group](#rds-create-a-parameter-group)
- [ ] [Apply the parameter group](#rds-apply-the-parameter-group)
- [ ] [Apply the parameter group to the database](#rds-apply-the-parameter-group-to-the-database)
- [ ] [Reboot the database instance](#rds-reboot-the-database-instance)


title: Prepare AWS Aurora PostgreSQL/AWS RDS PostgreSQL for RDI

summary: Prepare AWS Aurora PostgreSQL databases to work with RDI

title: Overview

text:
Follow the steps in the sections below to prepare an
[AWS Aurora PostgreSQL](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/CHAP_GettingStartedAurora.CreatingConnecting.AuroraPostgreSQL.html) or [AWS RDS PostgreSQL](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/CHAP_GettingStarted.CreatingConnecting.PostgreSQL.html)
database to work with RDI.

[code example]

title: Create and apply parameter group

text:
RDI requires some changes to database parameters. On AWS RDS and AWS Aurora, you change these parameters via a parameter group.

[code example]

1.
    In the [Relational Database Service (RDS) console](https://console.aws.amazon.com/rds/), navigate to **Parameter groups**.

    If you have no existing parameter group,
    [create a new parameter group](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_WorkingWithParamGroups.CreatingCluster.html)
    with the following settings:

    | Name | Value |
    | :-- | :-- |
    | **Parameter group name**  | Enter a suitable parameter group name, like `rdi-aurora-pg` or `rdi-rds-pg` |
    | **Description**  | (Optional) Enter a description for the parameter group |
    | **Engine Type**  | Choose **Aurora PostgreSQL** for Aurora PostgreSQL or **PostgreSQL** for AWS RDS PostgreSQL. |
    | **Parameter group family**  | Choose **aurora-postgresql15** for Aurora PostgreSQL or **postgresql13** for AWS RDS PostgreSQL. |

    Select **Create** to create the parameter group.

    If you *do* have an existing parameter group, select it and then either:

    -   Select **Edit** from **Parameter group actions** to
        [modify the parameter group](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_WorkingWithParamGroups.ModifyingCluster.html)
        with the settings shown in the table above.
    -   Select **Copy** from **Parameter group actions** to
        [copy the existing parameter group](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_WorkingWithParamGroups.CopyingCluster.html)
        and then modify the copy with the settings shown in the table above.

1.
    Ensure that the parameter group you have just created or modified is selected
    and then select **Edit**. Change the following parameter:

    | Name | Value |
    | :-- | :-- |
    | `rds.logical_replication`  | `1` |

    Select **Save Changes** to apply the changes to the parameter group.

1.
    Go back to your database on the RDS console, select **Modify** and then scroll down to **Additional Configuration**. Set the **DB Cluster Parameter Group** to the group you just created.

    Select **Save changes** to apply the parameter group to your database.

1.
    Reboot your database instance. See [Rebooting a DB instance within an Aurora cluster](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/aurora-reboot-db-instance.html) or [Rebooting a DB instance (RDS)](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_RebootInstance.html) for more information.

title: Create Debezium user

text:
The Debezium connector needs a user account to connect to PostgreSQL. This
user must have appropriate permissions on all databases where you want Debezium
to capture changes.

[code example]

1.
    Connect to PostgreSQL as the `postgres` user and create a new user for the connector:

    [code example]

    Replace `<username>` and `<password>` with a username and password for the new user.

1.
    Grant the user the necessary replication permissions:

    [code example]

    Replace `<username>` with the username of the Debezium user.

1.
    Connect to your database as the `postgres` user and grant the new user access to one or more schemas in the database:

    [code example]

    Replace `<username>` with the username of the Debezium user and `<schema>` with the schema name.

1.
    Connect to your database as the `postgres` user and allow the Debezium user to connect to the database:

    [code example]

    Replace `<database>` with the name of the database and `<username>` with the username of the Debezium user.

1.
    Connect to your database as the `postgres` user and grant the new user usage on the schema:

    [code example]

    Replace `<schema>` with the schema name and `<username>` with the username of the Debezium user.">

1.
    Connect to your database as the `postgres` user and grant the new user the necessary privileges for the future:

    [code example]

    Replace `<schema>` with the schema name and `<username>` with the username of the Debezium user.

1.
    Connect to your database as the `postgres` user and create a publication for the database:

    [code example]

language: checklist {id="aurorapostgresql" nointeractive="true" }

code:
- [ ] [Create and apply parameter group](#create-and-apply-parameter-group)
- [ ] [Create Debezium user](#create-debezium-user)

language: checklist {id="aurorapostgresql-param-group" nointeractive="true" }

code:
- [ ] [Create/modify a parameter group](#create-a-parameter-group)
- [ ] [Apply the parameter group](#apply-the-parameter-group)
- [ ] [Apply the parameter group to the database](#apply-the-parameter-group-to-the-database)
- [ ] [Reboot the database instance](#reboot-the-database-instance)

section_id: create-and-apply-parameter-group

language: checklist {id="aurorapostgresql-create-debezium-user" nointeractive="true" }

code:
- [ ] [Connect to PostgreSQL as the `postgres` user](#connect-to-postgresql-as-the-postgres-user)
- [ ] [Grant the user the necessary replication permissions](#grant-the-user-the-necessary-replication-permissions)
- [ ] [Grant the user access to the database](#grant-the-user-access-to-the-database)

section_id: create-debezium-user

code: CREATE ROLE <username> WITH LOGIN PASSWORD '<password>' VALID UNTIL 'infinity';

code: GRANT rds_replication TO <username>;

code: GRANT SELECT ON ALL TABLES IN SCHEMA <schema> TO <username>;

code: GRANT CONNECT ON DATABASE <database> TO <username>;

code: GRANT USAGE ON SCHEMA <schema> TO <username>;

code:
ALTER DEFAULT PRIVILEGES IN SCHEMA <schema>
        GRANT SELECT ON TABLES TO <username>;

code: CREATE PUBLICATION dbz_publication FOR ALL TABLES;


title: Prepare Microsoft SQL Server on AWS RDS for RDI

title: Overview

text:
Follow the steps in the sections below to prepare a [Microsoft SQL Server on AWS RDS](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/CHAP_GettingStarted.CreatingConnecting.SQLServer.html) database to work with RDI.

Change Data Capture (CDC) is not supported on SQL Server Express Edition. Only the Standard, Enterprise, and Developer editions support CDC and are supported by RDI.

[code example]

title: Create the Debezium user

text:
The Debezium connector needs a user account to connect to SQL Server. This
user must have appropriate permissions on all databases where you want Debezium
to capture changes.

[code example]

1.
    Connect to your database as an admin user and create a new user for the connector:

    [code example]

    Replace `<username>` and `<password>` with a username and password for the new user and replace `<database>` with the name of your database.

1.
    Grant the user the necessary permissions:

    [code example]

    Replace `<username>` with the username of the Debezium user and replace `<database>` with the name of your database.

title: Enable CDC on the database

text:
Change Data Capture (CDC) must be enabled for the database and for each table you want to capture.

[code example]

1.
    Enable CDC for the database by running the following command:

    [code example]

    Replace `<database>` with the name of your database.

1.
    Enable CDC for each table you want to capture by running the following commands:

    [code example]

    Replace `<database>` with the name of your database, `<schema>` with the name of the schema containing the table, `<table>` with the name of the table, and `<role>` with the name of a new role that will be created to manage access to the CDC data.

The value for `@role_name` can’t be a fixed database role, such as `db_datareader`.
Specifying a new name will create a corresponding database role that has full access to the
captured change data.

1.
    Add the Debezium user to the CDC role:

    [code example]

    Replace `<role>` with the name of the role you created in the previous step and replace `<username>` with the username of the Debezium user.

language: checklist {id="rds-sqlserver-list" nointeractive="true" }

code:
- [ ] [Create the Debezium user](#create-the-debezium-user)
- [ ] [Enable CDC on the database](#enable-cdc-on-the-database)

language: checklist {id="rds-sqlserver-create-debezium-user" nointeractive="true" }

code:
- [ ] [Connect to SQL Server as an admin user](#connect-to-sql-server-as-an-admin-user)
- [ ] [Grant the user the necessary permissions](#grant-the-user-the-necessary-permissions)

section_id: create-the-debezium-user

code:
USE master
    GO
    CREATE LOGIN <username> WITH PASSWORD = '<password>'
    GO
    USE <database>
    GO
    CREATE USER <username> FOR LOGIN <username>
    GO

code:
USE master
    GO
    GRANT VIEW SERVER STATE TO <username>
    GO
    USE <database>
    GO
    EXEC sp_addrolemember N'db_datareader', N'<username>'
    GO

language: checklist {id="rds-sqlserver-enable-cdc" nointeractive="true" }

code:
- [ ] [Enable CDC for the database](#enable-cdc-for-the-database)
- [ ] [Enable CDC for each table you want to capture](#enable-cdc-for-each-table-you-want-to-capture)
- [ ] [Add the Debezium user to the CDC role](#add-the-debezium-user-to-the-cdc-role)

section_id: enable-cdc-on-the-database

code:
EXEC msdb.dbo.rds_cdc_enable_db '<database>'
    GO

code:
USE <database>
    GO
    EXEC sys.sp_cdc_enable_table
    @source_schema = N'<schema>',
    @source_name   = N'<table>',
    @role_name     = N'<role>',
    @supports_net_changes = 0
    GO

code:
USE <database>
    GO
    EXEC sp_addrolemember N'<role>', N'<username>'
    GO


title: Prepare AWS RDS and Aurora databases for RDI

summary: Learn how to prepare AWS RDS and Aurora databases for RDI.

title: Content

text:
You can use RDI with databases on [AWS Relational Database Service (RDS)](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Welcome.html) and [AWS Aurora](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/CHAP_AuroraOverview.html).

The pages in this section give detailed instructions to get your source
database ready for Debezium to use:


title: Prepare source databases

summary: Prepare Oracle and Oracle RAC databases to work with RDI

title: Prepare Oracle and Oracle RAC for RDI

summary: Prepare MongoDB databases to work with RDI

title: Prepare MongoDB for RDI

summary: Prepare MySQL and MariaDB databases to work with RDI

title: Prepare MySQL/MariaDB for RDI

summary: Prepare PostgreSQL databases (including Supabase) to work with RDI

title: Prepare PostgreSQL/Supabase for RDI

summary: Prepare Google Cloud Spanner databases to work with RDI

title: Prepare Spanner for RDI

summary: Prepare SQL Server databases to work with RDI

title: Prepare SQL Server for RDI

summary: Prepare Neon databases to work with RDI

title: Prepare Neon for RDI

title: Content

text:
Each database uses a different mechanism to track changes to its data and
generally, these mechanisms are not switched on by default.
RDI's Debezium collector uses these mechanisms for change data capture (CDC),
so you must prepare your source database before you can use it with RDI.

RDI supports the following source databases:

The pages in this section give detailed instructions to get your source
database ready for Debezium to use:


title: Prepare MongoDB for RDI

title: Overview

text: This guide describes the steps required to prepare a MongoDB database as a source for Redis Data Integration (RDI) pipelines.

title: Prerequisites

text:
- **MongoDB version:** 6.0 or later (replica set, sharded cluster, or MongoDB Atlas).
- **User privileges:** You must have a MongoDB user with sufficient privileges to read the oplog and collections, and to use change streams.
- **Network access:** The RDI Collector must be able to connect to all MongoDB nodes in your deployment.

The MongoDB connector is not capable of monitoring the changes of a standalone MongoDB server, since standalone servers do not have an oplog. The connector will work if the standalone server is converted to a replica set with one member.

title: Summary

text:
The following table summarizes the considerations to prepare a MongoDB database for RDI.

| Requirement         | Description                                                                 |
|---------------------|-----------------------------------------------------------------------------|
| MongoDB Topology    | Replica Set, Sharded Cluster, or MongoDB Atlas                              |
| User Roles          | readAnyDatabase, clusterMonitor                                             |
| Oplog               | Sufficient size for snapshot and streaming                                  |
| Pre/Post Images     | Enable on collections **only if using a custom key**                        |
| Connection String   | Must include all hosts, replicaSet (if applicable), authSource, credentials |
| MongoDB Atlas       | **[SSL required](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#mongodb-property-mongodb-ssl-enabled)**, provide root CA as `SOURCE_DB_CACERT` secret in RDI       |
| Network             | RDI Collector must reach all MongoDB nodes on required ports                |

The following checklist shows the steps to prepare a MongoDB database for RDI,
with links to the sections that explain the steps in full detail.
You may find it helpful to track your progress with the checklist as you
complete each step.

[code example]

title: 1. Configure oplog size

text:
The Debezium MongoDB connector relies on the [oplog](https://www.mongodb.com/docs/manual/core/replica-set-oplog/) to capture changes from a replica set. The oplog is a fixed-size, capped collection. When it reaches its maximum size, it overwrites the oldest entries. If the connector is stopped and restarted, it attempts to resume from its last recorded position in the oplog. If that position has been overwritten, the connector may fail to start and report an invalid resume token error.

To prevent this, ensure the oplog retains enough history for Debezium to resume streaming after interruptions. You can do this by:

- **Increasing the oplog size:** Set the oplog size based on your workload, ensuring it can store more than the peak number of oplog entries generated per hour.
- **Setting a minimum oplog retention period (MongoDB 4.4+):** Configure MongoDB to retain oplog entries for a minimum number of hours, guaranteeing availability even if the oplog reaches its maximum size. This is generally preferred, but for high-throughput clusters nearing capacity, you may need to increase the oplog size instead.

For detailed guidance, see the Debezium [oplog configuration documentation](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#mongodb-optimal-oplog-config).

title: 2. Create a MongoDB user for RDI

text:
Create a user with the following roles on the source database:
- `readAnyDatabase` (optional) OR grant `read` for the specific database(s) you will use with RDI
- `clusterMonitor`

Example:
[code example]

title: 3. Connection string format

text:
The RDI Collector requires a MongoDB connection string that includes all relevant hosts and authentication details.

Example (Replica Set):
[code example]
Example (Sharded Cluster):
[code example]
- For Atlas, adjust the connection string accordingly (see example below).
- Set `replicaSet` and `authSource` as appropriate for your deployment.

title: 4. Enable change streams and pre/post images (only if using a custom key)

text:
Change Streams are required only if you are using a custom key in your RDI pipeline. Change streams are available by default on replica sets, sharded clusters, and MongoDB Atlas.

If your RDI pipeline uses a custom key, you must enable pre- and post-images on the relevant collections to capture the document state before and after updates or deletes. This allows RDI to access both the previous and updated versions of documents during change events, ensuring accurate synchronization.

Use the command below to enable change streams and pre/post images:

[code example]

title: 5. MongoDB Atlas specific requirements

text:
MongoDB Atlas only supports secure connections via SSL.
The root CA certificate for MongoDB Atlas must be added as a SOURCE_DB_CACERT secret in RDI.

- Download the MongoDB Atlas root CA certificate.
- In RDI, add this certificate as a secret named SOURCE_DB_CACERT.
- Ensure that the `mongodb.ssl.enabled: true` setting is present in your RDI configuration.

Example connection string for Atlas:
[code example]

title: 6. Network and security

text:
- Ensure the RDI Collector can connect to all MongoDB nodes on the required ports (default: 27017, or as provided by Atlas).
- If using TLS/SSL, provide the necessary certificates and connection options in the connection string.

title: 7. Configuration is complete

text: Once you have followed the steps above, your MongoDB database is ready for Debezium to use.

title: See also

role: related

text:
- [MongoDB Replica Set Documentation](https://www.mongodb.com/docs/manual/replication/)
- [MongoDB Sharded Cluster Documentation](https://www.mongodb.com/docs/manual/sharding/)
- [MongoDB Change Streams](https://www.mongodb.com/docs/manual/changeStreams/)
- [MongoDB User Management](https://www.mongodb.com/docs/manual/core/security-users/)
- [Debezium MongoDB Connector Documentation](https://debezium.io/documentation/reference/stable/connectors/mongodb.html)
- [MongoDB Atlas SSL Setup](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#mongodb-in-the-cloud)

code:
- [ ] [Configure oplog size](#1-configure-oplog-size)
- [ ] [Create a MongoDB user for RDI](#2-create-a-mongodb-user-for-rdi)
- [ ] [Connection string format](#3-connection-string-format)
- [ ] [Enable change streams and pre/post images (only if using a custom key)](#4-enable-change-streams-and-prepost-images-only-if-using-a-custom-key)
- [ ] [MongoDB Atlas specific requirements](#5-mongodb-atlas-specific-requirements)
- [ ] [Network and security](#6-network-and-security)

section_id: summary

code:
use admin;
db.createUser({
  user: "rdi_user",
  pwd: "rdi_password",
  roles: [
     // You can have multiple read roles, one per database.
    { role: "read", db: "your_database" },
    // Use the role below if you don't want to grant the `read` role for each database.
    // { role: "readAnyDatabase", db: "admin" },
    { role: "clusterMonitor", db: "admin" }
  ]
});

code:
db.runCommand({
  collMod: "your_collection",
  changeStreamPreAndPostImages: { enabled: true }
});


title: Prepare MySQL/MariaDB for RDI

title: Overview

text:
The following checklist summarizes the steps to prepare a MySQL or MariaDB
database for RDI, with links to the sections that explain the steps in
full detail. You may find it helpful to track your progress with the
checklist as you complete each step.

[code example]

title: 1. Create a CDC user

text:
The Debezium connector needs a user account to connect to MySQL/MariaDB. This
user must have appropriate permissions on all databases where you want Debezium
to capture changes.

Run the [MySQL CLI client](https://dev.mysql.com/doc/refman/8.3/en/mysql.html)
and then run the following commands:

[code example]

1.
    Create the CDC user:

    [code example]

1.
    Grant the required permissions to the user:

    [code example]

1.
    Finalize the user's permissions:

    [code example]

title: 2. Enable the binlog

text:
You must enable binary logging for MySQL replication. The binary logs record transaction
updates so that replication tools can propagate changes. You will need administrator
privileges to do this.

First, you should check whether the `log-bin` option is already set to `ON`, using
the following query:

[code example]

If `log-bin` is `OFF` then add the following properties to your
server configuration file:

[code example]

You can run the query above again to check that `log-bin` is now `ON`.

If you are using [Amazon RDS for MySQL](https://aws.amazon.com/rds/mysql/) then
you must enable automated backups for your database before it can use binary logging.
If you don't enable automated backups first then the settings above will have no
effect.

title: 3. Enable GTIDs

text:
*Global transaction identifiers (GTIDs)* uniquely identify the transactions that occur
on a server within a cluster. You don't strictly need to use them with a Debezium MySQL
connector, but you might find it helpful to enable them.
Use GTIDs to simplify replication and to confirm that the primary and replica servers are
consistent.

GTIDs are available in MySQL 5.6.5 and later. See the
[MySQL documentation about GTIDs](https://dev.mysql.com/doc/refman/8.0/en/replication-options-gtids.html#option_mysqld_gtid-mode) for more information.

Follow the steps below to enable GTIDs. You will need access to the MySQL configuration file
to do this.

[code example]

1.
    Enable `gtid_mode`:

    [code example]

1.
    Enable `enforce_gtid_consistency`:

    [code example]

1.
    Confirm the changes:

    [code example]

title: 4. Configure session timeouts

text:
RDI captures an initial *snapshot* of the source database when it begins
the CDC process (see the
[architecture overview]()
for more information). If your database is large then the connection could time out
while RDI is reading the data for the snapshot. You can prevent this using the
`interactive_timeout` and `wait_timeout` settings in your MySQL configuration file:

[code example]

title: 5. Enable query log events

text:
If you want to see the original SQL statement for each binlog event then you should
enable `binlog_rows_query_log_events` (MySQL configuration) or
`binlog_annotate_row_events` (MariaDB configuration):

[code example]

This option is available in MySQL 5.6 and later.

title: 6. Check `binlog_row_value_options`

text:
You should check the value of the `binlog_row_value_options` variable
to ensure it is not set to `PARTIAL_JSON`. If it *is* set to
`PARTIAL_JSON` then Debezium might not be able to see `UPDATE` events.

Check the current value of the variable with the following command:

[code example]

If the value is `PARTIAL_JSON` then you should unset the variable:

[code example]

title: 7. Configuration is complete

text:
After following the steps above, your MySQL/MariaDB database is ready
for Debezium to use.

code:
- [ ] [Create a CDC user](#1-create-a-cdc-user)
- [ ] [Enable the binlog](#2-enable-the-binlog)
- [ ] [Enable GTIDs](#3-enable-gtids)
- [ ] [Configure session timeouts](#4-configure-session-timeouts)
- [ ] [Enable query log events](#5-enable-query-log-events)
- [ ] [Check binlog_row_value_options](#6-check-binlog_row_value_options)

language: checklist {id="mysqlmariadb-create-cdc-user" nointeractive="true" }

code:
- [ ] [Create the CDC user](#create-the-cdc-user)
- [ ] [Grant the user the necessary permissions](#grant-the-user-the-necessary-permissions)
- [ ] [Finalize the user's permissions](#finalize-the-users-permissions)

code: mysql> CREATE USER 'user'@'localhost' IDENTIFIED BY 'password';

code:
# MySQL <v8.0
    mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';

    # MySQL v8.0 and above
    mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user'@'localhost';

code: mysql> FLUSH PRIVILEGES;

code:
// for MySql 5.x
mysql> SELECT variable_value as "BINARY LOGGING STATUS (log-bin) ::"
FROM information_schema.global_variables WHERE variable_name='log_bin';
// for MySql 8.x
mysql> SELECT variable_value as "BINARY LOGGING STATUS (log-bin) ::"
FROM performance_schema.global_variables WHERE variable_name='log_bin';

code:
server-id         = 223344 # Querying variable is called server_id, e.g. SELECT variable_value FROM information_schema.global_variables WHERE variable_name='server_id';
log_bin                     = mysql-bin
binlog_format               = ROW
binlog_row_image            = FULL
binlog_expire_logs_seconds  = 864000

language: checklist {id="mysqlmariadb-enable-gtids" nointeractive="true" }

code:
- [ ] [Enable gtid_mode](#enable-gtid_mode)
- [ ] [Enable enforce_gtid_consistency](#enable-enforce_gtid_consistency)
- [ ] [Confirm the changes](#confirm-the-changes)

code: mysql> gtid_mode=ON

code: mysql> enforce_gtid_consistency=ON

code:
mysql> interactive_timeout=<duration-in-seconds>
mysql> wait_timeout=<duration-in-seconds>

code:
mysql> binlog_rows_query_log_events=ON

mariadb> binlog_annotate_row_events=ON

code: mysql> set @@global.binlog_row_value_options="" ;


title: Prepare Neon for RDI

summary: Prepare Neon databases to work with RDI

title: Overview

text:
[Neon](https://neon.com/) is a serverless PostgreSQL platform. To use Neon as a source database for Redis Data Integration (RDI), you must enable logical replication for the compute endpoint that serves your branch, create a dedicated replication role, grant it the minimum required privileges, and allow inbound network traffic from the RDI connector host.

The following checklist summarizes the steps to prepare a Neon database for
RDI, with links to the sections that explain the steps in full detail. You may
find it helpful to track your progress with the checklist as you complete each
step.

[code example]

title: 1. Enable logical replication in Neon

text:
To capture changes from Neon, the compute endpoint for your branch must have
logical replication enabled. This sets the PostgreSQL `wal_level` parameter to
`logical` for that compute.

1. Sign in to the Neon Console and select the project that hosts the database
   you want to use with RDI.
1. On the **Branches** page, choose the branch you want to capture from.
1. Either create a new compute endpoint for that branch or edit an existing
   one.
1. In the compute configuration, enable **Logical replication**.
1. Save the changes and wait for the compute to restart if required.

After the compute is running, connect with a SQL client and confirm that
logical replication is enabled by running:

[code example]

The query should return `logical`.

Enabling logical replication increases the volume of write-ahead log (WAL)
data that Neon retains. Monitor the project for any impact on storage usage and
cost.

title: 2. Create a replication role for RDI

text:
It is strongly recommended to create a dedicated database role for the
connection that RDI uses, rather than reusing an existing superuser or
application role.

Connect to the Neon database as a user with sufficient privileges and create a
role similar to the following (replace the identifiers with values that match
your environment):

[code example]

This role:

- can sign in to the database (`LOGIN`)
- can consume logical replication streams (`REPLICATION`)
- does not have superuser privileges

title: 3. Grant database and table privileges

text:
The replication role must be able to connect to the database, access the
schemas you want to capture, and read from the tables in those schemas.

Run commands like the following, replacing `mydb` and `rdi_replication` with
your database name and replication role:

[code example]

If you use multiple schemas or need access to sequences, repeat the `GRANT`
and `ALTER DEFAULT PRIVILEGES` statements for each schema and include
`SEQUENCES` where appropriate.

title: 4. Allow inbound traffic to Neon

text:
RDI connects to Neon over the public Internet using the connection string for
your compute endpoint. You must ensure that Neon allows inbound traffic from
the IP addresses that the RDI connector uses.

1. In the Neon Console, open the project that contains your database.
1. Go to the **Settings** or **Networking** section for the project and locate
   the IP allow list or **Allowed IPs** configuration.
1. Add the public IP address or address range that the RDI connector uses to
   reach Neon. For production systems, restrict this to the smallest set of
   IPs possible.
1. Save the configuration.

For development or testing, you can temporarily allow access from your own
client machine or a broader IP range, but for production environments you
should always restrict access to known connector IP addresses.

You will also need the Neon connection string for use when you configure the
source in RDI. In the Neon Console, copy the PostgreSQL connection URI for the
compute endpoint you prepared above. It has the form:

[code example]

Use this URI, together with the replication role you created, when you set up
the Neon source connection in RDI.

code:
- [ ] [Enable logical replication in Neon](#1-enable-logical-replication-in-neon)
- [ ] [Create a replication role for RDI](#2-create-a-replication-role-for-rdi)
- [ ] [Grant database and table privileges](#3-grant-database-and-table-privileges)
- [ ] [Allow inbound traffic to Neon](#4-allow-inbound-traffic-to-neon)

code: SHOW wal_level;

code:
CREATE ROLE rdi_replication
  WITH LOGIN REPLICATION PASSWORD 'Strong_Password';

code:
GRANT CONNECT ON DATABASE mydb TO rdi_replication;

GRANT USAGE ON SCHEMA public TO rdi_replication;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO rdi_replication;

ALTER DEFAULT PRIVILEGES IN SCHEMA public
  GRANT SELECT ON TABLES TO rdi_replication;