SupportConsoleDevelopersStart a trialContact

Chapter 3. Sink connectors

Debezium provides sink connectors that can consume events from sources such as Apache Kafka topics. A sink connector standardizes the format of the data, and then persists the event data to a configured sink repository. Other systems, applications, or users can then access the events from the data sink.

Because the sink connector applies a consistent structure to the event data that it consumes, downstream applications that read from the data sink can more easily interpret and process that data.

Debezium provides the following sink connectors:

3.1. Debezium sink connector for JDBC
Copy link

The Debezium JDBC connector is a Kafka Connect sink connector implementation that can consume events from multiple source topics, and then write those events to a relational database by using a JDBC driver. This connector supports a wide variety of database dialects, including Db2, MySQL, Oracle, PostgreSQL, and SQL Server.

3.1.1. How the Debezium JDBC connector works
Copy link

The Debezium JDBC connector is a Kafka Connect sink connector, and therefore requires the Kafka Connect runtime. The connector periodically polls the Kafka topics that it subscribes to, consumes events from those topics, and then writes the events to the configured relational database. The connector supports idempotent write operations by using upsert semantics and basic schema evolution.

The Debezium JDBC connector provides the following features:

By default, Debezium source connectors produce complex, hierarchical change events. When Debezium connectors are used with other JDBC sink connector implementations, you might need to apply the ExtractNewRecordState single message transformation (SMT) to flatten the payload of change events, so that they can be consumed by the sink implementation. If you run the Debezium JDBC sink connector, it’s not necessary to deploy the SMT, because the Debezium sink connector can consume native Debezium change events directly, without the use of a transformation.

When the JDBC sink connector consumes a complex change event from a Debezium source connector, it extracts the values from the after section of the original insert or update event. When a delete event is consumed by the sink connector, no part of the event’s payload is consulted.

Important

The Debezium JDBC sink connector is not designed to read from schema change topics. If your source connector is configured to capture schema changes, in the JDBC connector configuration, set the topics or topics.regex properties so that the connector does not consume from schema change topics.

The Debezium JDBC sink connector guarantees that events that it consumes from Kafka topics are processed at least once.

You can run the Debezium JDBC sink connector across multiple Kafka Connect tasks. To run the connector across multiple tasks, set the tasks.max configuration property to the number of tasks that you want the connector to use. The Kafka Connect runtime starts the specified number of tasks, and runs one instance of the connector per task. Multiple tasks can improve performance by reading and processing changes from multiple source topics in parallel.

To enable the Debezium JDBC sink connector to correctly map the data type from an inbound message field to an outbound message field, the connector requires information about the data type of each field that is present in the source event. The connector supports a wide range of column type mappings across different database dialects. To correctly convert the destination column type from the type metadata in an event field, the connector applies the data type mappings that are defined for the source database. You can enhance the way that the connector resolves data types for a column by setting the column.propagate.source.type or datatype.propagate.source.type options in the source connector configuration. When you enable these options, Debezium includes extra parameter metadata, which assists the JDBC sink connector in more accurately resolving the data type of destination columns.

For the Debezium JDBC sink connector to process events from a Kafka topic, the Kafka topic message key, when present, must be a primitive data type or a Struct. In addition, the payload of the source message must be a Struct that has either a flattened structure with no nested struct types, or a nested struct layout that conforms to Debezium’s complex, hierarchical structure.

If the structure of the events in the Kafka topic do not adhere to these rules, you must implement a custom single message transformation to convert the structure of the source events into a usable format.

By default, the Debezium JDBC sink connector does not transform any of the fields in the source event into the primary key for the event. Unfortunately, the lack of a stable primary key can complicate event processing, depending on your business requirements, or when the sink connector uses upsert semantics. To define a consistent primary key, you can configure the connector to use one of the primary key modes described in the following table:

ModeDescription

none

No primary key fields are specified when creating the table.

kafka

The primary key consists of the following three columns:

  • __connect_topic
  • __connect_partition
  • __connect_offset

The values for these columns are sourced from the coordinates of the Kafka event.

record_key

The primary key is composed of the Kafka event’s key.

If the primary key is a primitive type, specify the name of the column to be used by setting the primary.key.fields property. If the primary key is a struct type, the fields in the struct are mapped as columns of the primary key. You can use the primary.key.fields property to restrict the primary key to a subset of columns.

record_value

The primary key is composed of the Kafka event’s value.

Because the value of a Kafka event is always a Struct, by default, all of the fields in the value become columns of the primary key. To use a subset of fields in the primary key, set the primary.key.fields property to specify a comma-separated list of fields in the value from which you want to derive the primary key columns.

record_header

The primary key is composed of the Kafka event’s headers.

When primary.key.mode is set to record_header, the connector derives the primary key from the Kafka record headers. The connector creates a Struct from all headers present in the Kafka record, using each header’s key as the field name and the corresponding header value as the field value. Each header is included as a primary key column.

Note

When primary.key.mode is set to record_header, the primary.key.fields property has no effect. All record headers are included as primary key columns, with no filtering available. This limitation is tracked in dbz#2795.

Warning

Avoid using nested Struct values in record headers when primary.key.mode is set to record_header. The way that the connector converts nested Struct types varies and depends on both the header.converter setting at the source and on the database dialect of the sink database. The SimpleHeaderConverter converts nested header values to MAP types, while the JsonConverter maps them to Struct types. In either case, the resulting JSON, JSONB, HSTORE, or VARCHAR column types might not be valid primary keys in the target database. If the target database does not permit the converted type in primary key columns, the sink operation can fail. This limitation is tracked in dbz#2796.

Important

Some database dialects might throw an exception if you set the primary.key.mode to kafka and set schema.evolution to basic. This exception occurs when a dialect maps a STRING data type mapping to a variable length string data type such as TEXT or CLOB, and the dialect does not allow primary key columns to have unbounded lengths. To avoid this problem, apply the following settings in your environment:

  • Set schema.evolution to a value other than basic.
  • Create the database table and primary key mappings in advance.
Important

If a column maps to a data type that the target database does not permit, explicitly specify the permissible columns in the primary.key.fields list. Consult the documentation for your database vendor to determine the permissible data types.

By default, when the connector consumes DELETE or tombstone events, it does not delete the corresponding row in the sink database. You might want the connector to remove rows when you need to maintain data parity between the source and sink databases, so that deletions in the source are also reflected in the sink.

If you want to the connector to remove rows, you must explicitly set delete.enabled=true in the connector configuration. To use this mode you must also set primary.key.fields to a value other than none. The preceding configuration is necessary, because deletes are executed based on the primary key mapping, so if a destination table has no primary key mapping, the connector is unable to delete rows.

In an at-least-once delivery pipeline, the connector can receive the same event more than once after a restart or rebalance. Configure the connector to use upsert mode so that data written to the sink database is idempotent, and replaying an event always produces the same final row state.

To enable the connector to perform idempotent writes, you must be explicitly set the insert.mode for the connector to upsert. An upsert operation is applied as either an update or an insert, depending on whether the specified primary key already exists.

If the primary key value already exists, the operation updates values in the row. If the specified primary key value doesn’t exist, an insert adds a new row.

Each database dialect handles idempotent writes differently, because there is no SQL standard for upsert operations. The following table shows, for each database dialect that Debezium supports, the upsert DML syntax and the constraints that trigger an update action during an upsert operation.

DialectUpsert SyntaxConstraint that triggers the update action

Db2

MERGE …​

Configured primary key only

MariaDB

INSERT …​ ON DUPLICATE KEY UPDATE …​

Any unique index

MySQL

INSERT …​ ON DUPLICATE KEY UPDATE …​

Any unique index

Oracle

MERGE …​

Configured primary key only

PostgreSQL

INSERT …​ ON CONFLICT …​ DO UPDATE SET …​

Configured primary key only

SQL Server

MERGE …​

Configured primary key only

Warning

