Configure Debezium Server

Learn how to configure Debezium Server

This document closely follows Debezium's documentation. It explains how to use Debezium Server's application.properties configuration file to configure Debezium Server. You will find similar explanations inside the application.properties file generated by the scaffold command:

redis-di scaffold --db-type cassandra|mysql|oracle|postgresql|sqlserver --dir <PATH_TO_DIR>

Target (sink) connector

The target connector is Redis.

Basic configuration

debezium.sink.type=redis
debezium.sink.redis.address=<REDIS_DI_BDB_HOST>:12001
debezium.sink.redis.password=<REDIS_DI_PASSWORD>

Preventing data loss

To prevent data loss the Debezium Redis Sink Connector needs to be configured to wait for write acknowledgment from the RDI database replica shard. Use the following property to achieve this:

debezium.sink.redis.wait.retry.enabled=true

Additionally, you can configure the wait timeout for replica shard acknowledgment and the delay between write retries (both 1000 milliseconds by default). See the reference section of this document for a full reference.

Redis data integration configuration reference

Notes

• When using Redis to store schema history and offsets, the values of the properties debezium.source.offset.storage.redis.* and debezium.source.schema.history.internal.redis.* will be inherited from the corresponding debezium.sink.redis.* properties.
• If you want to override any of these inherited defaults, add them explicitly as debezium.source.offset.storage.redis.* and/or debezium.source.schema.history.internal.redis.* properties.

Source (database) connector

The source connector depends on the type of database used as the source. The basic configurations are the same for all database types except for the connector class.

Essential source properties

Note: Add debezium.source. prefix to the listed properties when used in application.properties.

Advanced source properties

Note: Add a debezium.source. prefix to the listed properties when used in application.properties.

Debezium connectors for various databases

For additional properties consult Debezium's documentation for a specific connector:

• Cassandra
• MongoDB
• MySQL/MariaDB
• Oracle
• PostgreSQL
• SQL Server

Configure an initial snapshot without filtering queries (relevant for MySQL, Oracle, PostgreSQL and SQLServer)

Tables to be included in the initial snapshot require the property debezium.source.table.include.list. They should be specified as a comma-separated list of fully-qualified table names.

Use queries in initial snapshot (relevant for MySQL, Oracle, PostgreSQL and SQLServer)

• In case you want a snapshot to include only a subset of the rows in a table, you need to add the property debezium.source.snapshot.select.statement.overrides and add a comma-separated list of fully-qualified table names. The list should include every table for which you want to add a SELECT statement.

• For each table in the list above, add a further configuration property that specifies the SELECT statement for the connector to run on the table when it takes a snapshot.

  The specified SELECT statement determines the subset of table rows to include in the snapshot.

  Use the following format to specify the name of this SELECT statement property:

  • Oracle, SQLServer, PostrgreSQL: snapshot.select.statement.overrides.<databaseName>.<tableName>
  • MySQL: snapshot.select.statement.overrides.<schemaName>.<tableName>

• Add a comma-separated list of fully-qualified column names that are included in the SELECT statement: debezium.source.column.include.list = <databaseName>.<tableName>.<columnName1>,<databaseName>.<tableName>.<columnName2>,<databaseName>.<tableName>.<columnName3>...

• In case you want to include all the table columns in the SELECT statement, you can use a regular expression in the form of <databaseName>.<tableName>.* instead of adding each one of the table columns to the debezium.source.column.include.list property.

NOTE: Add all tables as a comma-separated list of fully-qualified table names to the property debezium.source.table.include.list.

Example

To select the columns CustomerId, FirstName, and LastName from customer table and join with the invoice table in order to get customers with total invoices greater than 8000, add the following properties to the application.properties file:

debezium.source.table.include.list = chinook.customer

debezium.source.column.include.list = chinook.customer.CustomerID,chinook.customer.FirstName,chinook.customer.LastName

debezium.source.snapshot.select.statement.overrides=chinook.customer

debezium.source.snapshot.select.statement.overrides.chinook.customer = SELECT c.CustomerId, c.FirstName, c.LastName \
FROM chinook.customer c INNER JOIN chinook.invoice inv \
ON c.CustomerId = inv.CustomerId  \
WHERE inv.total > 8000

Form custom message key(s) for change event records

• By default, Debezium uses the primary key column(s) of a table as the message key for records that it emits. In place of the default, or to specify a key for tables that lack a primary key, you can configure custom message keys based on one or more columns.

• To establish a custom message key for a table, list the table followed by the column to use as the message key. Each list entry takes the following format:

  debezium.source.message.key.columns=<databaseName>.<tableName>:<columnName>
  

• To base a table key on multiple column names, insert commas between the columns names:

  debezium.source.message.key.columns=<databaseName>.<tableName>:<columnName1>,<columnName2>...
  

• The property can include entries for multiple tables. Use a semicolon to separate table entries in the list:

  debezium.source.message.key.columns=<databaseName>.<tableName1>:<columnName1>,<columnName2>;<databaseName>:<tableName>:<columnName1>,<columnName2>
  

Note: In case the property column.include.list is defined in your application.properties file, make sure it includes all the column names that are specified in the property message.key.columns.

Fully-qualified table name

In this document we refer to the fully-qualified table name as <databaseName>.<tableName>. This format is for MySQL databases. For Oracle, SQLServer, and Postgresql databases use <schemaName>.<tableName> instead.

Notes

• You can specify the fully-qualified table name <databaseName>.<tableName> as a regular expression instead of providing the full name of the databaseName and tableName.
• There is no limit to the number of columns that can be used to create custom message keys. However, it’s best to use the minimum required number of columns to specify a unique key.

Examples

• The primary key of the tables customer and employee is CustomerID.

  To establish custom messages keys based on FirstName and LastName for the tables customer and employee, add the following property to the application.properties file:

  debezium.source.database.message.key.columns=chinook.customer:FirstName,LastName;chinook.employee:FirstName,LastName
  

• To specify the columns FirstName and LastName as the message keys for the table chinook.customer in any database add the following property to the application.properties file:

  chinook.customer:FirstName,LastName;(.*).employee:FirstName,LastName
  

Configuring debezium connector to fetch source database secrets from secret store

Providing the source database password in clear text is not an acceptable option. Fortunately, Debezium supports fetching secrets from environment variables or from a file. This is done using a Quarkus feature that provides property expression expansion on configuration values.

An expression string is a mix of plain strings and expression segments, which are wrapped by the sequence ${…​}.

Using environment variables

Property Expressions also work with environment variables, for example:

...
database.password="${MYSQL_PASSWORD}"
...

Getting secrets from a file

In addition, Debezium application.properties supports the ${file:path:key} variable syntax. The path is the absolute path to the file and the key is the property key.

The file storing the secrets should be in a format of a single key=value entry per line, for example:

...
debezium.source.config.providers=file
debezium.source.config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider
database.password="${file:/tmp/debezium/secrets.txt:db-password}"
...