Debezium Server

Debezium Server is a ready-to-use application that streams change events from a source database directly to messaging and storage systems without requiring Apache Kafka. You configure a source connector and a sink to define where events come from and where they are delivered.

Installation

Download and unpack the Debezium Server distribution to install the server on your system.

Procedure

• Debezium Server distribution

A directory named debezium-server is created with the following contents:

debezium-server/
    |-- config
    |-- connectors
    |-- debezium-server-dist-3.6.3.Final-SNAPSHOT-runner.jar
    |-- jmx
    |-- lib
    |-- lib_metrics
    |-- lib_opt
    |-- run.bat
    |-- run.sh

The lib directory stores dependencies, and the config directory contains configuration files. The run.bat or run.sh file starts the server.

Before you start the server, specify the source for the change events that you want to stream and the sink destination for those events. For details about how to configure Debezium Server, see the Configuration section.

Start the server

Run the Debezium Server startup script to begin streaming change events from your configured source connector to the configured sink.

Prerequisites

• Configure Debezium Server as described in Debezium Server configuration.
• To use the Oracle connector, copy the Oracle JDBC driver to the lib directory before you start the server.
• To use Oracle XStream, copy both the Oracle JDBC driver and the required XStream API files to the lib directory. For more information, see Obtaining the Oracle JDBC driver and XStream API files.

Procedure

1. Open a command prompt or terminal window.

2. Change to the <ServerInstallDirectory>/debezium-server directory.

3. Start Debezium Server by entering one of the following commands:

   • On Linux, run the run.sh script:

     ./run.sh

   • On Windows, run the run.bat script:

     run.bat

Debezium Server configuration

Debezium Server uses MicroProfile Configuration for configuration. MicroProfile configuration is a standard for supplying configuration values to Java applications from multiple sources, including configuration files, environment variables, and system properties.

Configuration properties that contain the $ character are normally interpreted by Quarkus for property expansion. If a property value includes a $ character as a prefix, such as ${my.property}, the Microprofile configuration parser determines that the value is a property expression and attempts to expand it.

To prevent the parser from attempting to expand values that use this syntax, escape the $ prefix with a second $ character; that is, provide the value as $${my.property}. The value of the property is then passed to Debezium as ${my.property}.

The main configuration file is config/application.properties. There are multiple sections configured:

• debezium.source is for source connector configuration; each instance of Debezium Server runs exactly one connector
• debezium.sink is for the sink system configuration
• debezium.format is for the output serialization format configuration
• debezium.transforms is for the configuration of message transformations
• debezium.predicates is for the configuration of message transformation predicates

The following example shows an excerpt from a configuration file:

debezium.sink.type=kinesis
debezium.sink.kinesis.region=eu-central-1
debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.offset.flush.interval.ms=0
debezium.source.database.hostname=localhost
debezium.source.database.port=5432
debezium.source.database.user=postgres
debezium.source.database.password=postgres
debezium.source.database.dbname=postgres
debezium.source.topic.prefix=tutorial
debezium.source.schema.include.list=inventory

The configuration file in the preceding example specifies the following settings:

• The sink is set up for AWS Kinesis in region eu-central-1
• The source connector is set up for PostgreSQL using the default Debezium decoderbufs plugin. If using PostgreSQL’s built-in pgoutput plugin, set debezium.source.plugin.name=pgoutput
• The source connector captures events from a schema named inventory. If you want to capture all changes in the database, remove this line. Otherwise, update this line to correspond to your preferred schema or tables.
• The source offset is stored in a file named offsets.dat in the data directory. To prevent startup errors, the directory that you specify must exists before you start the server.

After the server starts, it generates a sequence of log messages similar to the following example:

__  ____  __  _____   ___  __ ____  ______
 --/ __ \/ / / / _ | / _ \/ //_/ / / / __/
 -/ /_/ / /_/ / __ |/ , _/ ,< / /_/ /\ \
--\___\_\____/_/ |_/_/|_/_/|_|\____/___/
2020-05-15 11:33:12,189 INFO  [io.deb.ser.kin.KinesisChangeConsumer] (main) Using 'io.debezium.server.kinesis.KinesisChangeConsumer$$Lambda$119/0x0000000840130c40@f58853c' stream name mapper
2020-05-15 11:33:12,628 INFO  [io.deb.ser.kin.KinesisChangeConsumer] (main) Using default KinesisClient 'software.amazon.awssdk.services.kinesis.DefaultKinesisClient@d1f74b8'
2020-05-15 11:33:12,628 INFO  [io.deb.ser.DebeziumServer] (main) Consumer 'io.debezium.server.kinesis.KinesisChangeConsumer' instantiated
2020-05-15 11:33:12,754 INFO  [org.apa.kaf.con.jso.JsonConverterConfig] (main) JsonConverterConfig values:
    converter.type = key
    decimal.format = BASE64
    schemas.cache.size = 1000
    schemas.enable = true

2020-05-15 11:33:12,757 INFO  [org.apa.kaf.con.jso.JsonConverterConfig] (main) JsonConverterConfig values:
    converter.type = value
    decimal.format = BASE64
    schemas.cache.size = 1000
    schemas.enable = false

2020-05-15 11:33:12,763 INFO  [io.deb.emb.EmbeddedEngine$EmbeddedConfig] (main) EmbeddedConfig values:
    access.control.allow.methods =
    access.control.allow.origin =
    admin.listeners = null
    bootstrap.servers = [localhost:9092]
    client.dns.lookup = default
    config.providers = []
    connector.client.config.override.policy = None
    header.converter = class org.apache.kafka.connect.storage.SimpleHeaderConverter
    internal.key.converter = class org.apache.kafka.connect.json.JsonConverter
    internal.value.converter = class org.apache.kafka.connect.json.JsonConverter
    key.converter = class org.apache.kafka.connect.json.JsonConverter
    listeners = null
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    offset.flush.interval.ms = 0
    offset.flush.timeout.ms = 5000
    offset.storage.file.filename = data/offsets.dat
    offset.storage.partitions = null
    offset.storage.replication.factor = null
    offset.storage.topic =
    plugin.path = null
    rest.advertised.host.name = null
    rest.advertised.listener = null
    rest.advertised.port = null
    rest.extension.classes = []
    rest.host.name = null
    rest.port = 8083
    ssl.client.auth = none
    task.shutdown.graceful.timeout.ms = 5000
    topic.tracking.allow.reset = true
    topic.tracking.enable = true
    value.converter = class org.apache.kafka.connect.json.JsonConverter

2020-05-15 11:33:12,763 INFO  [org.apa.kaf.con.run.WorkerConfig] (main) Worker configuration property 'internal.key.converter' is deprecated and may be removed in an upcoming release. The specified value 'org.apache.kafka.connect.json.JsonConverter' matches the default, so this property can be safely removed from the worker configuration.
2020-05-15 11:33:12,763 INFO  [org.apa.kaf.con.run.WorkerConfig] (main) Worker configuration property 'internal.value.converter' is deprecated and may be removed in an upcoming release. The specified value 'org.apache.kafka.connect.json.JsonConverter' matches the default, so this property can be safely removed from the worker configuration.
2020-05-15 11:33:12,765 INFO  [org.apa.kaf.con.jso.JsonConverterConfig] (main) JsonConverterConfig values:
    converter.type = key
    decimal.format = BASE64
    schemas.cache.size = 1000
    schemas.enable = true

2020-05-15 11:33:12,765 INFO  [org.apa.kaf.con.jso.JsonConverterConfig] (main) JsonConverterConfig values:
    converter.type = value
    decimal.format = BASE64
    schemas.cache.size = 1000
    schemas.enable = true