Secondary unique constraints can create ordering dependencies. For more information about their use, see secondary unique constraints.

Secondary unique constraints can create ordering dependencies between records that have different primary keys. Kafka guarantees record order only within a partition, and the connector batches and reduces events based only on the configured primary key. For these reasons, you should avoid defining secondary unique constraints or secondary unique indexes on destination tables. If a uniqueness rule is required, enforce the rule in the source database, and omit the constraint from the destination table.

If dependent records are routed to different partitions, the sink connector might apply records out of source order, which can lead to inconsistent results. The following table shows how different database dialects handle conflicts involving secondary unique constraints:

DialectBehavior

Dialects that use INSERT …​ ON DUPLICATE KEY UPDATE

Consider conflicts on any unique index, not only the configured primary key. A conflict involving a secondary unique constraint can therefore update an existing row whose primary key differs from the incoming record, causing silent data inconsistency.

Dialects that use INSERT …​ ON CONFLICT or MERGE

Consider conflicts only on the configured primary key. A conflict involving a secondary unique constraint causes the write operation to fail with a constraint violation error rather than updating a different row.

To prevent data inconsistencies when a destination table includes a secondary unique constraint, the data pipeline must meet all of the following requirements:

Dependent events routed through the same partition
Route all dependent events through the same Kafka partition, and preserve the source event order. For a table with dependencies between arbitrary rows, use a single partition for its Kafka topic. Setting tasks.max=1 is not sufficient to restore ordering between records that are already in different partitions.
Reduction buffering disabled
Retain the default value (false) of the use.reduction.buffer property. Reduction by primary key can remove intermediate row states that are required to satisfy secondary unique constraints.
Batch size set to 1

Set the value of the batch.size property to 1 if any batches might contain multiple events for a primary key that is not defined in the destination table.

Some dialects and driver configurations process a batch in a single pass, rather than validating each row against the effects of the preceding rows in the same batch. For example, MERGE-based dialects can attempt two INSERT operations for the same primary key instead of an INSERT followed by an UPDATE. Similarly, when dialect.postgres.unnest.insert.enabled is set to true, PostgreSQL combines a batch into a single multi-row statement. In each of these cases, the batch fails.

The schema.evolution property controls whether the connector automatically creates or alters destination tables when event schemas change. The following table describes the available modes, from no evolution (none) through startup validation (validate-only) to automatic table creation and column addition (basic).

ModeDescription

none

The connector does not perform any DDL schema evolution.

basic

The connector automatically detects fields that are in the event payload but that do not exist in the destination table. The connector alters the destination table to add the new fields.

When schema.evolution is set to basic, the connector automatically creates or alters the destination database table according to the structure of the incoming event.

When an event is received from a topic for the first time, and the destination table does not yet exist, the Debezium JDBC sink connector uses the event’s key, or the schema structure of the record to resolve the column structure of the table. If schema evolution is enabled, the connector prepares and executes a CREATE TABLE SQL statement before it applies the DML event to the destination table.

When the Debezium JDBC connector receives an event from a topic, if the schema structure of the record differs from the schema structure of the destination table, the connector uses either the event’s key or its schema structure to identify which columns are new, and must be added to the database table. If schema evolution is enabled, the connector prepares and executes an ALTER TABLE SQL statement before it applies the DML event to the destination table. Because changing column data types, dropping columns, and adjusting primary keys can be considered dangerous operations, the connector is prohibited from performing these operations.

The schema of each field determines whether a column is NULL or NOT NULL. The schema also defines the default values for each column. If the connector attempts to create a table with a nullability setting or a default value that don’t want, you must either create the table manually, ahead of time, or adjust the schema of the associated field before the sink connector processes the event. To adjust nullability settings or default values, you can introduce a custom single message transformation that applies changes in the pipeline, or modifies the column state defined in the source database.

A field’s data type is resolved based on a predefined set of mappings. For more information, see JDBC field types.

Important

When you introduce new fields to the event structure of tables that already exist in the destination database, you must define the new fields as optional, or the fields must have a default value specified in the database schema. If you want a field to be removed from the destination table, use one of the following options:

  • Remove the field manually.
  • Drop the column.
  • Assign a default value to the field.
  • Define the field a nullable.

The Debezium JDBC sink connector consumes Kafka messages by constructing either DDL (schema changes) or DML (data changes) SQL statements that are executed on the destination database. By default, the connector uses the names of the source topic and the event fields as the basis for the table and column names in the destination table. The constructed SQL does not automatically delimit identifiers with quotes to preserve the case of the original strings. As a result, by default, the text case of table or column names in the destination database depends entirely on how the database handles name strings when the case is not specified.

For example, if the destination database dialect is Oracle and the event’s topic is orders, the destination table will be created as ORDERS because Oracle defaults to uppercase names when the name is not quoted. Similarly, if the destination database dialect is PostgreSQL and the event’s topic is ORDERS, the destination table will be created as orders because PostgreSQL defaults to lower-case names when the name is not quoted.

To explicitly preserve the case of the table and field names that are present in a Kafka event, in the connector configuration, set the value of the quote.identifiers property to true. When this options is set, when an incoming event is for a topic called orders, and the destination database dialect is Oracle, the connector creates a table with the name orders, because the constructed SQL defines the name of the table as "orders". Enabling quoting results in the same behavior when the connector creates column names.

Connection Idle Timeouts

Connections held idle in the pool can time out if they remain inactive beyond the database’s configured threshold, causing subsequent write operations to fail. Set the hibernate.agroal.idleValidation property to enable periodic validation of pooled connections and prevent idle timeout errors.

The JDBC sink connector for Debezium leverages a connection pool to enhance performance. Connection pools are engineered to establish an initial set of connections, maintain a specified number of connections, and efficiently allocate connections to the application as required. However, a challenge arises when connections linger idle in the pool, potentially triggering timeouts if they remain inactive beyond the configured idle timeout threshold of the database.

To mitigate the potential for idle connection threads to trigger timeouts, connection pools offer a mechanism that periodically validates the activity of each connection. This validation ensures that connections remain active, and prevents the database from flagging them as idle. In the event of a network disruption, if Debezium attempts to use a terminated connection, the connector prompts the pool to generate a new connection.

By default, the Debezium JDBC sink connector does not conduct idle timeout tests. However, you can configure the connector to request the pool to perform timeout tests at a specified interval by setting the hibernate.agroal.idleValidation property. For example:

Example timeout configuration

{
  "hibernate.agroal.idleValidation": "300"
}

The Debezium JDBC sink connector uses the Hibernate Agroal connection pool. You can customize the Agroal connection pool by setting properties in the hibernate.agroal.* configuration namespace. In the preceding example, the setting of the hibernate.agroal.idleValidation property configures the connection pool to perform idle timeout tests every 300 seconds. After you apply the configuration, the connection pool begins to assess unused connections every five minutes.

Changing how table names are resolved

The Debezium JDBC sink connector uses a io.debezium.sink.naming.CollectionNamingStrategy to control how an event is mapped to a target table. The default behavior is to use the collection.name.format configuration property and apply the following rules:

  • Replace all occurrences of . with _ in the event’s topic name.
  • Replace any reference to ${topic} in the collection.name.format with the sanitized topic name.
  • Replace any reference to ${source.<fieldName>} in the collection.name.format with the appropriate field value from the source block.

While this behavior covers many use cases, this may not always be ideal in all situations. In some cases, you may wish to directly map the event’s topic name to the table name, without any sanitization. This can be accomplished by setting the collection.naming.strategy to use a passthrough implementation. For example:

Example using the passthrough collection naming strategy

{
  "collection.naming.strategy": "io.debezium.sink.naming.PassthroughCollectionNamingStrategy"
}

The passthrough naming strategy applies the following simplified rules:

  • Replace any reference to ${topic} in the collection.name.format with the event’s topic name, without sanitization.
  • Ignore any reference to ${source.<fieldName>} in the collection.name.format.

