如何在Kafka Connect场景下禁用Azure Schema Registry自动注册模式？

可以通过两种方式实现禁用Azure Schema Registry AvroConverter的自动模式注册行为，以下是具体方案：

方法一：使用auto.register.schemas配置项 #

你可以在连接器配置中添加value.converter.auto.register.schemas=false来直接禁用自动注册。该配置项会传递给AvroConverter内部的序列化器，阻止其向Azure Schema Registry自动提交新模式。

更新后的连接器配置示例：

{
  "name": "jdbc-postgresql-avro-connector-azure",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "value.converter": "com.microsoft.azure.schemaregistry.kafka.avro.AvroConverter",
    "value.converter.schema.registry.url": "{schemaRegistryUrl}",
    "value.converter.schema.group": "postgreskafkaschemagroup",
    "value.converter.tenant.id": "{tenantId}",
    "value.converter.client.id": "{clientId}",
    "value.converter.client.secret": "{clientSecret}",
    // 新增禁用自动注册的配置
    "value.converter.auto.register.schemas": "false",
    "connection.url": "{postgresConnectionUrl}",
    "connection.user": "{user}",
    "connection.password": "{password}",
    "connection.attempts": 3,
    "mode": "incrementing",
    "query": "SELECT * FROM users",
    "table.types": "TABLE",
    "topic.prefix": "users",
    "incrementing.column.name": "id"
  }
}

注意：该配置项仅在Azure Schema Registry Kafka插件的1.0.0及以上版本中支持，若你使用的是旧版本，请先升级插件。

方法二：指定预先注册的模式信息 #

如果方法一不生效，你可以预先在Azure Schema Registry中注册好所需的Avro模式，然后通过配置强制转换器直接使用已有的模式：

• 指定模式ID：添加value.converter.schema.id配置，值为你已注册的模式ID
• 指定模式全名：添加value.converter.schema.full.name配置，值为模式的完整名称（格式通常为{命名空间}.{模式名}）

示例配置（以指定模式ID为例）：

{
  "name": "jdbc-postgresql-avro-connector-azure",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "value.converter": "com.microsoft.azure.schemaregistry.kafka.avro.AvroConverter",
    "value.converter.schema.registry.url": "{schemaRegistryUrl}",
    "value.converter.schema.group": "postgreskafkaschemagroup",
    "value.converter.tenant.id": "{tenantId}",
    "value.converter.client.id": "{clientId}",
    "value.converter.client.secret": "{clientSecret}",
    // 指定已注册的模式ID
    "value.converter.schema.id": "your-pre-registered-schema-uuid",
    "connection.url": "{postgresConnectionUrl}",
    "connection.user": "{user}",
    "connection.password": "{password}",
    "connection.attempts": 3,
    "mode": "incrementing",
    "query": "SELECT * FROM users",
    "table.types": "TABLE",
    "topic.prefix": "users",
    "incrementing.column.name": "id"
  }
}

关键注意事项 #

• 无论使用哪种方法，都必须确保目标Avro模式已经在Azure Schema Registry中完成注册，否则连接器会抛出模式不存在的异常。
• 若使用JDBC源连接器，需要确保连接器生成的Avro模式与你预先注册的模式完全匹配（包括字段名称、类型、顺序等），否则会出现序列化失败的问题。

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