2020-05-15 11:33:12,767 INFO  [io.deb.ser.DebeziumServer] (main) Engine executor started
2020-05-15 11:33:12,773 INFO  [org.apa.kaf.con.sto.FileOffsetBackingStore] (pool-3-thread-1) Starting FileOffsetBackingStore with file data/offsets.dat
2020-05-15 11:33:12,835 INFO  [io.deb.con.com.BaseSourceTask] (pool-3-thread-1) Starting PostgresConnectorTask with configuration:
2020-05-15 11:33:12,837 INFO  [io.deb.con.com.BaseSourceTask] (pool-3-thread-1)    connector.class = io.debezium.connector.postgresql.PostgresConnector
2020-05-15 11:33:12,837 INFO  [io.deb.con.com.BaseSourceTask] (pool-3-thread-1)    offset.flush.interval.ms = 0
2020-05-15 11:33:12,838 INFO  [io.deb.con.com.BaseSourceTask] (pool-3-thread-1)    database.user = postgres
2020-05-15 11:33:12,838 INFO  [io.deb.con.com.BaseSourceTask] (pool-3-thread-1)    database.dbname = postgres
2020-05-15 11:33:12,838 INFO  [io.deb.con.com.BaseSourceTask] (pool-3-thread-1)    offset.storage.file.filename = data/offsets.dat
2020-05-15 11:33:12,838 INFO  [io.deb.con.com.BaseSourceTask] (pool-3-thread-1)    database.hostname = localhost
2020-05-15 11:33:12,838 INFO  [io.deb.con.com.BaseSourceTask] (pool-3-thread-1)    database.password = ********
2020-05-15 11:33:12,839 INFO  [io.deb.con.com.BaseSourceTask] (pool-3-thread-1)    name = kinesis
2020-05-15 11:33:12,839 INFO  [io.deb.con.com.BaseSourceTask] (pool-3-thread-1)    topic.prefix = tutorial
2020-05-15 11:33:12,839 INFO  [io.deb.con.com.BaseSourceTask] (pool-3-thread-1)    database.port = 5432
2020-05-15 11:33:12,839 INFO  [io.deb.con.com.BaseSourceTask] (pool-3-thread-1)    schema.include.list = inventory
2020-05-15 11:33:12,908 INFO  [io.quarkus] (main) debezium-server 1.2.0-SNAPSHOT (powered by Quarkus 1.4.1.Final) started in 1.198s. Listening on: http://0.0.0.0:8080
2020-05-15 11:33:12,911 INFO  [io.quarkus] (main) Profile prod activated.
2020-05-15 11:33:12,911 INFO  [io.quarkus] (main) Installed features: [cdi, smallrye-health]

Source configuration

Source configuration properties specify which connector Debezium Server runs and configure offset and schema history storage outside of Kafka Connect.

Most source configuration properties correspond to properties that are documented for individual connectors. To use these properties in Debezium Server, prepend the debezium.source. prefix to the connector property name. Debezium Server also provides additional source configuration properties for settings that Kafka Connect normally handles automatically. For example, debezium.source.offset.storage and debezium.source.offset.storage.file.filename to configure file-based offset storage.

Property Default Description

debezium.source.connector.class

The name of the Java class implementing the source connector.

debezium.source.offset.storage

org.apache.kafka.connect.storage.FileOffsetBackingStore

Class to use for storing and retrieving offsets for non-Kafka deployments. Available options

• org.apache.kafka.connect.storage.FileOffsetBackingStore for non-Kafka deployments
• org.apache.kafka.connect.storage.MemoryOffsetBackingStore volatile store for test environments
• io.debezium.storage.jdbc.offset.JdbcOffsetBackingStore for database using JDBC
• io.debezium.storage.redis.offset.RedisOffsetBackingStore for Redis deployments

debezium.source.offset.storage.file.filename

If using a file offset store (default), the file in which connector offsets are stored for non-Kafka deployments.

debezium.source.offset.flush.interval.ms

Defines how frequently the offsets are flushed into the file.

debezium.source.offset.storage.redis.address

(Optional) If using Redis to store offsets, an address, formatted as host:port, at which the Redis target streams are provided. If not supplied, will attempt to read debezium.sink.redis.address

debezium.source.offset.storage.redis.user

(Optional) If using Redis to store offsets, a user name used to communicate with Redis. If the redis.address configuration is not supplied, and the redis.address is taken from the Redis sink, will attempt to load the value from debezium.sink.redis.user

debezium.source.offset.storage.redis.password

(Optional) If using Redis to store offsets, a password (of respective user) used to communicate with Redis. A password must be set if a user is set. If the redis.address configuration is not supplied, and the redis.address is taken from the Redis sink, will attempt to load the value from debezium.sink.redis.password

debezium.source.offset.storage.redis.ssl.enabled

(Optional) If using Redis to store offsets, whether or not to use SSL to communicate with Redis. If the redis.address configuration is not supplied, and the redis.address is taken from the Redis sink, will attempt to load the value from debezium.sink.redis.ssl.enabled. Default is 'false'

debezium.source.offset.storage.redis.ssl.hostname.verification.enabled

(Optional) If using Redis to store offsets, whether or not to enable hostname verification with Redis. If the redis.address configuration is not supplied, and the redis.address is taken from the Redis sink, will attempt to load the value from debezium.sink.redis.ssl.hostname.verification.enabled. Default is 'false'

debezium.source.offset.storage.redis.ssl.truststore.path

(Optional) If using Redis to store offsets with SSL enabled, the path to the trust store file. If set, Redis connections will use this property over other configuration or system properties.

debezium.source.offset.storage.redis.ssl.truststore.password

(Optional) If using Redis to store offsets with SSL enabled, the password for the trust store file. If set, Redis connections will use this property over other configuration or system properties.

debezium.source.offset.storage.redis.ssl.truststore.type

JKS

(Optional) If using Redis to store offsets with SSL enabled, the type of the trust store file. If set, Redis connections will use this property over other configuration or system properties.

debezium.source.offset.storage.redis.ssl.keystore.path

(Optional) If using Redis to store offsets with SSL enabled, the path to the key store file. If set, Redis connections will use this property over other configuration or system properties.

debezium.source.offset.storage.redis.ssl.keystore.password

(Optional) If using Redis to store offsets with SSL enabled, the password for the key store file. If set, Redis connections will use this property over other configuration or system properties.

debezium.source.offset.storage.redis.ssl.keystore.type

JKS

(Optional) If using Redis to store offsets with SSL enabled, the type of the key store file. If set, Redis connections will use this property over other configuration or system properties.

debezium.source.offset.storage.redis.key

(Optional) If using Redis to store offsets, define the hash key in redis. If the redis.key configuration is not supplied, and the default value is metadata:debezium:offsets

debezium.source.offset.storage.redis.wait.enabled

false

If using Redis to store offsets, enables wait for replica. In case Redis is configured with a replica shard, this allows to verify that the data has been written to the replica. For more information see Redis WAIT command.

debezium.source.offset.storage.redis.wait.timeout.ms

1000

If using Redis to store offsets, defines the timeout in milliseconds when waiting for replica. Must have a positive value.

debezium.source.offset.storage.redis.wait.retry.enabled

false

If using Redis to store offsets, enables retry on wait for replica failure.

debezium.source.offset.storage.redis.wait.retry.delay.ms

1000

If using Redis to store offsets, defines the delay of retry on wait for replica failure.

debezium.source.offset.storage.redis.cluster.enabled

false

If you configure Debezium Server to store offsets in Redis, set this property to specify whether to use Redis Cluster mode. Set the value to true to configure Debezium to use a JedisCluster client to route offset data to Redis nodes.

debezium.source.schema.history.internal

io.debezium.storage.kafka.history.KafkaSchemaHistory