If neither of these solutions provide the behavior needed, you can also implement your own io.debezium.sink.naming.CollectionNamingStrategy. The custom implementation should be placed alongside the Debezium JDBC sink connector files. Be sure to set the collection.naming.strategy to the fully qualified class name of the custom implementation.

Before the Debezium JDBC sink connector sends data to a sink database, it converts data types in the original source record to a corresponding type in the target system. Appropriate data type mapping ensures that the original data is accurately represented in the destination database.

The Debezium JDBC sink connector resolves a column’s data type by using a logical or primitive type-mapping system. Primitive types include values such as integers, floating points, Booleans, strings, and bytes. Typically, Kafka messages use a specific Kafka Connect Schema type code to represent primitive data types.

By contrast, to represent more complex data, Debezium uses logical schema names, which provide logical groupings of named fields that can represent a range of data types (strings, arrays, JSON, XML, and so forth). These structured types take into account the semantic meaning of the data and provide a more opinionated interpretation about how to serialize the underlying primitive types. Logical types are useful in representing values that have a specific encoding, such as numbers that represent the time since the epoch.

The following examples show representative structures of primitive and logical data types:

Example 3.1. Primitive field schema

{
  "schema": {
    "type": "INT64"
  }
}

Example 3.2. Logical field schema

{
  "schema": {
    "type": "INT64",
    "name": "org.apache.kafka.connect.data.Date"
  }
}

Kafka Connect is not the only source for these complex, logical types. In fact, when a Debezium source connector emits a change event, it can assign similar logical types to event fields to represent such data types as timestamps, dates, and even JSON data.

The Debezium JDBC sink connector uses these primitive and logical types to resolve a column’s type to a JDBC SQL code, which represents a column’s type. These JDBC SQL codes are then used by the underlying Hibernate persistence framework to resolve the column’s type to a logical data type for the dialect in use. The following tables illustrate the primitive and logical mappings between Kafka Connect and JDBC SQL types, and between Debezium and JDBC SQL types. The actual final column type varies with for each database type.

  1. Section 3.1.2.1, “Mappings between Kafka Connect primitive data types and sink database column types”
  2. Section 3.1.2.2, “Mappings between Kafka Connect logical data types and sink database column types”
  3. Section 3.1.2.3, “Mappings between Debezium logical data types and JDBC column types”
  4. Section 3.1.2.4, “Mappings between Debezium dialect-specific logical data types and RDBMS column data types”
  5. Section 3.1.2.5, “Mappings between Debezium vector data types and RDBMS column types”

The following table maps each Kafka Connect primitive type to its corresponding JDBC SQL type, which the Debezium JDBC sink connector uses when creating or altering columns in the destination database.

Primitive TypeJDBC SQL Type

INT8

Types.TINYINT

INT16

Types.SMALLINT

INT32

Types.INTEGER

INT64

Types.BIGINT

FLOAT32

Types.FLOAT

FLOAT64

Types.DOUBLE

BOOLEAN

Types.BOOLEAN

STRING

Types.CHAR, Types.NCHAR, Types.VARCHAR, Types.NVARCHAR

BYTES

Types.VARBINARY

Note

The connector does not support BOOLEAN data in Oracle 23. If you configure the JDBC connector to use an Oracle 23 database as the sink target, you cannot rely on schema evolution to automatically create a field that uses the Oracle 23 BOOLEAN data type. The connector maps BOOLEAN data to BIT data types, which are universal across Oracle versions. As a workaround, if you need a BOOLEAN data type for an Oracle 23 sink, add the field manually to the target table.

The following table maps each Kafka Connect logical type to its corresponding JDBC SQL type, which the Debezium JDBC sink connector uses when writing structured data such as dates, times, and decimals to the destination database.

Logical TypeJDBC SQL Type

org.apache.kafka.connect.data.Decimal

Types.DECIMAL

org.apache.kafka.connect.data.Date

Types.DATE

org.apache.kafka.connect.data.Time

Types.TIMESTAMP

org.apache.kafka.connect.data.Timestamp

Types.TIMESTAMP

The following table maps Debezium logical types to their corresponding JDBC SQL types, which the connector uses when writing Debezium change event fields — such as timestamps, durations, and variable-scale decimals — to the destination database.

Logical TypeJDBC SQL Type

io.debezium.time.Date

Types.DATE

io.debezium.time.Time

Types.TIMESTAMP

io.debezium.time.MicroTime

Types.TIMESTAMP

io.debezium.time.NanoTime

Types.TIMESTAMP

io.debezium.time.ZonedTime

Types.TIME_WITH_TIMEZONE

io.debezium.time.Timestamp

Types.TIMESTAMP

io.debezium.time.MicroTimestamp

Types.TIMESTAMP

io.debezium.time.NanoTimestamp

Types.TIMESTAMP

io.debezium.time.ZonedTimestamp

Types.TIMESTAMP_WITH_TIMEZONE

io.debezium.data.VariableScaleDecimal

Types.DOUBLE

Important

If the database does not support time or timestamps with time zones, the mapping resolves to its equivalent without timezones.

The following table maps Debezium logical types to dialect-specific column types for each supported database, which you can use to understand how the connector resolves fields such as JSON, geospatial data, and bit strings when targeting a specific database dialect.

Table 3.1. Mappings between Debezium dialect-specific logical types and column data types
Logical typeDb2 typeMySQL typePostgreSQL typeOracle typeSQL Server type

io.debezium.data.Bits

blob

bit(n)

bit varyingbit(n)1

blob

varbinary(n)

io.debezium.data.Enum

n/a

enum

text

n/a

n/a

io.debezium.data.Json

clob

json

json

json

n/a

io.debezium.data.EnumSet

n/a

set

n/a

n/a

n/a

io.debezium.time.Year

integer

year(n)

int4

number

int

io.debezium.time.MicroDuration

bigint

interval

int8

number

bigint

io.debezium.data.Ltree

clobvarchar(n)1

longtextvarchar1

ltree

clobvarchar2(n)1

varchar

io.debezium.data.Uuid

clobvarchar(n)1

uuid

textvarchar(n)1

clobvarchar2(n)1

varchar

io.debezium.data.Tsvector

clobvarchar(n)1

longtextvarchar1

tsvector

clobvarchar2(n)1

varchar

io.debezium.data.Xml

clob

longtext

xml

xmltype

xml

io.debezium.data.Geography

n/a

geometry

postgis.geography

sdo_geometry

geometry

io.debezium.data.Geometry

n/a

geometry

postgis.geometry

sdo_geometry

geometry

1 When a logical type maps to more than one type, the Debezium JDBC connector defaults to the first more generalized data type. If the connector consumes change events from topics that are populated by a Debezium source connector, the JDBC connector can use stricter, more concrete data types when it sends data to some targets. To enable the use of these stricter data type conversions, set the column.propagate.source.type or datatype.propagate.source.type properties in the source connector configuration. When these properties are set in the source connector, the events that it generates include additional column metadata that the JDBC connector then uses to construct columns with a stricter concrete data type. In the preceding table, where a second entry, marked with an asterisk, appears in a table cell, it represents the stricter data type.

The Debezium JDBC sink connector supports direct mapping of logical vector data types from source events to sink destinations, provided that the target database supports a comparable representation.

If a vector field in the source has a logical name, Debezium uses it to determine the appropriate mapping. For certain databases, the connector recognizes the following logical names under the io.debezium.data.vector.* namespace:

  • FloatVector
  • DoubleVector
  • SparseVector

When the connector processes a message, if these logical types are present, the connector checks for a corresponding mapping in the target database. If a mapping exists, it applies it; otherwise, it defaults to serializing the data as a string, based on the Kafka Connect schema type.

