如何在Connector级别设置value.converter.ignore.default.for.nullables属性？

我需要使用配置属性：

value.converter.ignore.default.for.nullables=true

将其设置为环境变量CONNECT_VALUE_CONVERTER_IGNORE_DEFAULT_FOR_NULLABLES: true时可以正常工作，但在Debezium MySQL 1.9.7.Final连接器配置中添加该属性却无效，请问是操作有误还是该属性无法在Connector级别设置？

原连接器配置：

{
    "name": "source-mysql-true-local",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "taxify-binlog-db",
        "database.port": "3306",
        "database.user": "root",
        "database.password": "",
        "database.server.id": "184054",
        "database.server.name": "cdc_mysql_avro.local",
        "database.history.kafka.bootstrap.servers": "kafka:29092",
        "database.history.kafka.topic": "cdc_dbhistory_mysql.local",
        "include.schema.changes": "true",
        "snapshot.mode": "schema_only",
        "decimal.handling.mode": "string",
        "database.exclude.list": "infra,test",
        "table.exclude.list": ".*_gho$,.*_ghc$",
        "max.queue.size": "24576",
        "max.batch.size": "6144",
        "sanitize.field.names": "true",
        "value.converter.ignore.default.for.nullables": "true"
    }
}

原因说明 #

这个属性是Kafka Connect Converter组件的配置参数，并非Debezium MySQL连接器本身的参数。

通过环境变量设置时，它作用于整个Connect集群的全局Converter配置，因此生效；但直接在连接器配置中添加该属性时，如果没有同时显式指定对应的Converter类，Connect会使用全局的Converter实例，而全局实例只会读取全局配置（环境变量或connect-distributed.properties）中的参数，不会读取连接器配置里的Converter相关参数。

正确操作方式 #

若要在连接器级别单独设置该属性，需要在连接器配置中显式声明你使用的Converter类，并同时配置对应的参数，示例如下：

{
    "name": "source-mysql-true-local",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "taxify-binlog-db",
        "database.port": "3306",
        "database.user": "root",
        "database.password": "",
        "database.server.id": "184054",
        "database.server.name": "cdc_mysql_avro.local",
        "database.history.kafka.bootstrap.servers": "kafka:29092",
        "database.history.kafka.topic": "cdc_dbhistory_mysql.local",
        "include.schema.changes": "true",
        "snapshot.mode": "schema_only",
        "decimal.handling.mode": "string",
        "database.exclude.list": "infra,test",
        "table.exclude.list": ".*_gho$,.*_ghc$",
        "max.queue.size": "24576",
        "max.batch.size": "6144",
        "sanitize.field.names": "true",
        // 显式指定Converter类（以AvroConverter为例）
        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "value.converter.schema.registry.url": "http://schema-registry:8081",
        // 配置Converter的目标属性
        "value.converter.ignore.default.for.nullables": "true"
    }
}

这样配置后，连接器会使用你指定的Converter实例，并应用该属性配置。

内容的提问来源于stack exchange，提问作者Opal