Some of the connectors (e.g MySQL, SQL Server, Db2, Oracle) track the database schema evolution over time and stores this data in a database schema history. This is by default based on Kafka. There are also other options available

• io.debezium.storage.file.history.FileSchemaHistory for non-Kafka deployments
• io.debezium.relational.history.MemorySchemaHistory volatile store for test environments
• io.debezium.storage.redis.history.RedisSchemaHistory for Redis deployments
• io.debezium.storage.rocketmq.history.RocketMqSchemaHistory for RocketMQ deployments
• io.debezium.storage.azure.blob.history.AzureBlobSchemaHistory for Azure Blob Storage deployments

debezium.source.schema.history.internal.file.filename

The name and location of the file to which FileSchemaHistory persists its data.

debezium.source.schema.history.internal.redis.address

The Redis host:port to connect to if using RedisSchemaHistory.

debezium.source.schema.history.internal.redis.user

The Redis user to use if using RedisSchemaHistory.

debezium.source.schema.history.internal.redis.password

The Redis password to use if using RedisSchemaHistory.

debezium.source.schema.history.internal.redis.ssl.enabled

Use SSL connection if using RedisSchemaHistory.

debezium.source.schema.history.internal.redis.ssl.hostname.verification.enabled

Enable hostname verification if using RedisSchemaHistory.

debezium.source.schema.history.internal.redis.ssl.truststore.path

(Optional) If using Redis to store schema history with SSL enabled, the path to the trust store file. If set, Redis connections will use this property over other configuration or system properties.

debezium.source.schema.history.internal.redis.ssl.truststore.password

(Optional) If using Redis to store schema history with SSL enabled, the password for the trust store file. If set, Redis connections will use this property over other configuration or system properties.

debezium.source.schema.history.internal.redis.ssl.truststore.type

JKS

(Optional) If using Redis to store schema history with SSL enabled, the type of the trust store file. If set, Redis connections will use this property over other configuration or system properties.

debezium.source.schema.history.internal.redis.ssl.keystore.path

(Optional) If using Redis to store schema history with SSL enabled, the path to the key store file. If set, Redis connections will use this property over other configuration or system properties.

debezium.source.schema.history.internal.redis.ssl.keystore.password

(Optional) If using Redis to store schema history with SSL enabled, the password for the key store file. If set, Redis connections will use this property over other configuration or system properties.

debezium.source.schema.history.internal.redis.ssl.keystore.type

JKS

(Optional) If using Redis to store schema history with SSL enabled, the type of the key store file. If set, Redis connections will use this property over other configuration or system properties.

debezium.source.schema.history.internal.redis.key

The Redis key to use for storage if using RedisSchemaHistory. Default: metadataschema_history

debezium.source.schema.history.internal.redis.retry.initial.delay.ms

The initial delay in case of a connection retry to Redis if using RedisSchemaHistory. Default: 300 (ms)

debezium.source.schema.history.internal.redis.retry.max.delay.ms

The maximum delay in case of a connection retry to Redis if using RedisSchemaHistory. Default: 10000 (ms)

debezium.source.schema.history.internal.redis.retry.max.attempts

The maximum number of attempts to connect to Redis. Default: 10

debezium.source.schema.history.internal.redis.connection.timeout.ms

Connection timeout of Redis client if using RedisSchemaHistory. Default: 2000 (ms)

debezium.source.schema.history.internal.redis.socket.timeout.ms

Socket timeout of Redis client if using RedisSchemaHistory. Default: 2000 (ms)

debezium.source.schema.history.internal.redis.wait.enabled

false

If using Redis to store schema history, enables wait for replica. In case Redis is configured with a replica shard, this allows to verify that the data has been written to the replica. For more information see Redis WAIT command.

debezium.source.schema.history.internal.redis.wait.timeout.ms

1000

If using Redis to store schema history, defines the timeout in milliseconds when waiting for replica. Must have a positive value.

debezium.source.schema.history.internal.redis.wait.retry.enabled

false

If using Redis to store schema history, enables retry on wait for replica failure.

debezium.source.schema.history.internal.redis.wait.retry.delay.ms

1000

If using Redis to store schema history, defines the delay of retry on wait for replica failure.

debezium.source.schema.history.internal.redis.cluster.enabled

false

If you configure Debezium Server to store schema history in Redis, set this property to specify whether to use Redis Cluster mode. Set the value to true to configure Debezium to use a JedisCluster client to route history data to Redis nodes.

debezium.source.schema.history.internal.rocketmq.topic

The name of the rocketmq topic for the database schema history.

debezium.source.schema.history.internal.rocketmq.name.srv.addr

localhost:9876

RocketMQ service discovery NameServer address configuration.

debezium.source.schema.history.internal.rocketmq.acl.enabled

false

RocketMQ access control enable configuration, default is 'false'.

debezium.source.schema.history.internal.rocketmq.access.key

RocketMQ access key. If debezium.source.schema.history.internal.rocketmq.acl.enabled is true, the value cannot be empty.

debezium.source.schema.history.internal.rocketmq.secret.key

RocketMQ secret key. If debezium.source.schema.history.internal.rocketmq.acl.enabled is true, the value cannot be empty.

debezium.source.schema.history.internal.rocketmq.recovery.attempts

60

The maximum number of attempts to recover database schema history.

debezium.source.schema.history.internal.rocketmq.recovery.poll.interval.ms

1000

The number of milliseconds to wait while polling for persisted data during recovery.

debezium.source.schema.history.internal.rocketmq.store.record.timeout.ms

60000

Timeout for sending messages to RocketMQ.

debezium.source.schema.history.internal.azure.storage.account.connectionstring

Azure Blob Storage account connection string.

debezium.source.schema.history.internal.azure.storage.account.name

Azure Blob Storage account name. This should be set if and only if debezium.source.schema.history.internal.azure.storage.account.connectionstring is empty, which will then use Azure Active Directory authentication.

debezium.source.schema.history.internal.azure.storage.account.blob.endpoint