For example, a connector that is configured to send data to a PostgreSQL database sink maps fields with the io.debezium.data.FloatVector logical name to the halfvector column type, using the special override mapping. By contrast, if no direct mapping is available in the sink database, as for example with an Oracle sink, the connector defaults to serializing vector data in its raw string format.

Support for the special vector types is not available for all releases of the supported databases. Sink databases must meet the following requirements to support direct mapping of logical vector types:

  • MariaDB version 11.7 or later
  • MySQL version 9.0 or later
  • PostgreSQL requires the pgvector extension

Earlier versions of MariaDB or MySQL, and PostgreSQL without pgvector, do not support special vector types.

The following table maps Debezium logical vector types to their corresponding column data types for each supported sink database.

Logical TypeDb2MySQLPostgreSQLOracleSQL Server

io.debezium.data.DoubleVector

Unsupported

'vector'

vector

Unsupported

Unsupported

io.debezium.data.FloatVector

Unsupported

'vector'

halfvec

Unsupported

Unsupported

io.debezium.data.SparseVector

Unsupported

Unsupported

sparsevec

Unsupported

Unsupported

Note

If there is not a direct mapping between one of the vector logical types and your target relational database’s column types, you can use the VectorToJsonConverter to convert the vector logical type to JSON so that it can be written to any target relational database.

Column and data type propagation

In addition to the primitive and logical mappings shown in the preceding tables, if the source of the change events is a Debezium source connector, the resolution of the column type, along with its length, precision, and scale, can be further influenced by enabling column or data type propagation. Column and data type propagation helps to determine how the structure and data types of incoming data are translated and applied to the sink destination. As discussed earlier in Data and column type mappings, you can enforce propagation by setting one of the following properties in the source connector configuration:

  • column.propagate.source.type
  • datatype.propagate.source.type

The Debezium JDBC sink connector applies only the values with the higher precedence.

To illustrate how propagation affects how the connector maps data types, let’s look at an example. The following example shows a field schema that might be included within a change event:

Example 3.3. Debezium change event field schema with column or data type propagation enabled

{
  "schema": {
    "type": "INT8",
    "parameters": {
      "__debezium.source.column.type": "TINYINT",
      "__debezium.source.column.length": "1"
    }
  }
}

Because the source connector that emitted the event was configured to use column or data type propagation, the event includes parameters that specify column type and length.

If propagation were not enabled for the source connector, the type and length parameters would be absent, and the Debezium JDBC sink connector would default to mapping the INT8 value in the type field to a column type of Types.SMALLINT. Depending on the SQL dialect of the target database, the connector would then resolve the JDBC Types.SMALLINT type to any of several logical types, For example, for a MySQL sink target, the JDBC connector defaults to converting Types.SMALLINT to a TINYINT column type with no specified length.

However, because type propagation is enabled in the source connector, the event that it emitted includes the type and length, providing more specific mapping instructions. Thus, rather than using the default mapping, the Debezium JDBC sink connector uses the given parameter values to refine the mapping, and creates a column in the sink database with the type TINYINT(1).

Note

Typically, the effect of using column or data type propagation is most significant when the source and sink databases are of the same type. Because the source and sink database share the same underlying schema structure, it’s easier to map data types between them without having to apply complex transformations or interpretations.

The Debezium JDBC connector provides several transformations that can be added to the connector configuration to modify the consumed events in-flight before they’re processed by the connector.

3.1.3.1. Transformation that alter topic names
Copy link

The Debezium JDBC connector provides two naming transformations that you can apply to change the naming styles of topic or field names within an event. Optionally, you can also use the naming transformations to apply a prefix or suffix to the transformed name.

You can configure naming transformations to use one of the following naming styles:

camel_case
Removes all period (.) and underscore (_) characters, and converts the immediate next character to uppercase. For example, the value inventory.customers is changed to inventoryCustomers.
snake_case
Removes all period (.) characters and replaces them with underscore (_) characters. Additionally, all numeric sequences are prefixed with an underscore (_), and all uppercase characters are converted to lowercase and then prefixed with an underscore (_) character. For example, the value public.inventory becomes public_inventory, while TopicWith123Numbers becomes topic_with_123_numbers.
upper_case
Converts to uppercase. For example, the value public.inventory would become PUBLIC.INVENTORY.
lower_case
Converts to lowercase. For example, the value PUBLIC.INVENTORY would become public.inventory.

The connector provides the following naming transformations:

CollectionNameTransformation

The CollectionNameTransformation provides a way to change the case of the topic names before the event is consumed. This transformation has the following configuration properties:

PropertyDefaultDescription

collection.naming.style

none

Specifies the naming style to apply to the event’s topic.

collection.naming.prefix

empty

Specifies the prefix that is applied to the topic name after transformation.

collection.naming.suffix

empty

Specifies the suffix that is applied to the topic name after transformation.

The following configuration example illustrates using the CollectionNameTransformation SMT to set the topic name to uppercase and prefix the topic name with ADT_. For example, given a topic called public.inventory, the topic name becomes ADT_PUBLIC.INVENTORY.

Example CollectionNameTransformation configuration

{
  "transforms": "topic-uppercase",
  "transforms.topic-uppercase.type": "io.debezium.connector.jdbc.transforms.CollectionNameTransformation",
  "transforms.topic-uppercase.collection.naming.style": "upper_case",
  "transforms.topic-uppercase.collection.naming.prefix": "ADT_"
}

FieldNameTransformation

The FieldNameTransformation provides a way to change the case of field names before the event is consumed. If the event is a Debezium source connector event, only the fields within the before and after sections are changed. When the event is not a Debezium source connector event, all top-level field names are changed.

This transformation has the following configuration properties:

PropertyDefaultDescription

column.naming.style

none

Specifies the naming style to apply to the field name.

column.naming.prefix

empty

Specifies the prefix that is applied to the field name after transformation.

column.naming.suffix

empty

Specifies the suffix that is applied to the field name after transformation.

The following example shows how to configure the FieldNameTransformation SMT to convert field names name to lowercase and prefix the names with the string adt_. After you apply the SMT to an event that includes a field with the name ID, the field is renamed to adt_id.

Example FieldNameTransformation configuration

{
  "transforms": "topic-lowercase",
  "transforms.topic-lowercase.type": "io.debezium.connector.jdbc.transforms.FieldNameTransformation",
  "transforms.topic-lowercase.column.naming.style": "lower_case",
  "transforms.topic-lowercase.column.naming.prefix": "adt_"
}

3.1.4. Deployment of Debezium JDBC connectors
Copy link

To deploy a Debezium JDBC connector, use Streams for Apache Kafka to automatically create a Kafka Connect image that includes the connector plug-in.

Important

Due to licensing requirements, the Debezium JDBC connector archive does not include the drivers that Debezium requires to connect to the Db2 and Oracle databases. To enable the connector to access these databases, you must add the drivers to your connector environment. For information about how to obtain drivers that are not supplied with the connector, see Obtaining drivers not included in the connector archive.

Due to licensing requirements, the JDBC driver files that Debezium requires to connect to Db2 Database and Oracle Database are not included in the Debezium JDBC connector archive. These drivers are available for download from Maven Central. To add the connector to your Kafka Connect image, add the Maven Central location for the driver to builds.plugins.artifact.url in the KafkaConnect custom resource as shown in Section 3.1.4.3, “Using Streams for Apache Kafka to deploy a Debezium JDBC connector”.

The preferred method for deploying a Debezium connector is to use Streams for Apache Kafka to build a Kafka Connect container image that includes the connector plug-in.

During the deployment process, you create and use the following custom resources (CRs):

  • A KafkaConnect CR that defines your Kafka Connect instance and includes information about the connector artifacts needs to include in the image.
  • A KafkaConnector CR that provides details that include information the connector uses to access the source database. After Streams for Apache Kafka starts the Kafka Connect pod, you start the connector by applying the KafkaConnector CR.

