Debezium Custom Converters - TimestampConverter
Tags: debezium, cdc, java, tutorial
Published: 2020-05-26 09:37:29

---
Title: Debezium Custom Converters  
Subtitle: Creating custom converters using Debezium's new SPI to override value conversions  
Date: April 25, 2020  
Author: Oryan Moshe  
Email: oryan@moshe.ai  
Web: https://in.dev/  
Blog: https://dev.to/oryanmoshe/  
GitHub: https://github.com/oryanmoshe/  
---

# Debezium Custom Converters #
_Creating custom converters using Debezium's new SPI to override value conversions_
## Introduction ##
### Background about the `TimestampConverter`
Hey, my name is Oryan Moshe, and I started my own community for senior developers in Israel named [in.dev](https://in.dev/).  
I'm also a Lead Architect at [Rivery](https://rivery.io/), We are a fully managed data integration platform, and part of my job here is developing the ability to stream changes straight from the clients’ databases, to our platform, using change data capture (CDC).

The last time I coded in Java was 7 years ago, so if you have any suggestions to improve the code shown here please feel free to comment below!

You can find the converter right here:
[https://github.com/oryanmoshe/debezium-timestamp-converter/](https://github.com/oryanmoshe/debezium-timestamp-converter/)

### CDC — Change Data Capture ###

Before we talk about Debezium, we have to talk about CDC.

CDC is a way for us to get the _changes_ happening in the database (as opposed to the _actual data_)  
This means we can actually get _every state_ that _every record_ has been through in the database.

CDC is useful for a number of cases:

- Compiling a log of record changes
- Undoing (or reverting) a change
- Tracking record deletion (which is not simply a matter of using `SELECT`)

### What is Debezium Anyway? ###
[Debezium](https://debezium.io) is an open source platform, maintained by Red Hat, that allows developers to implement CDC into a Kafka infrastructure.  
Debezium actuates CDC by configuring connections using the provided Kafka Connect data connectors. Currently there's support for MySQL, PostgreSQL, Microsoft SQL Server, MongoDB, and even some limited support for Oracle.

### What are converters, and why would we need a custom one?
All messages produced by Debezium are processed before entering the designated topic.  
This ensures that all fields of a given type (defined by the schema) behave the same.  
In other words, _all_ `DATE` fields on _all_ of the databases will be transformed into the same format. This is, by default "Days since epoch".  
But this behavior isn't always wanted, especially in this temporal example.

For **our** particular use case we need all temporal fields to be in the same format, whether the type is `DATE`, `DATETIME`, `DATETIME2`, `TIME` or `TIMESTAMP`.  
The format we chose was `YYYY-MM-dd'T'HH:mm:ss.SSS'Z'`.  

---

## Creating a custom converter
Here's an explanation for each step needed to create our `TimestampConverter`.  

### The basics of custom converters
To allow such behavior, the Debezium SPI (Service Provider Interface) was added to Debezium Version 1.1.  
This allows developers to make their own converters with Java, by creating a class that implements the `io.debezium.spi.converter.CustomConverter` interface.

### The First Gotcha
What we didn't know when we started developing this converter, is that once we registered a custom converter to a temporal column, Debezium's behavior became sporadic. Sometimes it'll pass a `DATE` column as "Days since epoch", as expected, but sometimes it'll pass it as a string, matching the date format of the database it came from.

This meant we had to have all of our bases covered, both for numeric values (let's say, "Days since epoch") and for all date format databases can produce (`YYYY-MM-dd`, `dd/MM/YYYY`, `YYYY-MMM-dd`, etc.)

Things got _a bit_ complicated on the logic front, but let's not get into this right now.

### What's needed for our custom converter to work
Each converter has to implement at least two methods to be harnessed by Debezium:  

#### configure
This method runs when the connector is initialised. It accepts one argument:

**props** 
An object of type `java.util.Properties`, containing all of the properties we passed to our converter instance.

#### converterFor
This method runs once for each column defined in our schema, and its job is to define (a.k.a "register") the converter for each. It accepts two arguments:

**column** 
An object of type `io.debezium.spi.converter.RelationalColumn`, containing the definition of the column we're currently handling, including its name, type, size, table, etc.

**registration** 
An object of type `io.debezium.spi.converter.CustomConverter.ConverterRegistration`, an internal definition, that has one method: `register`.

### Using the `configure` method
As stated above, we use the `configure` method to pass properties into our converter. This is important because **we can use the same converter for multiple connectors**, and change its behavior according to these properties.  

For our `TimestampConverter` we wanted to pass four properties:

- `debug` – Indicates whether to print debug messages. Defaults to `false`.
- `format.date` – The format to convert all columns of type `DATE`. Defaults to `YYYY-MM-dd`.
- `format.time` – The format to convert all columns of type `TIME`. Defaults to `HH:mm:ss`.
- `format.datetime` – The format to convert **all other temporal columns**. Defaults to `YYYY-MM-dd'T'HH:mm:ss.SSS'Z'`.

All of these properties are optional and have default values associated with them.  
To support them we defined each of them as a class property with the default value. Inside the `configure` method we assigned them with the passed value:

```java
public class TimestampConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {

    public static final String DEFAULT_DATETIME_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";
    public static final String DEFAULT_DATE_FORMAT = "YYYY-MM-dd";
    public static final String DEFAULT_TIME_FORMAT = "HH:mm:ss.SSS";

    public String strDatetimeFormat, strDateFormat, strTimeFormat;
    public Boolean debug;

    private SimpleDateFormat simpleDatetimeFormatter, simpleDateFormatter, simpleTimeFormatter;

    @Override
    public void configure(Properties props) {
        this.strDatetimeFormat = props.getProperty("format.datetime", DEFAULT_DATETIME_FORMAT);
        this.simpleDatetimeFormatter = new SimpleDateFormat(this.strDatetimeFormat);

        this.strDateFormat = props.getProperty("format.date", DEFAULT_DATE_FORMAT);
        this.simpleDateFormatter = new SimpleDateFormat(this.strDateFormat);

        this.strTimeFormat = props.getProperty("format.time", DEFAULT_TIME_FORMAT);
        this.simpleTimeFormatter = new SimpleDateFormat(this.strTimeFormat);

        this.debug = props.getProperty("debug", "false").equals("true");

        this.simpleDatetimeFormatter.setTimeZone(TimeZone.getTimeZone("UTC"));
        this.simpleTimeFormatter.setTimeZone(TimeZone.getTimeZone("UTC"));
    }
}
```

### Using the `converterFor` method
Now it's time for the big moment. Each column must be converted to its respective format.

First of all, we have to understand the type of the column we're currently handling. This is determined using `column.typeName`.  
If the type is any of the temporal types (defined as a class constant) we handle it accordingly. If it's not, we do nothing, and Debezium will take control.

To tell Debezium to convert a specific column to something else, we need to use the `registration` passed to us. Then `register` it, providing a `schema` (create one of type `string` and make it `optional`) and a converter.

The converter is just a function, or in our case a lambda, that receives an `Object`. This is the source value, and returns a value matching the schema we provided. In our case, we needed to return a `String` (or `null`, because we made it `optional`).

```java
@Override
public void converterFor(RelationalColumn column, ConverterRegistration<SchemaBuilder> registration) {
    if (SUPPORTED_DATA_TYPES.stream().anyMatch(s -> s.toLowerCase().equals(column.typeName().toLowerCase()))) {
        boolean isTime = "time".equals(column.typeName().toLowerCase());
        registration.register(datetimeSchema, rawValue -> {
            if (rawValue == null)
                return rawValue;

            Long millis = getMillis(rawValue.toString(), isTime);
           
            Instant instant = Instant.ofEpochMilli(millis);
            Date dateObject = Date.from(instant);
            
            switch (column.typeName().toLowerCase()) {
                case "time":
                    return simpleTimeFormatter.format(dateObject);
                case "date":
                    return simpleDateFormatter.format(dateObject);
                default:
                    return simpleDatetimeFormatter.format(dateObject);
            }
        });
    }
}
```

In this code snippet look at the two crucial parts we have mentioned before. These are the call to `registration.register`, and the `return` statements.  

## Using a Custom Converter with Debezium
### Installation
Installation in our Debezium cluster is straight-forward. We just need to add the `.jar` file of the converter to the connector we want to use it in.
#### The Second Gotcha
Notice I said " ... to the connecter we want ... ", this is a thing Debezium didn't make clear in the documentation. We need to add this converter to **every connector** if we want to use it in.  
Let's say the base folder for connectors is `/kafka/connect`. Then inside we'll find folders like `debezium-connector-mysql`, or `debezium-connector-postgres`.

We need to add the converter `.jar` file to each of those folders if we intend to use it.

### Configuration
After adding the `.jar` file to our connector, we can configure our connectors to use it!

To do so all we need to do is add the following keys to our existing configuration:

```json
"converters": "timestampConverter",
"timestampConverter.type": "oryanmoshe.kafka.connect.util.TimestampConverter"
```

If we need to customize the formats of specific data types, we can use these additional configuration keys:

```json
"timestampConverter.format.time": "HH:mm:ss.SSS",
"timestampConverter.format.date": "YYYY-MM-dd",
"timestampConverter.format.datetime": "YYYY-MM-dd'T'HH:mm:ss.SSS'Z'",
"timestampConverter.debug": "false"
```

## Conclusions
The addition of an SPI to Debezium brought a lot to the table in term of custom converters.  
This allows us to get a tailored fit CDC connector, with the data streaming into our Kafka cluster exactly in the format we want.

I didn't include the actual logic, converting the values from their raw format into the epoch time (this part is contained in the `getMillis` method)  
**But** I have published the `TimestampConverter` as open source, so anyone can read the code there, use the converter in an application (be it as a `.jar` file found in the releases section, or as a dependency found in the packages section), or contribute to its development!

Feel free to suggest contributions to this converter, and share with me what kind of converters _you_ created using the new Debezium SPI, and which ones you wish were made!

### Links
To read more about Debezium Custom Converter visit their official documentation:
[https://debezium.io/documentation/reference/1.1/development/converters.html](https://debezium.io/documentation/reference/1.1/development/converters.html)

Link to the repository of my `TimestampConverter`:
[https://github.com/oryanmoshe/debezium-timestamp-converter](https://github.com/oryanmoshe/debezium-timestamp-converter)