An optional Azure Blob Storage account endpoint URL. Use this to override the default public endpoint (https://<account>.blob.core.windows.net) for sovereign clouds such as Azure Government (https://<account>.blob.core.usgovcloudapi.net) or Azure China (https://<account>.blob.core.chinacloudapi.cn). Only used when debezium.source.schema.history.internal.azure.storage.account.connectionstring is not set. Should not be set together with debezium.source.schema.history.internal.azure.storage.account.name.

debezium.source.schema.history.internal.azure.storage.account.container.name

Azure Blob Storage account container name.

debezium.source.schema.history.internal.azure.storage.blob.name

Azure Blob Storage blob name that persists schema history data.

Format configuration

Configure how Debezium Server serializes message keys and values. You can configure key and value formats independently by using either the default JSON format or a custom Kafka Connect Converter implementation.

Transformation configuration

Configure Kafka Connect Single Message Transformations (SMTs) that Debezium Server applies to messages before it delivers them to the sink. Specify the transformations to apply and the configuration options for each transformation.

The server supports single message transformations defined by Kafka Connect.

Predicates configuration

Configure predicates that evaluate messages and determine when Debezium Server applies message transformations. Debezium Server supports the Kafka Connect predicate framework for conditional Single Message Transformations (SMTs). Predicate properties specify the predicates to use, their implementation classes, and their configuration options.

Enabling message filtering

You can enable the filter SMT in Debezium Server to selectively pass or discard change events based on expression-based conditions that you define.

Because filter expressions are evaluated by a scripting engine running inside the Debezium Server process, this feature is only safe to enable in environments where access to the server configuration is restricted to authorized operators. Scripting support is therefore disabled by default and must be explicitly enabled.

To enable message filtering, set the environment variable ENABLE_DEBEZIUM_SCRIPTING to true before starting Debezium Server. When yhou enable this setting, the startup script adds the debezium-scripting JAR and the JSR 223 script engine implementations (Groovy and GraalVM JavaScript) from the opt_lib directory of the Debezium Server distribution to the server classpath. These libraries are kept in opt_lib rather than on the default classpath precisely so that scripting is never available unless an operator makes a conscious decision to include them.

Additional resources

• Debezium message filtering SMT.

Asynchronous Engine Properties

By default, Debezium Server uses the asynchronous embedded engine (AsyncEmbeddedEngine) to process change events. Use the following properties to configure record-processing threads, task lifecycle management, and engine shutdown behavior.

Additional configuration

Debezium Server runs on the Quarkus framework. All configuration properties that Quarkus exposes are available in Debezium Server. The following table lists commonly used Quarkus properties for configuring HTTP endpoint access and logging behavior.

You can disable JSON logging by setting quarkus.log.console.json=false in the config/application.properties file. For details, see the example in the file.

Sink configuration

Debezium Server supports implementations for delivering change events to different sink destinations. Configure the destination to which Debezium Server delivers change events. Set the debezium.sink.type property to specify the sink implementation, and then configure the properties that apply to the selected sink.

Amazon Kinesis

Amazon Kinesis is an implementation of data streaming system with support for stream sharding and other techniques for high scalability. Kinesis exposes a set of REST APIs and provides a (not-only) Java SDK that is used to implement the sink.

Injection points

The Kinesis sink behaviour can be modified by a custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

Google Cloud Pub/Sub

Google Cloud Pub/Sub is an implementation of messaging/eventing system designed for scalable batch and stream processing applications. Pub/Sub exposes a set of REST APIs and provides a (not-only) Java SDK that is used to implement the sink.

Injection points

The Pub/Sub sink behaviour can be modified by a custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

Pub/Sub Lite

Google Cloud Pub/Sub Lite is a cost-effective alternative to Google Cloud Pub/Sub. Pub/Sub exposes a set of REST APIs and provides a (not-only) Java SDK that is used to implement the sink.

Injection points

The Pub/Sub Lite sink behaviour can be modified by a custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

HTTP Client

Configure the HTTP Client sink to deliver change events from Debezium Server to an HTTP endpoint for downstream processing. The HTTP Client sink supports optional JWT and OAuth 2.0 client credentials authentication, and provides options for configuring batched event delivery.

You can use the HTTP Client sink to integrate Debezium Server with Knative and enable it to act as a Knative Event Source.

Property Default Description

debezium.sink.type

Must be set to http

debezium.sink.http.url

The HTTP Server URL to stream events to. This can also be set by defining the K_SINK environment variable, which is used by the Knative source framework.

debezium.sink.http.timeout.ms

60000

The number of seconds to wait for a response from the server before timing out. (default of 60s)

debezium.sink.http.retries

5

The number retries before exception is thrown (default 5 times).

debezium.sink.http.retry.interval.ms

1000

The number of milliseconds to wait before another attempt to send record is made after failure (default of 1s).

debezium.sink.http.headers.prefix

X-DEBEZIUM-

Headers will be prefixed with this value (defaults to X-DEBEZIUM-).

debezium.sink.http.headers.encode.base64

true

Header values will be base64 encoded (defaults to true).

debezium.sink.http.batch.enabled

false

When set to true, aggregates all change events in a batch into a JSON array and sends them in a single HTTP POST request instead of sending each event individually.

debezium.sink.http.batch.max-size

200

The maximum number of events per HTTP request when batch mode is enabled. If the engine delivers more events than this limit, they are chunked into multiple requests.

debezium.sink.http.authentication.type

Specifies the type of authentication the HTTP client sink uses when connecting to an HTTP server. Supports one of the following options:

jwt

JSON Web Token (JWT) authentication.

oauth2

OAuth2 client credentials grant (RFC 6749 Section 4.4).

standard-webhooks

Standard Webhooks.

If you omit this property, the HTTP client sink does not use authentication headers for the connection

debezium.sink.http.authentication.jwt.username

Specifies the username for JWT authentication.

debezium.sink.http.authentication.jwt.password

Specifies the password for JWT authentication.

debezium.sink.http.authentication.jwt.url

Specifies the base URL (e.g., http://myserver:8000/) for JWT authentication. The paths auth/authenticate and auth/refreshToken are appended for the JWT initial and authentication REST requests.

debezium.sink.http.authentication.jwt.token_expiration

Requested duration (in minutes) before the authentication token expires.

debezium.sink.http.authentication.jwt.refresh_token_expiration

Requested duration (in minutes) before the refresh token expires.

debezium.sink.http.authentication.webhook.secret

The webhooks signing secret that Debezium uses to generate HMAC-SHA256 signatures for the webhook request. The secret must be Base64-encoded, with a size from 24 bytes to 64 bytes (192–512 bits). Optionally, you can add the prefix whsec_ to the secret to help distinguish it from other types of keys or tokens. For more information about implementing or validating webhook signatures, see the Standard Webhooks specification.

debezium.sink.http.authentication.oauth2.client_id

Specifies the OAuth2 client ID for the client credentials grant.

debezium.sink.http.authentication.oauth2.client_secret

Specifies the OAuth2 client secret for the client credentials grant.

debezium.sink.http.authentication.oauth2.token_url

The URL of the OAuth2 token endpoint (e.g., https://auth.example.com/oauth/token).

debezium.sink.http.authentication.oauth2.scope

Optional space-separated list of OAuth2 scopes to request (e.g., data:read data:write).

debezium.sink.http.authentication.oauth2.client_auth_method

client_secret_basic

Specifies how client credentials are sent to the token endpoint. Supports one of the following options:

client_secret_basic

Sends credentials as an HTTP Basic Authorization header (RFC 6749 Section 2.3.1).

client_secret_post

Sends client_id and client_secret as form fields in the POST request body.

debezium.sink.http.authentication.oauth2.token_url.http_method

POST

The HTTP method to use for token requests. The OAuth2 specification requires POST, but some providers accept GET with query parameters. Set to GET only for providers that do not support the standard POST method.

debezium.sink.http.authentication.oauth2.params.*

Optional additional parameters to include in the token request. For example, some providers require an audience or resource parameter. Set these using the params. prefix, for example, debezium.sink.http.authentication.oauth2.params.audience=\https://my-api.example.com.

Apache Pulsar

Apache Pulsar is high-performance, low-latency server for server-to-server messaging. Pulsar exposes a REST APIs and a native endpoint provides a (not-only) Java client that is used to implement the sink.

Injection points

The Pulsar sink behaviour can be modified by a custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

Azure Event Hubs

Azure Event Hubs is a big data streaming platform and event ingestion service that can receive and process millions of events per second. Data sent to an event hub can be transformed and stored by using any real-time analytics provider or batching/storage adapters.

Property Default Description

debezium.sink.type

Must be set to eventhubs.

debezium.sink.eventhubs.connectionstring

Connection string required to communicate with Event Hubs. The format is: Endpoint=sb://<NAMESPACE>/;SharedAccessKeyName=<ACCESS_KEY_NAME>;SharedAccessKey=<ACCESS_KEY_VALUE>

debezium.sink.eventhubs.hubname

Name of the Event Hub

debezium.sink.eventhubs.dynamicpartitionrouting

An optional setting that controls the behavior of dynamic partition routing. There are three possible values:

• default: If the record key is not null, it is used as the batch partition key. Else, if the record’s partition ID is not null, events are batched and routed to the specified partition ID. Otherwise, round-robin routing is applied. This is the default when no value is provided.
• key: If the record key is not null, it is used as the batch partition key. Otherwise, round-robin routing is applied.
• partitionid: If the record’s partition ID is not null, events are batched and routed to the specified partition ID. Otherwise, round-robin routing is applied.

The debezium.sink.eventhubs.dynamicpartitionrouting configuration is only applied when both of the following options are unset:

• debezium.sink.eventhubs.partitionid
• debezium.sink.eventhubs.partitionkey

debezium.sink.eventhubs.partitionid

(Optional) The identifier of the Event Hub partition that the events will be sent to. Use this if you want all the change events received by Debezium to be sent to a specific partition in Event Hubs. Do not use if you have specified debezium.sink.eventhubs.partitionkey

debezium.sink.eventhubs.partitionkey

(Optional) The partition key will be used to hash the events. Use this if you want all the change events received by Debezium to be sent to a specific partition in Event Hubs. Do not use if you have specified debezium.sink.eventhubs.partitionid

debezium.sink.eventhubs.maxbatchsize

Sets the maximum size for the batch of events, in bytes.

debezium.sink.eventhubs.hashmessagekeyfunction

No default value

(Optional) Specifies the hash function that Debezium uses to encrypt encrypt Azure Event Hubs message keys.

Specify one of the following values:

• java
• md5
• sha1
• sha256

Using partitions in EventHubs

Azure Event Hubs supports multiple partition routing strategies that control how Debezium distributes change events across partitions. You can specify fixed partition IDs, partition keys, or the Partition Routing transformer to match the throughput and ordering requirements in your deployment.

By default, when neither of the optional debezium.sink.eventhubs.partitionid or debezium.sink.eventhubs.partitionkey properties are defined, the EventHubs sink will send events round-robin to all available partitions.

You can enforce all messages to be sent to a single, fixed, partition by setting the debezium.sink.eventhubs.partitionid property. Alternatively, you can use the debezium.sink.eventhubs.partitionkey property to specify a fixed partition key that EventHubs will use to route all events to a specific partition.

If you have more specific routing requirements you can use the Partition Routing transformer. Ensure that the number of partitions specified in the transformer’s partition.topic.num setting is equal or less to the number of partitions available in your EventHubs namespace, so that events cannot be routed to non-existing partition IDs. As an example, to route all events to 5 partitions based on their source schema name, you can set the following in your application.properties:

# Uses a hash of `source.db` to calculate which partition to send the event to. Ensures all events from the same source schema are sent to the same partition.
debezium.transforms=PartitionRouter
debezium.transforms.PartitionRouter.type=io.debezium.transforms.partitions.PartitionRouting
debezium.transforms.PartitionRouter.partition.payload.fields=source.db
debezium.transforms.PartitionRouter.partition.topic.num=5

Injection points

The default sink behaviour can be modified by a custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

Redis (Stream)

The Redis sink stores change events as entries in a Redis stream, supporting both single-instance and cluster deployments. You configure the sink by setting the debezium.sink.type property to redis and specifying connection, authentication, and storage options.

Redis is an open source (BSD licensed) in-memory data structure store, used as a database, cache and message broker. The Stream is a data type which models a log data structure in a more abstract way. It implements powerful operations to overcome the limitations of a log file.

Debezium can use either a single Redis instance or use Redis Cluster mode for sink operations:

Single instance mode

Connects to a single Redis server instance.

Cluster mode

Connects to a Redis Cluster for high availability and horizontal scaling.

To enable Redis Cluster mode, set the debezium.sink.redis.cluster.enabled property to true and provide comma-separated host:port addresses in the debezium.sink.redis.address property.

Message Format

The Redis sink supports two message formats, compact and extended, that differ in how the key and value are stored within a Redis stream entry.

Set the redis.message.format property to specify which format to use when sending message to a Redis sink.

• the extended format, using two pairs {1), 2)}={"key", "message key"} and {3), 4)}={"value", "message value"}:

1) 1) "1639304527499-0"
   2) 1) "key"
      2) "{\"schema\": {\"type\": \"struct\", \"fields\": [{\"type\": \"int32\", \"optional\": false, \"field\": \"empno\"}], \"optional\": false, \"name\": \"redislabs.dbo.emp.Key\"}, \"payload\": {\"empno\": 11}}"
      3) "value"
      4) "{\"schema\": {\"type\": \"struct\", \"fields\": [{\"type\": \"struct\", \"fields\": [{\"type\": \"int32\", \"optional\": false, \"field\": \"empno\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"fname\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"lname\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"job\"}, {\"type\": \"int32\", \"optional\": true, \"field\": \"mgr\"}, {\"type\": \"int64\", \"optional\": true, \"name\": \"io.debezium.time.Timestamp\", \"version\": 1, \"field\": \"hiredate\"}, {\"type\": \"bytes\", \"optional\": true, \"name\": \"org.apache.kafka.connect.data.Decimal\", \"version\": 1, \"parameters\": {\"scale\": \"4\", \"connect.decimal.precision\": \"19\"}, \"field\": \"sal\"}, {\"type\": \"bytes\", \"optional\": true, \"name\": \"org.apache.kafka.connect.data.Decimal\", \"version\": 1, \"parameters\": {\"scale\": \"4\", \"connect.decimal.precision\": \"19\"}, \"field\": \"comm\"}, {\"type\": \"int32\", \"optional\": true, \"field\": \"dept\"}], \"optional\": true, \"name\": \"redislabs.dbo.emp.Value\", \"field\": \"before\"}, {\"type\": \"struct\", \"fields\": [{\"type\": \"int32\", \"optional\": false, \"field\": \"empno\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"fname\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"lname\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"job\"}, {\"type\": \"int32\", \"optional\": true, \"field\": \"mgr\"}, {\"type\": \"int64\", \"optional\": true, \"name\": \"io.debezium.time.Timestamp\", \"version\": 1, \"field\": \"hiredate\"}, {\"type\": \"bytes\", \"optional\": true, \"name\": \"org.apache.kafka.connect.data.Decimal\", \"version\": 1, \"parameters\": {\"scale\": \"4\", \"connect.decimal.precision\": \"19\"}, \"field\": \"sal\"}, {\"type\": \"bytes\", \"optional\": true, \"name\": \"org.apache.kafka.connect.data.Decimal\", \"version\": 1, \"parameters\": {\"scale\": \"4\", \"connect.decimal.precision\": \"19\"}, \"field\": \"comm\"}, {\"type\": \"int32\", \"optional\": true, \"field\": \"dept\"}], \"optional\": true, \"name\": \"redislabs.dbo.emp.Value\", \"field\": \"after\"}, {\"type\": \"struct\", \"fields\": [{\"type\": \"string\", \"optional\": false, \"field\": \"version\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"connector\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"name\"}, {\"type\": \"int64\", \"optional\": false, \"field\": \"ts_ms\"}, {\"type\": \"string\", \"optional\": true, \"name\": \"io.debezium.data.Enum\", \"version\": 1, \"parameters\": {\"allowed\": \"true,last,false\"}, \"default\": \"false\", \"field\": \"snapshot\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"db\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"sequence\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"schema\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"table\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"change_lsn\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"commit_lsn\"}, {\"type\": \"int64\", \"optional\": true, \"field\": \"event_serial_no\"}], \"optional\": false, \"name\": \"io.debezium.connector.sqlserver.Source\", \"field\": \"source\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"op\"}, {\"type\": \"int64\", \"optional\": true, \"field\": \"ts_ms\"}, {\"type\": \"struct\", \"fields\": [{\"type\": \"string\", \"optional\": false, \"field\": \"id\"}, {\"type\": \"int64\", \"optional\": false, \"field\": \"total_order\"}, {\"type\": \"int64\", \"optional\": false, \"field\": \"data_collection_order\"}], \"optional\": true, \"field\": \"transaction\"}], \"optional\": false, \"name\": \"redislabs.dbo.emp.Envelope\"}, \"payload\": {\"before\": {\"empno\": 11, \"fname\": \"Yossi\", \"lname\": \"Mague\", \"job\": \"PFE\", \"mgr\": 1, \"hiredate\": 1562630400000, \"sal\": \"dzWUAA==\", \"comm\": \"AYag\", \"dept\": 3}, \"after\": null, \"source\": {\"version\": \"1.6.0.Final\", \"connector\": \"sqlserver\", \"name\": \"redislabs\", \"ts_ms\": 1637859764960, \"snapshot\": \"false\", \"db\": \"RedisConnect\", \"sequence\": null, \"schema\": \"dbo\", \"table\": \"emp\", \"change_lsn\": \"0000003a:00002f50:0002\", \"commit_lsn\": \"0000003a:00002f50:0005\", \"event_serial_no\": 1}, \"op\": \"d\", \"ts_ms\": 1637859769370, \"transaction\": null}}"

