235 lines
12 KiB
Plaintext
235 lines
12 KiB
Plaintext
// Category: debezium-using
|
|
// Type: assembly
|
|
// ModuleID: developing-debezium-custom-data-type-converters
|
|
// Title: Developing {prodname} custom data type converters
|
|
[id="custom-converters"]
|
|
= Custom Converters
|
|
|
|
ifdef::community[]
|
|
:source-highlighter: highlight.js
|
|
:toc:
|
|
:toc-placement: macro
|
|
:linkattrs:
|
|
:icons: font
|
|
toc::[]
|
|
|
|
[NOTE]
|
|
====
|
|
This feature is currently in incubating state, i.e. exact semantics, configuration options etc. may change in future revisions, based on the feedback we receive. Please let us know if you encounter any problems while using this extension.
|
|
====
|
|
|
|
== Datatype Conversion
|
|
endif::community[]
|
|
|
|
ifdef::product[]
|
|
[IMPORTANT]
|
|
====
|
|
The use of custom-developed converters is a Technology Preview feature only.
|
|
Technology Preview features are not supported with Red Hat production service level agreements (SLAs) and might not be functionally complete.
|
|
Red Hat does not recommend using them in production.
|
|
These features provide early access to upcoming product features, enabling customers to test functionality and provide feedback during the development process.
|
|
For more information about the support scope of Red Hat Technology Preview features, see link:https://access.redhat.com/support/offerings/techpreview[https://access.redhat.com/support/offerings/techpreview].
|
|
====
|
|
endif::product[]
|
|
|
|
Each field in a {prodname} change event record represents a field or column in the source table or data collection.
|
|
The connector converts data types in the source to a corresponding Kafka Connect schema types.
|
|
Column values are likewise converted to match the schema type of the destination field.
|
|
For each connector, a default mapping specifies how the connector converts each data type.
|
|
The documentation for each connector provides details about the default mappings that the connector uses to convert data types.
|
|
|
|
The default mappings are sufficient to satisfy most needs, but for some applications it might be necessary to apply an alternate mapping.
|
|
For example, the default mapping for a column might export values using the format of milliseconds since the UNIX epoch, but you have a downstream application that requires the values to be formatted strings.
|
|
To customize data type mappings you can develop and deploy custom converters.
|
|
You can configure a custom converter to apply to all columns of a certain type, or to a specific table column only.
|
|
The converter function intercepts conversion requests for columns that match a specified criteria, and performs the specified format conversion.
|
|
The converter ignores columns that do not match the specified criteria.
|
|
|
|
Custom converters are Java classes that implement the Debezium service provider interface (SPI).
|
|
You enable and configure a custom converter by setting the `converters` property in the connector configuration.
|
|
The `converters` property defines the criteria for identifying the columns that you want the converter to process and provides other details that determine conversion behavior.
|
|
|
|
After you start a connector, any converters that are enabled in the connector configuration are instantiated and are added to a registry.
|
|
The registry associates each converter with the columns or fields for it to process.
|
|
Whenever {prodname} processes a new change event, it invokes the configured converter to convert the columns or fields for which it is registered.
|
|
|
|
// Type: procedure
|
|
// Title: Creating a {prodname} custom data type converter
|
|
// ModuleID: creating-a-debezium-custom-data-type-converter
|
|
[id="implementing-a-custom-converter"]
|
|
== Implementing custom converters
|
|
|
|
The following example shows a converter implementation of a Java class that implements the interface `io.debezium.spi.converter.CustomConverter`:
|
|
|
|
[source,java,indent=0]
|
|
----
|
|
public interface CustomConverter<S, F extends ConvertedField> {
|
|
|
|
@FunctionalInterface
|
|
interface Converter { // <1>
|
|
Object convert(Object input);
|
|
}
|
|
|
|
public interface ConverterRegistration<S> { // <2>
|
|
void register(S fieldSchema, Converter converter); // <3>
|
|
}
|
|
|
|
void configure(Properties props);
|
|
|
|
void converterFor(F field, ConverterRegistration<S> registration); // <4>
|
|
}
|
|
----
|
|
<1> A function for converting data from one type to another.
|
|
<2> Callback for registering a converter.
|
|
<3> Registers the given schema and converter for the current field.
|
|
Should not be invoked more than once for the same field.
|
|
<4> Registers the customized value and schema converter for use with a specific field.
|
|
|
|
.Custom converter methods
|
|
The `configure()` and `converterFor()` methods are mandatory for each {prodname} custom converter:
|
|
|
|
`configure()`::
|
|
Passes the properties specified in the connector configuration to the converter instance.
|
|
The `configure` method runs when the connector is initialized.
|
|
You can use a converter with multiple connectors and modify its behavior based on the connector's property settings. +
|
|
The `configure` method accepts the following argument:
|
|
|
|
`props`::: Contains the properties to pass to the converter instance.
|
|
Each property specifies the format for converting the values of a particular type of column.
|
|
|
|
`converterFor()`::
|
|
Registers the converter to process specific columns or fields in the data source.
|
|
{prodname} invokes the `converterFor()` method to prompt the converter to call `registration` for the conversion.
|
|
The `converterFor` method runs once for each column. +
|
|
The method accepts the following arguments:
|
|
|
|
`field`:::
|
|
An object that passes metadata about the field or column that is processed.
|
|
The column metadata can include the name of the column or field, the name of the table or collection, the data type, size, and so forth.
|
|
|
|
`registration`:::
|
|
An object of type `io.debezium.spi.converter.CustomConverter.ConverterRegistration` that provides the target schema definition and the code for converting the column data.
|
|
The converter calls the `registration` parameter when the source column matches the type that the converter should process.
|
|
calls the `register` method to define the converter for each column in the schema.
|
|
Schemas are represented using the Kafka Connect link:https://kafka.apache.org/31/javadoc/org/apache/kafka/connect/data/SchemaBuilder.html[`SchemaBuilder`] API.
|
|
ifdef::community[]
|
|
In the future, an independent schema definition API will be added.
|
|
endif::community[]
|
|
|
|
The following example implements a simple converter that performs the following operations:
|
|
|
|
* Runs the `configure` method, which configures the converter based on the value of the `schema.name` property that is specified in the connector configuration.
|
|
The converter configuration is specific to each instance.
|
|
* Runs the `converterFor` method, which registers the converter to process values in source columns for which the data type is set to `isbn`.
|
|
** Identifies the target `STRING` schema based on the value that is specified for the `schema.name` property.
|
|
** Converts ISBN data in the source column to `String` values.
|
|
|
|
=== {prodname} custom converter example
|
|
[id="example-debezium-simple-custom-converter"]
|
|
.A simple custom converter
|
|
====
|
|
[source,java,indent=0]
|
|
----
|
|
public static class IsbnConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {
|
|
|
|
private SchemaBuilder isbnSchema;
|
|
|
|
@Override
|
|
public void configure(Properties props) {
|
|
isbnSchema = SchemaBuilder.string().name(props.getProperty("schema.name"));
|
|
}
|
|
|
|
@Override
|
|
public void converterFor(RelationalColumn column,
|
|
ConverterRegistration<SchemaBuilder> registration) {
|
|
|
|
if ("isbn".equals(column.typeName())) {
|
|
registration.register(isbnSchema, x -> x.toString());
|
|
}
|
|
}
|
|
}
|
|
----
|
|
====
|
|
|
|
// Type: procedure
|
|
[id="debezium-and-kafka-connect-api-module-dependencies"]
|
|
=== {prodname} and Kafka Connect API module dependencies
|
|
|
|
The converter code depends on the {prodname} and Kafka Connect API library modules.
|
|
To enable your converter code to compile, add these dependencies to your converter Java project as shown in the following example:
|
|
|
|
[source,xml]
|
|
----
|
|
<dependency>
|
|
<groupId>io.debezium</groupId>
|
|
<artifactId>debezium-api</artifactId>
|
|
<version>${version.debezium}</version> // <1>
|
|
</dependency>
|
|
<dependency>
|
|
<groupId>org.apache.kafka</groupId>
|
|
<artifactId>connect-api</artifactId>
|
|
<version>${version.kafka}</version> <2>
|
|
</dependency>
|
|
----
|
|
<1> `${version.debezium}` represents the version of the {prodname} connector.
|
|
<2> `${version.kafka}` represents the version of Apache Kafka in your environment.
|
|
|
|
// Type: assembly
|
|
// Title: Using custom converters with {prodname} connectors
|
|
// ModuleID: deploying-and-configuring-debezium-custom-data-type-converters
|
|
[id="configuring-and-using-converters"]
|
|
== Configuring and Using Converters
|
|
|
|
To use the converter with a connector, you deploy the converter JAR file alongside the connector file, and then configure the connector to use the converter.
|
|
|
|
// Type: procedure
|
|
[id="deploying-a-debezium-custom-converter"]
|
|
=== Deploying a custom converter
|
|
|
|
.Procedure
|
|
* To use a custom converter with a {prodname} connector, export the Java project to a JAR file, and add the file to the directory that contains the JAR file for each {prodname} connector that you want to use it with. +
|
|
+
|
|
For example, in a typical deployment, you might store {prodname} connector files in subdirectories of a Kafka Connect directory, such as `/kafka/connect`,
|
|
and then store the JAR for each connector in its own subdirectory (`debezium-connector-db2`, `debezium-connector-mysql`, and so forth).
|
|
To use a converter with a connector, add the converter JAR file to the connector subdirectory.
|
|
|
|
NOTE: To use a converter with multiple connectors, add the connector JAR file to the directory for each of the connectors.
|
|
|
|
// Type: procedure
|
|
[id="configuring-a-connectors-to-use-a-custom-converter"]
|
|
=== Configuring a connector to use a custom converter
|
|
|
|
Custom converters act on specific columns or column types in a source table to specify how to convert their data types.
|
|
To enable a connector to use the custom converter, you add properties to the connector configuration that specify the converter name and class.
|
|
If the converter requires further information to customize the formats of specific data types, you can also define other coniguration options to provide that information.
|
|
|
|
.Prerequisites
|
|
* You have a custom converter Java program.
|
|
|
|
.Procedure
|
|
|
|
* Enable a converter for a connector instance by adding the following mandatory properties to the connector configuration:
|
|
+
|
|
[subs="+quotes"]
|
|
----
|
|
converters: _<converterSymbolicName>_ // <1>
|
|
_<converterSymbolicName>_.type: _<fullyQualifiedConverterClassName>_ // <2>
|
|
----
|
|
<1> The `converters` property is mandatory and enumerates a comma-separated list of symbolic names of the converter instances to use with the connector.
|
|
The values listed for this property serve as prefixes in the names of other properties that you specify for the converter.
|
|
<2> The `_<converterSymbolicName>_.type` property is mandatory, and specifies the name of the class that implements the converter.
|
|
For example, for the earlier xref:example-debezium-simple-custom-converter[custom converter example], you would add the following properties to the connector configuration:
|
|
+
|
|
----
|
|
converters: isbn
|
|
isbn.type: io.debezium.test.IsbnConverter
|
|
----
|
|
|
|
* If provide further configuration properties for a converter, prefix the property names with the symbolic name of the converter, followed by a dot (`.`).
|
|
The symbolic name is label that you specify as a value for the `converters` property.
|
|
For example, to add a property for the preceding `isbn` converter to specify the `schema.name` to pass to the `configure` method in the converter code, add the following property:
|
|
+
|
|
----
|
|
isbn.schema.name: io.debezium.postgresql.type.Isbn
|
|
----
|