Feature request or enhancement

Today, a hardcoded list of Converters are usable for both Headers and Key/Value Converters.

This list is here: https://github.com/debezium/debezium/blob/c691dcecd110aef0cb314311d24222e940eefc36/debezium-embedded/src/main/java/io/debezium/embedded/ConverterBuilder.java#L190-L221 

It is also set in debezium-server: https://github.com/debezium/debezium-server/blob/0faa19f3a4be598e1b0a2bc6f72460fe56fd1a74/debezium-server-core/src/main/java/io/debezium/server/DebeziumServer.java#L210-L230 

There is no reason we couldn't support Custom Converters short of these locked down if blocks.

Which use case/requirement will be addressed by the proposed feature?

We have a custom Header Converter for ByteArray Headers and Custom Value Converter for Large Records. Without an entry point, there is no way to use these without building Debezium with an additional if block

Example:

else if (isFormat(format, LargeRecord.class)) {
    converterConfig = converterConfig.edit().withDefault(FIELD_CLASS, "com.shopify.cdc.converters.LargeRecordConverter").build();
} 

Implementation ideas (optional)

This idea isn't optimal because rather than passing a converter you have to pass a Serialization Format. Short of using code that is within Debezium vs. Kafka, I don't see another way but open to ideas! This is just meant to be a starting point.

/**
* The logical description of the output format used by the given instance of \{@link DebeziumEngine}.
*/
public abstract class SerializationFormat<T> {
// The fully qualified class name of the converter class
private final String converterClass;

public SerializationFormat(String converterClass) {
this.converterClass = converterClass;
}

public String getConverterClass() {
return converterClass;
}
}

 
Then users would provide this class in a jar
 

/**
* A \{@link SerializationFormat} defining the Shopify Large Record format
*/
public class LargeRecordSerializationFormat extends SerializationFormat<byte[]> {
public LargeRecordSerializationFormat() {
super("com.shopify.cdc.converters.LargeRecordConverter");
}
}

These can then be class loaded in debezium-server here

So for example you would pass:

debezium.format.value: protobuf

or

debezium.format.value: com.shopify.cdc.formats.LargeRecordSerializationFormat

We will need to change debezium-embedded to take an instance of the class instead of the Class type.

Note: that apicurio would need to no longer use `json` with a special config. But in my opinion that isn't obvious anyway. Instead they would use `apicuriojson` and `apicurioavro`.