• and the compact format, using only one pair {1), 2)}={"message key", "message value"}:

1) 1) "1639304527499-0"
   2) 1) "{\"schema\": {\"type\": \"struct\", \"fields\": [{\"type\": \"int32\", \"optional\": false, \"field\": \"empno\"}], \"optional\": false, \"name\": \"redislabs.dbo.emp.Key\"}, \"payload\": {\"empno\": 11}}"
      2) "{\"schema\": {\"type\": \"struct\", \"fields\": [{\"type\": \"struct\", \"fields\": [{\"type\": \"int32\", \"optional\": false, \"field\": \"empno\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"fname\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"lname\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"job\"}, {\"type\": \"int32\", \"optional\": true, \"field\": \"mgr\"}, {\"type\": \"int64\", \"optional\": true, \"name\": \"io.debezium.time.Timestamp\", \"version\": 1, \"field\": \"hiredate\"}, {\"type\": \"bytes\", \"optional\": true, \"name\": \"org.apache.kafka.connect.data.Decimal\", \"version\": 1, \"parameters\": {\"scale\": \"4\", \"connect.decimal.precision\": \"19\"}, \"field\": \"sal\"}, {\"type\": \"bytes\", \"optional\": true, \"name\": \"org.apache.kafka.connect.data.Decimal\", \"version\": 1, \"parameters\": {\"scale\": \"4\", \"connect.decimal.precision\": \"19\"}, \"field\": \"comm\"}, {\"type\": \"int32\", \"optional\": true, \"field\": \"dept\"}], \"optional\": true, \"name\": \"redislabs.dbo.emp.Value\", \"field\": \"before\"}, {\"type\": \"struct\", \"fields\": [{\"type\": \"int32\", \"optional\": false, \"field\": \"empno\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"fname\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"lname\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"job\"}, {\"type\": \"int32\", \"optional\": true, \"field\": \"mgr\"}, {\"type\": \"int64\", \"optional\": true, \"name\": \"io.debezium.time.Timestamp\", \"version\": 1, \"field\": \"hiredate\"}, {\"type\": \"bytes\", \"optional\": true, \"name\": \"org.apache.kafka.connect.data.Decimal\", \"version\": 1, \"parameters\": {\"scale\": \"4\", \"connect.decimal.precision\": \"19\"}, \"field\": \"sal\"}, {\"type\": \"bytes\", \"optional\": true, \"name\": \"org.apache.kafka.connect.data.Decimal\", \"version\": 1, \"parameters\": {\"scale\": \"4\", \"connect.decimal.precision\": \"19\"}, \"field\": \"comm\"}, {\"type\": \"int32\", \"optional\": true, \"field\": \"dept\"}], \"optional\": true, \"name\": \"redislabs.dbo.emp.Value\", \"field\": \"after\"}, {\"type\": \"struct\", \"fields\": [{\"type\": \"string\", \"optional\": false, \"field\": \"version\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"connector\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"name\"}, {\"type\": \"int64\", \"optional\": false, \"field\": \"ts_ms\"}, {\"type\": \"string\", \"optional\": true, \"name\": \"io.debezium.data.Enum\", \"version\": 1, \"parameters\": {\"allowed\": \"true,last,false\"}, \"default\": \"false\", \"field\": \"snapshot\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"db\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"sequence\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"schema\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"table\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"change_lsn\"}, {\"type\": \"string\", \"optional\": true, \"field\": \"commit_lsn\"}, {\"type\": \"int64\", \"optional\": true, \"field\": \"event_serial_no\"}], \"optional\": false, \"name\": \"io.debezium.connector.sqlserver.Source\", \"field\": \"source\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"op\"}, {\"type\": \"int64\", \"optional\": true, \"field\": \"ts_ms\"}, {\"type\": \"struct\", \"fields\": [{\"type\": \"string\", \"optional\": false, \"field\": \"id\"}, {\"type\": \"int64\", \"optional\": false, \"field\": \"total_order\"}, {\"type\": \"int64\", \"optional\": false, \"field\": \"data_collection_order\"}], \"optional\": true, \"field\": \"transaction\"}], \"optional\": false, \"name\": \"redislabs.dbo.emp.Envelope\"}, \"payload\": {\"before\": {\"empno\": 11, \"fname\": \"Yossi\", \"lname\": \"Mague\", \"job\": \"PFE\", \"mgr\": 1, \"hiredate\": 1562630400000, \"sal\": \"dzWUAA==\", \"comm\": \"AYag\", \"dept\": 3}, \"after\": null, \"source\": {\"version\": \"1.6.0.Final\", \"connector\": \"sqlserver\", \"name\": \"redislabs\", \"ts_ms\": 1637859764960, \"snapshot\": \"false\", \"db\": \"RedisConnect\", \"sequence\": null, \"schema\": \"dbo\", \"table\": \"emp\", \"change_lsn\": \"0000003a:00002f50:0002\", \"commit_lsn\": \"0000003a:00002f50:0005\", \"event_serial_no\": 1}, \"op\": \"d\", \"ts_ms\": 1637859769370, \"transaction\": null}}"

You can read more about Redis Streams here.

Injection points

The Redis sink behavior can be modified by a custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

Redis Configuration Examples

The following examples show complete application.properties configurations for connecting Debezium Server to a Redis sink in both single-instance and cluster deployment modes.

Single Instance Mode

Use the following configuration when connecting to a single Redis instance for sink, offset storage, and schema history.

# Sink configuration
debezium.sink.type=redis
debezium.sink.redis.address=localhost:6379
debezium.sink.redis.password=password
debezium.sink.redis.cluster.enabled=false

# Offset storage configuration
debezium.source.offset.storage=io.debezium.storage.redis.offset.RedisOffsetBackingStore
debezium.source.offset.storage.redis.address=localhost:6379
debezium.source.offset.storage.redis.password=password
debezium.source.offset.storage.redis.cluster.enabled=false

# Schema history storage configuration
debezium.source.schema.history.internal=io.debezium.storage.redis.history.RedisSchemaHistory
debezium.source.schema.history.internal.redis.address=localhost:6379
debezium.source.schema.history.internal.redis.password=password
debezium.source.schema.history.internal.redis.cluster.enabled=false

Cluster Mode

Use the following configuration when connecting Debezium Server to a Redis Cluster, specifying multiple node addresses for sink, offset storage, and schema history.

# Sink configuration
debezium.sink.type=redis
debezium.sink.redis.address=redis-node-1:7001,redis-node-2:7002,redis-node-3:7003
debezium.sink.redis.password=password
debezium.sink.redis.cluster.enabled=true

# Offset storage configuration
debezium.source.offset.storage=io.debezium.storage.redis.offset.RedisOffsetBackingStore
debezium.source.offset.storage.redis.address=redis-node-1:7001,redis-node-2:7002,redis-node-3:7003
debezium.source.offset.storage.redis.password=password
debezium.source.offset.storage.redis.cluster.enabled=true