In the build specification for the Kafka Connect image, you can specify the connectors that are available to deploy. For each connector plug-in, you can also specify other components that you want to make available for deployment. For example, you can add Apicurio Registry artifacts, or the Debezium scripting component. When Streams for Apache Kafka builds the Kafka Connect image, it downloads the specified artifacts, and incorporates them into the image.

The spec.build.output parameter in the KafkaConnect CR specifies where to store the resulting Kafka Connect container image. Container images can be stored in a container registry, such as quay.io, or in an OpenShift ImageStream. To store images in an ImageStream, you must create the ImageStream before you deploy Kafka Connect. ImageStreams are not created automatically.

Note

If you use a KafkaConnect resource to create a cluster, afterwards you cannot use the Kafka Connect REST API to create or update connectors. You can still use the REST API to retrieve information.

Additional resources

You can use the build configuration in Streams for Apache Kafka to automatically build a Kafka Connect container image to OpenShift. The build image includes the Debezium connector plug-ins that you specify.

During the build process, the Streams for Apache Kafka Operator transforms input parameters in a KafkaConnect custom resource, including Debezium connector definitions, into a Kafka Connect container image. The build downloads the necessary artifacts from the Red Hat Maven repository or from another configured HTTP server.

The newly created container is pushed to the container registry that is specified in .spec.build.output, and is used to deploy a Kafka Connect cluster. After Streams for Apache Kafka builds the Kafka Connect image, you create KafkaConnector custom resources to start the connectors that are included in the build.

Prerequisites

  • You have access to an OpenShift cluster in which the cluster Operator is installed.
  • The Streams for Apache Kafka Operator is running.
  • An Apache Kafka cluster is deployed as documented in Deploying and Managing Streams for Apache Kafka on OpenShift.
  • Kafka Connect is deployed on Streams for Apache Kafka
  • You have a Kafka topic from which the connector can read change event records.
  • A destination database is installed and is configured to accept JDBC connections.
  • You have a Red Hat build of Debezium license.
  • The OpenShift oc CLI client is installed or you have access to the OpenShift Container Platform web console.

  • Depending on how you intend to store the Kafka Connect build image, you need registry permissions or you must create an ImageStream resource:

    To store the build image in an image registry, such as Red Hat Quay.io or Docker Hub
    • An account and permissions to create and manage images in the registry.
    To store the build image as a native OpenShift ImageStream

Procedure

  1. Log in to the OpenShift cluster.

  2. Create a Debezium KafkaConnect custom resource (CR) for the connector, or modify an existing one.

    For example, create a KafkaConnect CR with the name dbz-jdbc-connect.yaml that specifies the annotations and image properties, as shown in the following excerpt. In the example that follows, the Kafka Connect custom resource is configured to download the following artifacts:

    • The Debezium JDBC connector archive.

    • A JDBC driver that is required to connect to an Oracle or Db2 sink database. You can omit this entry for other sink destinations. 

      apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: debezium-kafka-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 4.2
  replicas: 1
  bootstrapServers: debezium-kafka-cluster-kafka-bootstrap:9093
  build:
    output:
      type: imagestream
      image: debezium-streams-connect:latest
    plugins:
      - name: debezium-jdbc-connector
        artifacts:
          - type: zip
            url: https://maven.repository.redhat.com/ga/io/debezium/debezium-connector-jdbc/3.6.3.Final-redhat-00008/debezium-connector-jdbc-3.6.3.Final-redhat-00008-plugin.zip
          - type: jar
            url: https://repo1.maven.org/maven2/com/oracle/ojdbc/ojdbc11/23.26.1.0.0/ojdbc11-23.26.1.0.0.jar
  ...

      The following list describes select fields in the preceding KafkaConnect custom resource example:

      metadata.annotations (strimzi.io/use-connector-resources)
      Sets the strimzi.io/use-connector-resources annotation to true to enable the Cluster Operator to use KafkaConnector resources to configure connectors in this Kafka Connect cluster.
      spec.build
      Specifies where to store the build image and lists the plug-ins to include in the image, along with the location of the plug-in artifacts.
      build.output
      Specifies the registry in which the newly built image is stored.
      output.type
      Specifies the name and image name for the image output. Valid values are docker to push into a container registry such as Quay, or imagestream to push the image to an internal OpenShift ImageStream. To use an ImageStream, an ImageStream resource must be deployed to the cluster. For more information about specifying the build.output in the KafkaConnect configuration, see the Build schema reference in the Streams for Apache Kafka API Reference.
      spec.plugins
      Lists all of the connectors that you want to include in the Kafka Connect image. For each entry in the list, specify a plug-in name, and information about the artifacts that are required to build the connector. Optionally, for each connector plug-in, you can include other components that you want to be available for use with the connector. For example, you can add Service Registry artifacts, or the Debezium scripting component.
      artifacts.type
      Specifies the file type of the artifact specified in the artifacts.url. Valid types are zip, tgz, or jar. Debezium provides connector archives in .zip file format. JDBC driver files are in .jar format. The type value of the artifact must match the extension of the file that is referenced in the url field.
      artifacts.url (connector archive)
      Specifies the address of an HTTP server, such as a Maven repository, that stores the file for the connector artifact. The OpenShift cluster must have access to the specified server.
      artifacts.url (JDBC driver)

      (For Db2 or Oracle sinks only) Specifies the location of the JDBC JDBC driver in Maven Central. The drivers that are required for Debezium to connect to these databases are not included in the Debezium connector archives.

      The example provides the Maven URL for the Oracle Database JDBC driver. The Db2 JDBC driver is available at the following Maven location: https://repo1.maven.org/maven2/com/ibm/db2/jcc/11.5.9.0/jcc-11.5.9.0.jar

  3. Apply the KafkaConnect build specification to the OpenShift cluster by entering the following command: 

    oc create -f dbz-jdbc-connect.yaml

    Based on the configuration specified in the custom resource, the Streams Operator prepares a Kafka Connect image to deploy.

    After the build completes, the Operator pushes the image to the specified registry or ImageStream, and starts the Kafka Connect cluster. The connector artifacts that you listed in the configuration are available in the cluster.

  4. Create a KafkaConnector resource to define an instance of each connector that you want to deploy.

    For example, create the following KafkaConnector CR, and save it as orders-to-postgresql-jdbc-connector.yaml 

    apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: orders-topic-to-postgresql-via-jdbc-sink-connector
  labels: strimzi.io/cluster: debezium-kafka-connect-cluster
spec:
  class: io.debezium.connector.jdbc.JdbcSinkConnector
  tasksMax: 1
   config:
    connection.url: jdbc://postgresql://<host>:<port>/<database-name>
    connection.username: <database-user>
    connection.password: <database-pwd>
    insert.mode: upsert
    delete.enabled: true
    topics: orders
    primary.key.mode: record_key
    schema.evolution: basic
    use.time.zone: UTC

    The following list describes select fields in the preceding KafkaConnector configuration example:

    metadata.name
    The name of the Kafka connector instance that Streams for Apache Kafka creates in Kafka Connect.
    metadata.labels: strimzi.io/cluster
    Identifies the KafkaConnect cluster resource that hosts the connector. Streams for Apache Kafka uses this label to associate the KafkaConnector with a specific KafkaConnect deployment.
    spec.class
    The name of the connector class.
    spec.tasksMax
    The number of tasks that can operate concurrently.
    spec.config
    The connector’s configuration.
    connection.url
    The JDBC connection URL for the sink database. The URL specifies the port number and any authentication properties that are required to connect to the database. For example, jdbc:oracle:thin:@myhost.example.com:1521/myservice
    connection.username
    The name of the account that Debezium uses to connect to the database.
    connection.password
    The password that Debezium uses to connect to the database user account.
    topics
    Specifies a comma-separated list of Kafka topics that the connector reads. Events from each topic are streamed to tables with the same name in the sink database.

  5. Create the connector resource by running the following command: 

    oc create -n <namespace> -f <kafkaConnector>.yaml

    For example, 

    oc create -n debezium -f jdbc-inventory-connector.yaml

    The connector is registered to the Kafka Connect cluster and starts to run against the database that is specified by spec.config.database.dbname in the KafkaConnector CR. After the connector pod is ready, Debezium is running.

