Chapter 3. Sink connectors

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.

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:

The Debezium JDBC sink connector can delete rows in the destination database when a DELETE or tombstone event is consumed. By default, the JDBC sink connector does not enable delete mode.

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.

The Debezium JDBC sink connector can perform idempotent writes, enabling it to replay the same records repeatedly and not change the final database 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 the upsert DML syntax for the database dialects that Debezium supports:

You can use the following schema evolution modes with the Debezium JDBC sink connector:

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.

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.

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.c3p0.idle_test_period property. For example:

{
  "hibernate.c3p0.idle_test_period": "300"
}

{
  "hibernate.c3p0.idle_test_period": "300"
}

The Debezium JDBC sink connector uses the Hibernate C3P0 connection pool. You can customize the CP30 connection pool by setting properties in the hibernate.c3p0.*` configuration namespace. In the preceding example, the setting of the hibernate.c3p0.idle_test_period 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.

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:

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:

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

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:

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

{
  "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).

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:

The connector provides the following naming transformations:

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. Depending on the deployment method that you use, you can use one of the following metods to retrieve the drivers:

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):

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.

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.

You can deploy a Debezium JDBC connector by building a custom Kafka Connect container image that contains the Debezium connector archive, and then push this container image to a container registry. Afterwards, you create the following custom resources (CRs) to define the connector configuration:

cat <<EOF >debezium-jdbc-connector-container.yaml 
FROM registry.redhat.io/amq-streams/kafka-40-rhel9:3.0.0
USER root:root
RUN mkdir -p /opt/kafka/plugins/debezium 
RUN cd /opt/kafka/plugins/debezium/ \
&& curl -O https://maven.repository.redhat.com/ga/io/debezium/debezium-connector-jdbc/3.2.4.Final-redhat-00000/debezium-connector-jdbc-3.2.4.Final-redhat-00000-plugin.zip \
&& unzip debezium-connector-jdbc-3.2.4.Final-redhat-00000-plugin.zip \
&& rm debezium-connector-jdbc-3.2.4.Final-redhat-00000-plugin.zip
RUN cd /opt/kafka/plugins/debezium/ \
&& curl -O https://repo1.maven.org/maven2/com/oracle/ojdbc/ojdbc11/21.15.0.0/ojdbc11-21.15.0.0.jar
USER 1001
EOF

cat <<EOF >debezium-jdbc-connector-container.yaml 

FROM registry.redhat.io/amq-streams/kafka-40-rhel9:3.0.0
USER root:root
RUN mkdir -p /opt/kafka/plugins/debezium 

RUN cd /opt/kafka/plugins/debezium/ \
&& curl -O https://maven.repository.redhat.com/ga/io/debezium/debezium-connector-jdbc/3.2.4.Final-redhat-00000/debezium-connector-jdbc-3.2.4.Final-redhat-00000-plugin.zip \
&& unzip debezium-connector-jdbc-3.2.4.Final-redhat-00000-plugin.zip \
&& rm debezium-connector-jdbc-3.2.4.Final-redhat-00000-plugin.zip
RUN cd /opt/kafka/plugins/debezium/ \
&& curl -O https://repo1.maven.org/maven2/com/oracle/ojdbc/ojdbc11/21.15.0.0/ojdbc11-21.15.0.0.jar
USER 1001
EOF

 +
.Descriptions of Containerfile settings for building a custom Kafka Connect container image

 +
.Descriptions of Containerfile settings for building a custom Kafka Connect container image

+ The command creates a Containerfile with the name debezium-jdbc-connector-container.yaml in the current directory.

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.