Kafka Connect访问Schema Registry的Basic Authentication配置问题

你遇到的报错：io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Unauthorized; error code: 401 本质是连接器请求Schema Registry时未携带有效身份凭证导致，两种配置方式均支持，适用场景不同：

1. 全局配置（写入connect-distributed.properties文件） #

• 适用场景：所有连接器使用同一套Schema Registry访问凭证，配置后对所有连接器默认生效
• 需要新增的配置项如下：

# 开启Schema Registry基础认证
key.converter.schema.registry.basic.auth.credentials.source=USER_INFO
key.converter.schema.registry.basic.auth.user.info=你的用户名:你的密码
value.converter.schema.registry.basic.auth.credentials.source=USER_INFO
value.converter.schema.registry.basic.auth.user.info=你的用户名:你的密码
# 若使用了HeaderConverter也需要补充对应配置
header.converter.schema.registry.basic.auth.credentials.source=USER_INFO
header.converter.schema.registry.basic.auth.user.info=你的用户名:你的密码

• 注意：修改配置后需要重启Kafka Connect集群才能生效

2. 单连接器配置（写入连接器配置参数） #

• 适用场景：不同连接器需要使用不同的Schema Registry凭证，或不想重启Connect集群的场景
• 直接在连接器的配置中添加上述相同参数即可，优先级高于全局配置，会覆盖全局默认值
• 示例配置片段：

{
  "name": "自定义连接器名称",
  "config": {
    "connector.class": "对应连接器类名",
    "topics": "关联Topic名称",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://Schema Registry服务地址:8081",
    "key.converter.schema.registry.basic.auth.credentials.source": "USER_INFO",
    "key.converter.schema.registry.basic.auth.user.info": "你的用户名:你的密码",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://Schema Registry服务地址:8081",
    "value.converter.schema.registry.basic.auth.credentials.source": "USER_INFO",
    "value.converter.schema.registry.basic.auth.user.info": "你的用户名:你的密码"
    // 其余连接器配置项
  }
}

常见疑问说明 #

• 为什么部分连接器正常、部分报错：正常运行的连接器已经在单连接器配置中显式传入了凭证，报错的连接器未配置该参数，同时全局也未配置默认凭证
• 为什么curl请求正常：curl请求时你手动传入了认证信息，和连接器的配置逻辑互相独立，没有关联

内容的提问来源于stack exchange，提问作者Bünyamin Şentürk