The Debezium JDBC sink connector has several configuration properties that you can use to achieve the connector behavior that meets your needs. Many properties have default values. Information about the properties is organized as follows:

3.1.5.1. JDBC connector Kafka consumer properties
Copy link

The following table describes the Kafka consumer properties that control how the Debezium JDBC sink connector identifies itself to Kafka Connect and specifies the topics it consumes.

PropertyDefaultDescription

name

No default

Unique name for the connector. A failure results if you attempt to reuse this name when registering a connector. This property is required by all Kafka Connect connectors.

class

No default

The name of the Java class for the connector. For the Debezium JDBC connector, specify the value io.debezium.connector.jdbc.JdbcSinkConnector.

tasks.max

1

Maximum number of tasks to use for this connector.

topics

No default

List of topics to consume, separated by commas. Do not use this property in combination with the topics.regex property.

topics.regex

No default

A regular expression that specifies the topics to consume. Internally, the regular expression is compiled to a java.util.regex.Pattern. Do not use this property in combination with the topics property.

3.1.5.2. JDBC connector connection properties
Copy link

The following table describes the properties that control how the Debezium JDBC sink connector connects to and authenticates with the destination database, including connection pool settings.

PropertyDefaultDescription

connection.provider

org.hibernate.agroal.internal.AgroalConnectionProvider

The connection provider implementation to use.

connection.url

No default

The JDBC connection URL used to connect to the database.

connection.username

No default

The name of the database user account that the connector uses to connect to the database.

connection.password

No default

The password that the connector uses to connect to the database.

connection.pool.min_size

5

Specifies the minimum number of connections in the pool.

connection.pool.max_size

32

Specifies the maximum number of concurrent connections that the pool maintains.

connection.pool.timeout

1800

Specifies the number of seconds that an unused connection is kept before it is discarded.

connection.restart.on.errors

false

Specifies whether the connector retries after a transient JDBC connection error.

When enabled (true), the connector treats connection issues (such as socket closures or timeouts) as retriable, allowing it to retry processing instead of failing the task. This reduces downtime and improves resilience against temporary disruptions.

Note

Setting this option to true can reduce downtime. However, in master-replica environments with asynchronous replication, it may lead to data loss if retries occur before all changes are fully replicated.

Use with caution where strong data consistency is required.

3.1.5.3. JDBC connector runtime properties
Copy link

The following table describes the runtime properties that control how the Debezium JDBC sink connector processes events and writes data to the destination database, including insert mode, primary key resolution, schema evolution, and batching behavior.

PropertyDefaultDescription

use.time.zone

UTC

Specifies the timezone used when inserting JDBC temporal values.

delete.enabled

false

Specifies whether the connector processes DELETE or tombstone events and removes the corresponding row from the database. Use of this option requires that you set the primary.key.mode to record.key.

truncate.enabled

false

Specifies whether the connector processes TRUNCATE events and truncates the corresponding tables from the database.

Note

Although support for TRUNCATE statements has been available in Db2 since version 9.7, currently, the JDBC connector is unable to process standard TRUNCATE events that the Db2 connector emits.

To ensure that the JDBC connector can process TRUNCATE events received from Db2, perform the truncation by using an alternative to the standard TRUNCATE TABLE statement. For example:

ALTER TABLE <table_name> ACTIVATE NOT LOGGED INITIALLY WITH EMPTY TABLE

The user account that submits the preceding query requires ALTER privileges on the table to be truncated.

insert.mode

insert

Specifies the strategy used to insert events into the database. The following options are available:

insert
Specifies that all events should construct INSERT-based SQL statements. Use this option only when no primary key is used, or when you can be certain that no updates can occur to rows with existing primary key values.
update
Specifies that all events should construct UPDATE-based SQL statements. Use this option only when you can be certain that the connector receives only events that apply to existing rows.
upsert
Specifies that the connector adds events to the table using upsert semantics. That is, if the primary key does not exist, the connector performs an INSERT operation, and if the key does exist, the connector performs an UPDATE operation. When idempotent writes are required, the connector should be configured to use this option.

primary.key.mode

none

Specifies how the connector resolves the primary key columns from the event.

none
Specifies that no primary key columns are created.
kafka

Specifies that the connector uses Kafka coordinates as the primary key columns. The key coordinates are defined from the topic name, partition, and offset of the event, and are mapped to columns with the following names:

  • __connect_topic
  • __connect_partition
  • __connect_offset
record_key
Specifies that the primary key columns are sourced from the event’s record key. If the record key is a primitive type, the primary.key.fields property is required to specify the name of the primary key column. If the record key is a struct type, the primary.key.fields property is optional, and can be used to specify a subset of columns from the event’s key as the table’s primary key.
record_value
Specifies that the primary key columns is sourced from the event’s value. You can set the primary.key.fields property to define the primary key as a subset of fields from the event’s value; otherwise all fields are used by default.

primary.key.fields

No default

Either the name of the primary key column or a comma-separated list of fields to derive the primary key from.

When primary.key.mode is set to record_key and the event’s key is a primitive type, it is expected that this property specifies the column name to be used for the key.

When the primary.key.mode is set to record_key with a non-primitive key, or record_value, it is expected that this property specifies a comma-separated list of field names from either the key or value. If the primary.key.mode is set to record_key with a non-primitive key, or record_value, and this property is not specified, the connector derives the primary key from all fields of either the record key or record value, depending on the specified mode.

quote.identifiers

false

Specifies whether generated SQL statements use quotation marks to delimit table and column names. See the JDBC quoting case-sensitivity section for more details.

schema.evolution

none

Specifies how the connector evolves the destination table schemas. For more information, see Section 3.1.1.9, “Schema evolution modes for the Debezium JDBC connector”. The following options are available:

none
Specifies that the connector does not evolve the destination schema.
basic
Specifies that basic evolution occurs. The connector adds missing columns to the table by comparing the incoming event’s record schema to the database table structure.

collection.name.format

${topic}

Specifies a string pattern that the connector uses to construct the names of destination tables. When the property is set to its default value, ${topic}, after the connector reads an event from Kafka, it writes the event record to a destination table with a name that matches the name of the source topic.

You can also configure this property to extract values from specific fields in incoming event records and then use those values to dynamically generate the names of target tables. This ability to generate table names from values in the message source would otherwise require the use of a custom Kafka Connect single message transformation (SMT).

To configure the property to dynamically generate the names of destination tables, set its value to a pattern such as ${source._field_}. When you specify this type of pattern, the connector extracts values from the source block of the Debezium change event, and then uses those values to construct the table name. For example, you might set the value of the property to the pattern ${source.schema}_${source.table}. Based on this pattern, if the connector reads an event in which the schema field in the source block contains the value, user, and the table field contains the value, tab, the connector writes the event record to a table with the name user_tab.

dialect.postgres.postgis.schema

public

Specifies the schema name where the PostgreSQL PostGIS extension is installed. The default is public; however, if the PostGIS extension was installed in another schema, this property should be used to specify the alternate schema name.

dialect.postgres.unnest.insert.enabled

false

Specifies whether to enable UNNEST-based batch inserts for PostgreSQL.

When enabled, uses PostgreSQL UNNEST() for batch inserts, which can significantly improve performance by reducing the number of SQL statements executed. This optimization is compatible with INSERT and UPSERT modes.

For information about the performance benefits of enabling the UNNEST function, see "Boosting Postgres Insert Performance" in the Tiger Data blog.

dialect.sqlserver.identity.insert

false

