DatetimeValueConverter.java

package com.leapdb.cdc;
import io.debezium.spi.converter.CustomConverter;
import io.debezium.spi.converter.RelationalColumn;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import java.util.Properties;

public class DatetimeValueConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {

    private SchemaBuilder datetimeSchema;

    @Override
    public void configure(Properties props) {
        System.out.println(props);
        datetimeSchema = SchemaBuilder.string().name(props.getProperty("schema.name"));
    }

    @Override
    public void converterFor(RelationalColumn column,
            ConverterRegistration<SchemaBuilder> registration) {
                       
        if ("DATETIME".equals(column.typeName())) {
            System.out.println(column.typeName());
            registration.register(datetimeSchema, x -> 
              { System.out.println(x.getClass());
                
                System.out.println(x);
                if(x == null) { return null; }
                else if(x.toString().equals("0")) {
                  return("0000-00-00");

                } else return x.toString(); 
            });
        }
    }
}