更新Kafka Connector触发Schema兼容性异常问题排查

更新Kafka Connect 2.0.0连接器配置时触发以下报错：

io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException:
Schema being registered is incompatible with an earlier schema for
subject "value", details: [{errorType:'TYPE_MISMATCH',
description:'The type (path '/') of a field in the new schema does not
match with the old schema', additionalInfo:'reader type: STRING not
compatible with writer type: NULL'}, error code: 409; error code: 409

未修改SQL查询语句，但comment字段的Schema发生了变更：
旧Schema：

{
  "name": "comment",
  "type": [
    "null",
    "string"
  ],
  "default": null
}

新Schema：

{
  "name": "comment",
  "type": "string"
}

该字段在PostgreSQL中为jsonb类型，求解决方法。

1. 排查连接器Schema自动推断配置 #

• 检查连接器是否开启了auto.register.schemas或use.latest.version参数，这类参数会在连接器更新/重启时重新推断Schema，可能因数据分布变化改变字段的可空性判断。
• 确认jsonb.field.as.string这类针对PostgreSQL jsonb字段的转换器参数是否被修改，该参数会影响jsonb字段被解析为string类型的逻辑，进而影响null值的推断。
• 检查schema.ignore或exclude.columns配置，确保没有误操作导致字段的Schema推断逻辑异常。

2. 手动注册兼容Schema并禁用自动注册 #

如果新Schema的变更不符合预期，先手动把旧的兼容Schema注册到Schema Registry：

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{
    "schema": "{\"name\": \"comment\", \"type\": [\"null\", \"string\"], \"default\": null}"
  }' \
  http://<schema-registry-host>:<port>/subjects/value/versions

然后更新连接器配置，禁用自动Schema注册并指定使用这个Schema：

auto.register.schemas=false
schema.id=<你刚注册的Schema版本ID>

3. 调整Schema Registry兼容性级别 #

如果业务允许，可以修改Schema Registry的兼容性设置：

• 若要允许旧Schema读取新数据，设置全局或指定主题的兼容性为BACKWARD：

# 全局设置
curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility": "BACKWARD"}' \
  http://<schema-registry-host>:<port>/config

# 仅针对value主题设置
curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility": "BACKWARD"}' \
  http://<schema-registry-host>:<port>/config/value

• 若暂时不需要兼容性检查（需承担数据解析风险），可以设置为NONE：

curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility": "NONE"}' \
  http://<schema-registry-host>:<port>/config/value

4. 验证PostgreSQL数据的null值分布 #

虽然没改查询语句，但可能comment字段的实际数据已经没有null值了，导致连接器推断出非空Schema：
用SQL查询验证：

SELECT COUNT(*) FROM your_table WHERE comment IS NULL;

如果确实没有null值且业务允许，可接受新Schema，调整兼容性设置即可通过注册。

5. 强制指定字段Schema避免自动推断 #

在连接器配置中手动指定comment字段的Schema结构，覆盖自动推断：

transforms=forceCommentNullable
transforms.forceCommentNullable.type=org.apache.kafka.connect.transforms.SetSchemaMetadata$Value
transforms.forceCommentNullable.schema={\"name\": \"comment\", \"type\": [\"null\", \"string\"], \"default\": null}

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