Specifies whether the connector automatically sets an IDENTITY_INSERT before an INSERT or UPSERT operation into the identity column of SQL Server tables, and then unsets it immediately after the operation. When the default setting (false) is in effect, an INSERT or UPSERT operation into the IDENTITY column of a table results in a SQL exception.

batch.size

500

Specifies how many records to attempt to batch together into the destination table.

Note

Note that if you set consumer.max.poll.records in the Connect worker properties to a value lower than batch.size, batch processing will be caped by consumer.max.poll.records and the desired batch.size won’t be reached. You can also configure the connector’s underlying consumer’s max.poll.records using consumer.override.max.poll.records in the connector configuration.

use.reduction.buffer

false

Specifies whether to enable the Debezium JDBC connector’s reduction buffer.

Choose one of the following settings:

false
(default) The connector writes each change event that it consumes from Kafka as a separate logical SQL change.
true
The connector uses the reduction buffer to reduce change events before it writes them to the sink database. That is, if multiple events refer to the same primary key, the connector consolidates the SQL queries and writes only a single logical SQL change, based on the row state that is reported in the most recent offset record. Choose this option to reduce the SQL load on the target database. Do not enable the reduction buffer if destination constraints create dependencies between records with different primary keys. Reduction by primary key can remove intermediate row states that are required to satisfy secondary unique constraints.

To optimize query processing in a PostgreSQL sink database when the reduction buffer is enabled, you must also enable the database to execute the batched queries by adding the reWriteBatchedInserts parameter to the JDBC connection URL.

field.include.list

empty string

An optional, comma-separated list of field names that match the fully-qualified names of fields to include from the change event value. Fully-qualified names for fields are of the form fieldName or topicName:_fieldName_.

If you include this property in the configuration, do not set the field.exclude.list property.

field.exclude.list

empty string

An optional, comma-separated list of field names that match the fully-qualified names of fields to exclude from the change event value. Fully-qualified names for fields are of the form fieldName or topicName:_fieldName_.

If you include this property in the configuration, do not set the field.include.list property.

flush.max.retries

5

Specifies the maximum number of retries that the connector performs after an attempt to flush changes to the target database results in certain database errors. If the number of retries exceeds the retry value, the sink connector enters a FAILED state.

flush.retry.delay.ms

1000

Specifies the number of milliseconds that the connector waits to retry a flush operation that failed.

Note

When you set both the flush.retry.delay.ms and flush.max.retries properties, it can affect the behavior of the Kafka max.poll.interval.ms property. To prevent the connector from rebalancing, set the total retry time (flush.retry.delay.ms * flush.max.retries) to a value that is less than the value of max.poll.interval.ms (default is 5 minutes).

3.1.5.4. JDBC connector extendable properties
Copy link

The following table describes the properties that allow you to customize the naming strategies the Debezium JDBC sink connector uses to resolve destination table names and column names from incoming event data.

PropertyDefaultDescription

column.naming.strategy

io.debezium.connector.jdbc.naming.DefaultColumnNamingStrategy

Specifies the fully-qualified class name of a ColumnNamingStrategy implementation that the connector uses to resolve column names from incoming event field names.

The default behavior is to use the field name as the column name without any transformation.

collection.naming.strategy

io.debezium.connector.jdbc.nnaming.DefaultCollectionNamingStrategy

Specifies the fully-qualified class name of a CollectionNamingStrategy implementation that the connector uses to resolve table names from incoming event topic names.

The default behavior is to:

  • Sanitize the topic name by replacing dots (.) with underscores (_).
  • Replace the ${topic} placeholder in the collection.name.format configuration property with the event’s topic.

Kafka Connect supports passthrough configuration, enabling you to modify the behavior of an underlying system by passing certain properties directly from the connector configuration. By default, some Hibernate properties are exposed via the JDBC connector connection properties (for example, connection.url, connection.username, and connection.pool.*_size), and through the connector’s runtime properties (for example, use.time.zone, quote.identifiers).

If you want to customize other Hibernate behavior, you can take advantage of the passthrough mechanism by adding properties that use the hibernate.* namespace to the connector configuration. For example, to assist Hibernate in resolving the type and version of the target database, you can add the hibernate.dialect property and set it to the fully qualified class name of the database, for example, org.hibernate.dialect.MariaDBDialect.

3.1.6. JDBC connector frequently asked questions
Copy link

The following questions and answers address common decisions and scenarios when you configure and operate the Debezium JDBC sink connector, including event transformation requirements, schema evolution limitations, data type mapping, and table naming options.

Is the ExtractNewRecordState single message transformation required?
No, that is actually one of the differentiating factors of the Debezium JDBC connector from other implementations. While the connector is capable of ingesting flattened events like its competitors, it can also ingest Debezium’s complex change event structure natively, without requiring any specific type of transformation.
If a column’s type is changed, or if a column is renamed or dropped, is this handled by schema evolution?
No, the Debezium JDBC connector does not make any changes to existing columns. The schema evolution supported by the connector is quite basic. It simply compares the fields in the event structure to the table’s column list, and then adds any fields that are not yet defined as columns in the table. If a column’s type or default value change, the connector does not adjust them in the destination database. If a column is renamed, the old column is left as-is, and the connector appends a column with the new name to the table; however existing rows with data in the old column remain unchanged. These types of schema changes should be handled manually.
If a column’s type does not resolve to the type that I want, how can I enforce mapping to a different data type?
The Debezium JDBC connector uses a sophisticated type system to resolve a column’s data type. For details about how this type system resolves a specific field’s schema definition to a JDBC type, see the Section 3.1.1.4, “Description of Debezium JDBC connector data and column type mappings” section. If you want to apply a different data type mapping, define the table manually to explicitly obtain the preferred column type.
How do you specify a prefix or a suffix to the table name without changing the Kafka topic name?
In order to add a prefix or a suffix to the destination table name, adjust the collection.name.format connector configuration property to apply the prefix or suffix that you want. For example, to prefix all table names with jdbc_, specify the collection.name.format configuration property with a value of jdbc_${topic}. If the connector is subscribed to a topic called orders, the resulting table is created as jdbc_orders.
Why are some columns automatically quoted, even though identifier quoting is not enabled?
In some situations, specific column or table names might be explicitly quoted, even when quote.identifiers is not enabled. This is often necessary when the column or table name starts with or uses a specific convention that would otherwise be considered illegal syntax. For example, when the primary.key.mode is set to kafka, some databases only permit column names to begin with an underscore if the column’s name is quoted. Quoting behavior is dialect-specific, and varies among different types of database.

The Debezium MongoDB sink connector captures change event records from Apache Kafka topics and then transforms the records into MongoDB documents that it writes to collections in a specified MongoDB sink database. For applications that require high scalability and fast data retrieval, propagating change data to a cluster-based MongoDB environment, which uses such features as sharding and replica sets to optimize read operations, can significantly improve retrieval performance. The connector can process only change events that originate from a Debezium relational database connector.

Important

The Debezium MongoDB sink connector is Developer Preview software only. Developer Preview software is not supported by Red Hat in any way and is not functionally complete or production-ready. Do not use Developer Preview software for production or business-critical workloads. Developer Preview software provides early access to upcoming product software in advance of its possible inclusion in a Red Hat product offering. Customers can use this software to test functionality and provide feedback during the development process. This software might not have any documentation, is subject to change or removal at any time, and has received limited testing. Red Hat might provide ways to submit feedback on Developer Preview software without an associated SLA. For more information about the support scope of Red Hat Developer Preview software, see Developer Preview Support Scope.

For information about the MongoDB versions that are compatible with this connector, see the Debezium Supported Configurations page.

Information about the Debezium sink connector for MongoDB and and instructions about how to use it is available in the following topics:

The Debezium MongoDB sink connector mirrors the state of tables in a relational source database to collections in a MongoDB database by consuming Debezium change events from Kafka topics and applying them as document-level operations.