# Schema history storage configuration
debezium.source.schema.history.internal=io.debezium.storage.redis.history.RedisSchemaHistory
debezium.source.schema.history.internal.redis.address=redis-node-1:7001,redis-node-2:7002,redis-node-3:7003
debezium.source.schema.history.internal.redis.password=password
debezium.source.schema.history.internal.redis.cluster.enabled=true

NATS Streaming

NATS Streaming is a data streaming system powered by NATS, and written in the Go programming language.

Injection points

The NATS Streaming sink behavior can be modified by a custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

NATS JetStream

NATS has a built-in distributed persistence system called JetStream which enables new functionalities and higher qualities of service on top of the base 'Core NATS' functionalities and qualities of service.

Property Default Description

debezium.sink.type

Must be set to nats-jetstream.

debezium.sink.nats-jetstream.stream-name

DebeziumStream

Optional custom name for the stream that Debezium creates.

debezium.sink.nats-jetstream.url

URL (or comma separated list of URLs) to a node or nodes in the cluster formatted as nats://host:port.

debezium.sink.nats-jetstream.create-stream

If true, a basic stream will be created.

debezium.sink.nats-jetstream.subjects

*.*.*

A comma separated list of JetStream subjects, or messaging channel names. You can specify entries that contain wildcards, such as test.inventory.*

Data change events are published to table-specific subjects (for example, myapp.database.table) The value myapp.> matches any number of subject levels; similarly myapp.. matches exactly two levels after the prefix.

debezium.sink.nats-jetstream.storage

memory

Controls how the messages are saved in the stream. Can be memory or file.

debezium.sink.nats-jetstream.auth.jwt

No default value

Specifies the identity of the NATS server client. Add this property to the configuration to enable JSON Web Token (JWT) authentication with NATS. To use JWT authentication with NATS, you must specify the NKey seed. Do not enable JWT authentication, if password authentication is enabled.

debezium.sink.nats-jetstream.auth.seed

No default value

When JWT authentication is enabled for NATS, use this property so specify the NKey seed that represents the Debezium user. Debezium uses the specified NKey seed to derive a private key. It then uses the private key to cryptographically sign the nonce challenge that the NATS server issues during the authentication process. Debezium returns the signed nonce to the server, along with the public key for the specified debezium.sink.nats-jetstream.auth.jwt client.

debezium.sink.nats-jetstream.auth.user

No default value

Specifies the username of the authorized NAT user. When this property is present in the configuration, password authentication with NATS is enabled. To use password authentication with NATS, specify a debezium.sink.nats-jetstream.auth.password. Do not enable password authentication if JWT authentication is enabled.

debezium.sink.nats-jetstream.auth.password

No default value

Specifies the password to use when password authentication is enabled.

debezium.sink.nats-jetstream.async.enabled

true

(Optional) A Boolean value that specifies whether Debezium can stream asynchronously to a NATS JetStream server.

debezium.sink.nats-jetstream.async.timeout.ms

5000

(Optional) Specifies the maximum time, in milliseconds, that Debezium waits for acknowledgment from the NATS server after it sends a batch of messages for asynchronous processing. During asynchronous processing, each message is published with a timeout specified by asyncTimeoutMs.

If you need a more configurable stream, it can be created with nats cli. More about streams at: https://docs.nats.io/nats-concepts/jetstream/streams

Injection points

The NATS JetStream sink behavior can be modified by a custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

Apache Fluss

Apache Fluss is a streaming storage built for real-time analytics, providing unified streaming and batch processing capabilities with high performance, low latency, and strong consistency guarantees. Debezium Server can publish change events to Apache Fluss, supporting both primary key tables for upsert/delete operations and log tables for append-only streaming.

Property Default Description

debezium.sink.type

Must be set to fluss.

debezium.sink.fluss.bootstrap.servers

Comma-separated list of Fluss coordinator server addresses in host:port format.

debezium.sink.fluss.default.database

Default Fluss database used when resolving table paths from event destinations.

debezium.sink.fluss.primary.key.mode

auto

Determines the write mode selection based on primary key presence. Valid values:

auto

Automatically selects between append and upsert mode based on whether the table has a primary key.

upsert

Requires tables to have a primary key and performs insert/update/delete operations.

append

Writes all records in append-only mode, regardless of primary key existence.

debezium.sink.fluss.table.auto.create

false

When true, automatically creates Fluss tables if they do not exist, deriving the schema from the Debezium event schema.

debezium.sink.fluss.retries.max

5

Maximum number of retry attempts on transient write failures.

debezium.sink.fluss.retries.interval.ms

1000

The initial retry interval in milliseconds.

debezium.sink.fluss.retries.max.interval.ms

60000

The maximum retry interval in milliseconds.

debezium.sink.fluss.retries.backoff.multiplier

2.0

Multiplier applied to retry intervals during exponential backoff.

Injection points

The Fluss sink behavior can be modified by a custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

Apache Kafka

Configure the Apache Kafka sink to publish change events from Debezium Server to Kafka topics. Use the sink properties to specify broker connection details and other Kafka-specific settings.

Injection points

Customize Kafka sink behavior by providing implementations of supported extension points. For example, you can implement a custom StreamNameMapper to control how Debezium Server maps source streams to Kafka topics. If you do not provide a custom implementation, Debezium Server uses the default implementation.

Pravega

Pravega is a cloud-native storage system for event streams and data streams. This sink offers two modes: non-transactional and transactional. The non-transactional mode individually writes each event in a Debezium batch to Pravega. The transactional mode writes the Debezium batch to a Pravega transaction that commits when the batch is completed.

The Pravega sink expects destination scope and streams to already be created.

Injection points

Pravega sink behavior can be modified by custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

Infinispan

Infinispan is open-source in-memory data grid that offers rich set of caches types as well as cache stores. Due to very fast data access, Infinispan can be used, besides others, as a data source for various data processing and analytical tools.

The Infinispan sink expects that the destination cache is already defined and created within the Infinispan cluster.

Injection points

The Infinispan sink behaviour can be modified by a custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

Apache RocketMQ

Apache RocketMQ is a distributed messaging and streaming platform with low latency, high performance and reliability, trillion-level capacity and flexible scalability. Debezium server supports publishing captured change events to a configured RocketMQ.

Injection points

The RocketMQ sink behaviour can be modified by a custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

RabbitMQ Stream

RabbitMQ is an open source message broker, supporting multiple messaging protocols and can be deployed in distributed and federated configurations to meet high-scale, high-availability requirements. RabbitMQ supports messages queues and streams. Debezium Server supports publishing captured change events to a configured RabbitMQ Stream.

Property Default Description

debezium.sink.type

Must be set to rabbitmq.

debezium.sink.rabbitmq.connection.host

localhost

Host of RabbitMQ server.

debezium.sink.rabbitmq.connection.port

5672

Port of RabbitMQ server.

debezium.sink.rabbitmq.connection.*

The RabbitMQ module supports pass-through configuration. The connection configuration properties are passed to the RabbitMQ client with the prefix removed.

debezium.sink.rabbitmq.ackTimeout

30000

Defines the maximum time in milliseconds to wait a confirm from the broker after publishing a message.

debezium.sink.rabbitmq.exchange

topic name

(Optional) Exchange name to use when publishing messages.

debezium.sink.rabbitmq.routingKey

empty string

(Optional) Static routing key to use when publishing messages.

debezium.sink.rabbitmq.autoCreateRoutingKey

false

(Optional) If true the non-existing routing key is automatically created.

debezium.sink.rabbitmq.routingKeyDurable

true

(Optional) If true the target queue content will survive a RabbitMQ server restart.

debezium.sink.rabbitmq.routingKeyFromTopicName

false

(Optional) Deprecated, see debezium.sink.rabbitmq.routingKey.source.

