Debezium 自定义mariadb转换器的完整代码

下面是一个自定义的 MariaDB 转换器的完整代码示例：

import io.debezium.converters.CloudEventsConverter;
import io.debezium.converters.spi.Serializer;
import io.debezium.converters.spi.Serde;
import io.debezium.converters.spi.Validator;
import io.debezium.data.Envelope;
import io.debezium.relational.TableId;
import io.debezium.serde.DebeziumSerdes;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.errors.DataException;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.Collections;
import java.util.Map;

public class MyMariaDBConverter implements CloudEventsConverter {

    private static final String FIELD_SCHEMA_DATA = "data";
    private static final String FIELD_SCHEMA_PAYLOAD = "payload";
    private static final String FIELD_SCHEMA_SOURCE = "source";
    private static final String FIELD_SCHEMA_SPECVERSION = "specversion";

    private static final String FIELD_SOURCE_DATABASE = "db";
    private static final String FIELD_SOURCE_TABLE = "table";

    private static final String SPEC_VERSION = "1.0";

    private final Serializer<SourceRecord> serializer = new Serializer<SourceRecord>() {
        @Override
        public byte[] serialize(SourceRecord sourceRecord) {
            Struct payload = new Struct(SCHEMA_PAYLOAD);
            Struct source = new Struct(SCHEMA_SOURCE);
            TableId tableId = ((TableId) sourceRecord.sourceOffset().get("table"));
            String dbName = tableId != null ? tableId.catalog() : null;
            String tableName = tableId != null ? tableId.table() : null;

            payload.put(FIELD_SCHEMA_DATA, sourceRecord.value());
            source.put(FIELD_SOURCE_DATABASE, dbName);
            source.put(FIELD_SOURCE_TABLE, tableName);

            return CloudEventsConverter.serialize(SPEC_VERSION, payload, source);
        }

        @Override
        public void configure(Map<String, ?> map, boolean b) {

        }

        @Override
        public void close() {

        }
    };

    private final Serde<SourceRecord> serde = DebeziumSerdes.payloadJson(Envelope.schemaBuilder()
            .withName("io.debezium.connector.mysql.Source")
            .withRecordSchemaChangeKey(Envelope.FieldName.DDL_STATEMENTS)
            .build(), Collections.emptyMap(), null);

    private final Validator<SourceRecord> validator = new Validator<SourceRecord>() {
        @Override
        public boolean isValid(SourceRecord sourceRecord) {
            return true;
        }

        @Override
        public void configure(Map<String, ?> map) {

        }

        @Override
        public void close() {

        }
    };

    private static final Schema SCHEMA_PAYLOAD = SchemaBuilder.struct()
            .name("io.debezium.connector.mysql.Source")
            .field(FIELD_SCHEMA_DATA, Schema.STRING_SCHEMA)
            .build();

    private static final Schema SCHEMA_SOURCE = SchemaBuilder.struct()
            .name("io.debezium.connector.mysql.Source")
            .field(FIELD_SOURCE_DATABASE, Schema.OPTIONAL_STRING_SCHEMA)
            .field(FIELD_SOURCE_TABLE, Schema.OPTIONAL_STRING_SCHEMA)
            .build();

    @Override
    public Serializer<SourceRecord> serializer() {
        return serializer;
    }

    @Override
    public Serde<SourceRecord> serde() {
        return serde;
    }

    @Override
    public Validator<SourceRecord> validator() {
        return validator;
    }
}

这个示例代码实现了 `io.debezium.converters.CloudEventsConverter` 接口，并实现了 `serializer()`、`serde()` 和 `validator()` 方法。在 `serializer()` 方法中，将 `SourceRecord` 对象序列化为 CloudEvents 格式的字节数组；在 `serde()` 方法中，使用 Debezium 提供的 `payloadJson()` 方法创建一个 JSON 格式的 `Serde` 对象；在 `validator()` 方法中，始终返回 `true`。最后，定义了 CloudEvents 中的 `payload` 和 `source` 的 `Schema` 对象。

需要注意的是，这个示例代码是针对 MariaDB 的转换器，如果需要针对其他数据库实现自定义转换器，需要根据具体情况进行相应的修改。