The connector subscribes to Kafka topics that are populated with event messages produced by Debezium relational database source connectors. Each event message describes a database operation (insert, update, or delete) in a structured format that captures the details of the event. The connector transforms incoming change event records into MongoDB document format, and then writes the resulting documents into the target MongoDB collection.

After it receives an event, the connector parses the event payload and determines which MongoDB collection to send it to. Depending on the event type that is specified in the event payload, the connector then performs one of the following operations in the target collection:

Event type in payloadResulting operation

INSERT

Create document

UPDATE

Modify document with the specified identifier.

DELETE

Remove document with the specified identifier.

The connector uses the MongoDB Java driver to interact with the MongoDB database.

The mapping between topics and MongoDB collections is derived from the connector configuration. The document key serves as a unique identifier for the document, ensuring that updates, inserts, and deletions are propagated to the correct MongoDB document and collection, and that operations are applied in the correct order.

Through this process of mapping event messages to MongoDB documents, the connector is able to mirror the state of tables in your relational database to collections in a MongoDB database.

The Debezium MongoDB sink connector has the following limitations:

Relational database / RDBMS source connectors only

The MongoDB sink connector can consume only change events that originate from the Debezium connectors for the following relational databases:

  • MariaDB
  • MySQL
  • Oracle
  • PostgreSQL
  • SQL Server

The connector cannot process change event messages from any other Debezium connectors, including the Debezium MongoDB source connector.

Schema evolution
Although the connector can process basic schema changes, advanced schema evolution scenarios might require manual intervention or specific configuration. Because MongoDB is schemaless, it has only a limited ability to handle schema evolution.
Transaction Support
The connector processes individual change events in chronological order, based on the order in which operations are committed in the source system. Although MongoDB supports transactions, the Debezium MongoDB connector does not provide transactional guarantees across multiple CDC events or across multiple documents within a single sink task.

Deploy a basic instance of the MongoDB sink connector for testing.

Prerequisites

The following components are available and running in your environment:

  • Kafka cluster
  • Kafka Connect
  • A MongoDB instance.
  • Debezium relational database connector
  • Debezium MongoDB connector

Procedure

  1. Configure and start a Debezium source connector, for example, a Debezium PostgreSQL Connector, to stream changes from a relational database to Kafka.

  2. Configure and start the Debezium MongoDB sink connector to consume events that the source connector emits to Kafka, and send them to a MongoDB sink database.

    The following example provides a minimal configuration for a Debezium MongoDB sink connector. Replace the placeholders in the example with the actual values for your environment. 

    {
  "name": "mongodb-sink-connector",
  "config": {
    "connector.class": "io.debezium.connector.mongodb.sink.MongoDbSinkConnector",
    "topics.regex": "server1\.inventory\..*",
    "mongodb.connection.string": "mongodb://localhost:27017",
    "sink.database": "debezium"
  }
}

The MongoDB sink connector accepts a variety of configuration options, as described in following tables.

Table 3.2. Required Kafka Connect sink connector configuration properties
PropertyDefaultDescription

connector.class

No default value

Must be set to io.debezium.connector.mongodb.sink.MongoDbSinkConnector.

tasks.max

1

Maximum number of tasks.

topics or topics.regex

No default value

List of Kafka topics to consume from. If you set this value to topics.regex, the connector consumes from all topics that match the regular expression.

Table 3.3. Required MongoDB connection properties
PropertyDefaultDescription

mongodb.connection.string

No default value

MongoDB connection string (URI) that the sink uses to connect to MongoDB. This URI follows the standard MongoDB connection string format.

Example: mongodb://localhost:27017/?replicaSet=my-replica-set

sink.database

No default value

Name of the target MongoDB database.

Table 3.4. Sink behavior configuration
PropertyDefaultDescription

collection.naming.strategy

io.debezium.sink.naming.DefaultCollectionNamingStrategy

Specifies the strategy that the connector uses to derive the name of the target MongoDB collection from the name of the Kafka topic.

Specify one of the following values:

io.debezium.sink.naming.DefaultCollectionNamingStrategy
The connector takes the table name directly from the topic name, replacing dot (period) characters in the name of the source topic with underscores.
Custom implementation
You can provide your own CollectionNameStrategy implementation.

collection.name.format

${topic}

Template for deriving the target collection name from the Kafka topic name.

column.naming.strategy

io.debezium.sink.naming.DefaultColumnNamingStrategy

Specifies the strategy that the connector uses to name columns in the target collection.

Specify one of the following values:

io.debezium.sink.naming.DefaultColumnNamingStrategy
Use the original field name as the column name.
Custom implementation
Specify a custom CollectionNameStrategy implementation.
Table 3.5. Common sink options
PropertyDefaultDescription

field.include.list

empty string

An optional, comma-separated list of field names that match the fully-qualified names of fields to include from the change event value. Fully-qualified names for fields are of the form fieldName or topicName:fieldName.

If you include this property in the configuration, do not set the field.exclude.list property.

field.exclude.list

empty string

An optional, comma-separated list of field names that match the fully-qualified names of fields to exclude from the change event value. Fully-qualified names for fields are of the form fieldName or topicName:fieldName.

If you include this property in the configuration, do not set the field.include.list property.

batch.size

2048

Maximum number of records to write in a single batch.

The following example demonstrates a connector configuration that reads change events from specific topics in a relational database and writes the resulting documents to a MongoDB sink database. The example configures the connector to capture change events from the dbserver1.inventory.customers, dbserver1.inventory.orders, and dbserver1.inventory.products topics and writes the data to documents in a debezium collection in the MongoDB sink database.

The following example shows how you might configure the connector to read change events from three specific topics from the dbserver1.inventory database to modify a collection in the MongoDB sink database named debezium. 

{
    "name": "mongodb-sink-connector",
    "config": {
        "connector.class": "io.debezium.connector.mongodb.sink.MongoDbSinkConnector",
        "topics": "dbserver1.inventory.customers,dbserver1.inventory.orders,dbserver1.inventory.products",
        "mongodb.connection.string": "mongodb://localhost:27017",
        "sink.database": "debezium"
    }
}

This release of the connector does not expose any metrics.

When the connector processes events, it maps data to specific fields in the target MongoDB document.

  • Keys from Debezium change events, such as Kafka message keys, are mapped to the MongoDB _id field by default.
  • Values are mapped into MongoDB documents.
  • Updates and deletes are resolved based on the key field mapping.

The following example shows an event key in a Kafka topic: 

{
    "userId": 1,
    "orderId": 1
}

Based on the mapping logic, the preceding key is mapped to the _id field in a MongoDB document, as showin in the following example: 

{
    "_id": {
        "userId": 1,
        "orderId": 1
    }
}

The Debezium MongoDB sink connector can consume records serialized as CloudEvents. Debezium can emit change events in CloudEvents format, so that the event payload is encapsulated in a standardized envelope.

When you enable CloudEvents on the source connector, the MongoDB sink connector parses the CloudEvents envelope.

The actual Debezium event payload is extracted from the data section.

The event is then applied to the target MongoDB collection, following the standard insert, update, or delete semantics.

This process makes it possible to integrate Debezium with broader event-driven systems while still persisting the resulting events in MongoDB.

Table 3.6. CloudEvents sink options
PropertyDefaultDescription

cloud.events.schema.name.pattern

.*CloudEvents\.Envelope$

Regular expression pattern to identify CloudEvents messages by matching the schema name with this pattern.

Red Hat logoGithubRedditYoutubeTwitter

Learn

Try, buy, & sell

Communities

About Red Hat

We deliver hardened solutions that make it easier for enterprises to work across platforms and environments, from the core datacenter to the network edge.

Making open source more inclusive

Red Hat is committed to replacing problematic language in our code, documentation, and web properties. For more details, see the Red Hat Blog.

About Red Hat Documentation

We help Red Hat users innovate and achieve their goals with our products and services with content they can trust. Explore our recent updates.

Legal Notice

Theme

© 2026 Red HatBack to top