debezium.sink.rabbitmq.deliveryMode

2

(Optional) The way how the message is delivered to and stored on a RabbitMQ server

• 1 - Non-persistent
• 2 - Persistent

debezium.sink.rabbitmq.null.value

default

RabbitMQ does not support the notion of null payloads, as is the case with tombstone events. So this string will be used as value for records without a payload.

debezium.sink.rabbitmq.routingKey.source

static

(Optional) The way the routing key for the event is going to be obtained.

• static (default): the routing key will be obtained from debezium.sink.rabbitmq.routingKey.
• topic: the routing key is the same as the exchange name.
• key: the routing key will be obtained from the record key.

Injection points

RabbitMQ sink behavior can be modified by custom logic providing alternative implementations for specific functionalities. When the alternative implementations are not available then the default ones are used.

RabbitMQ Native Stream

Since RabbitMQ 3.9, Streams were introduced to RabbitMQ, utilizing a new blazingly-fast protocol that can be used alongside AMQP 0.9.1. Streams are great for large fan-outs, replay & time travel, and large logs, all with very high throughput (million messages per second).

Debezium Server is enhanced to support publishing captured change events to native RabbitMQ Streams leveraging RabbitMQ Stream Java Client.

Milvus

Milvus is an open-source vector database designed for similarity search and the retrieval of high-dimensional data, such as embeddings from machine learning models (for example. text, images, and audio). You can use Milvus to process vector datatypes that are captured from a source database, or use it with a transformation to calculate vectors from message fields and then use them as embeddings.

Milvus sink ingests incoming messages and upserts the after part of each message into a collection. The collection cannot contain dots, so the sink replaces all dots with underscore characters. When a delete message arrives, the matching record is removed from the collection.

Injection points

You can modify the behavior of the Milvus sink connector by applying custom logic that provides alternative implementations for specific functions. If the alternative implementations are not available, the connector uses the default implementations.

Qdrant Sink

Qdrant is an open-source vector database optimized for vector similarity search and extended with powerful filtering capabilities. It’s designed for high-load applications, and it enables you to efficiently store, manage, and search embedding vectors. You can use Qdrant to process vector datatypes that are captured directly from a source database, or you can use a transformation to calculate embeddings from message fields, and then send those embeddings to the database for processing.

Qdrant sink ingests incoming messages and upserts the after part of each message into a collection. When a delete message arrives, the matching record is removed from the collection.

The sink behaves according to the following rules:

• Each Debezium collection or table is mapped to a Qdrant collection.
• Primary keys are required and are used as Qdrant point IDs (only INT64 and UUID are supported).
• FloatVector and DoubleVector data can serve as sources for Qdrant vectors.
• Non-primary key and non-vector fields are mapped to Qdrant payloads.

Injection points

To modify connector behavior, you can apply custom logic to specify alternative implementations for certain functions. If you specify implementations that are not available, the connector uses the default implementations.

InstructLab

InstructLab is a community-driven project for augmenting large language models (LLMs) for use in generative artificial intelligence applications. Working with InstructLab, users who identify gaps in the capabilities of a base model can collaboratively develop a taxonomy to augment the model, with each contributor providing specific expertise and skills.

Debezium Server enables you to automate the process of adding skills and knowledge into a taxonomy by configuring a data sink based on InstructLab question and answer (qna.yml) files. The sink configuration defines a series of mappings that are used to derive question, answer, and context values from an event stream. These mappings can source values directly from fields within the event payload, headers, or statically configured constants. Periodically, users can then use InstructLab to train the model to take advantage of the new skills and knowledge added to the taxonomies.

Mapping definition

In the InstructLab sink configuration, you can set properties to specify how Debezium Server maps fields in event messages to question, answer, and contextattributes in an InstructLab qna.yml file.

For each attribute type that you want to populate in the qna.yml file, you specify a mapping prefix that determines the message field from which Debezium extracts the value. You can specify the following prefix values:

Value

If you prefix a mapping definition with the string value:, Debezium extracts the value of the specified field from the incoming event payload. For example, to populate an InstructLab question attribute with the value of the abc field in the event payload, set the debezium.sink.instructlab.taxonomy.<name>.question property to value:abc. When Debezium processes the message it takes the value of the payload field abc and adds it as a question attribute in the qna.yml file.

For events that have a Debezium-structured payload, Debezium extracts the specified field from the after section of the payload. If the event is flattened, the field is sourced directly from the event’s value.

Header

If you prefix a mapping definition with the string header:, Debezium extracts the value of the specified header field of the incoming event message. For example, if you specify the mapping header:h1, when Debezium detects a header with the name h1 in a source message, it extracts the value of the h1 field.

Constant

If the mapping definition does not include the header: or value: prefixes, when Debezium detects instances of the specified value in incoming messages, it treats them as constants, and uses them as-is. Use this option when you want to map the value of a specific static constant to an attribute in the qna.yml file.

JDBC

The JDBC sink writes change events directly to a relational database using JDBC. It leverages the Debezium JDBC connector under the hood, supporting a wide variety of database dialects, including Db2, MySQL, Oracle, PostgreSQL, and SQL Server.

The sink supports idempotent writes by using upsert semantics, basic schema evolution to automatically create or alter destination tables, and delete propagation.

Example JDBC sink configuration

debezium.sink.type=jdbc
debezium.sink.jdbc.connection.url=jdbc:mysql://localhost:3306/target_db
debezium.sink.jdbc.connection.username=root
debezium.sink.jdbc.connection.password=password
debezium.sink.jdbc.insert.mode=upsert
debezium.sink.jdbc.primary.key.mode=record_key
debezium.sink.jdbc.schema.evolution=basic
debezium.sink.jdbc.delete.enabled=true

Apache Iceberg

The Debezium Apache Iceberg connector is a sink that runs in Debezium Server. It consumes database change events from a source Debezium connector and writes them to Apache Iceberg tables. This allows for direct replication of a Database (Change Data Capture events) to Iceberg tables on cloud storage or HDFS, eliminating the need for intermediate systems like ETL Tool, Spark or other streaming platforms.

The Debezium Iceberg connector supports both upsert and append modes for data replication. When event and key schema information is enabled, the destination Iceberg tables are automatically created upon the initial startup.

The connector provides the following features:

• Upsert and Append modes.
• Optimizing batch size.
• Customizable table naming.
• Offset and Schema History storage in Iceberg.
• Automatic schema evolution.

• It is recommended to use ExtractNewRecordState transformation (event flattening). Key features like upsert mode and automatic primary key detection depend on the flattened event structure produced by this SMT. Without it, the connector can only operate in a more limited append-only mode.
• The Debezium Server sink for Apache Iceberg is a community-maintained, open-source project. Please note that it is maintained in a separate repository, which you can find here: Debezium Server Iceberg GitHub

Extensions

Debezium Server uses the Quarkus framework and relies on dependency injection to enable developer to extend its behaviour. Note that only the JVM mode of Quarkus is supported, but not native execution via GraalVM. The server can be extended in two ways by providing a custom logic:

• implementation of a new sink
• customization of an existing sink - i.e. non-standard configuration

Implementation of a new sink

The new sink can be implemented as a CDI bean implementing interface DebeziumEngine.ChangeConsumer and with annotation @Named and unique name and scope @Dependent. The name of the bean is used as the debezium.sink.type option.

The sink needs to read the configuration using Microprofile Config API. The execution path must pass the messages into the target system and regularly commit the passed/processed messages.

See the Kinesis sink implementation for further details.

Customization of an existing sink

Some of the sinks exposes dependency injections points that enable users to provide its own bean that would modify the behaviour of the sink. Typical examples are fine tuning of the target client setup, the destination naming etc.

See an example of a custom topic naming policy implementation for further details.