Flink 1.12 SQL消费Confluent Kafka Avro数据反序列化失败如何解决

你遇到的配置冲突是Flink 1.12版本Kafka连接器与avro-confluent格式的参数传递逻辑差异导致的：

• 带properties.前缀的参数只会传递给Kafka消费者客户端，不会被独立的Confluent Schema Registry客户端读取，因此访问Schema Registry时缺少认证信息返回401。
• 直接去掉properties.前缀放在WITH顶层会报错，是因为Flink 1.12原生的Kafka连接器顶层参数没有收录这两个Schema Registry认证配置，格式层面也不支持直接在表配置中单独指定该类参数。
• DataStream API可以正常运行是因为你在代码中手动构造Schema Registry客户端时可以直接注入认证参数，不受SQL层参数校验逻辑限制。

方案1：全局配置Schema Registry认证参数（改造成本最低） #

将认证参数配置为Flink作业的全局动态参数，提交作业时通过-D参数传入，或者提前写入集群的flink-conf.yaml配置文件：

-D avro-confluent.basic-auth.credentials-source=你的配置值
-D avro-confluent.basic-auth.user-info=你的用户名:密码

建表SQL中删除原来带properties.前缀的两个avro认证参数即可，其余配置保持不变。

方案2：升级适配版本的连接器 #

替换你当前使用的CDP打包的Flink Kafka连接器与avro-confluent依赖，使用官方适配Flink 1.12的高版本连接器依赖，此时可以直接在WITH参数中加value.前缀配置认证参数：

'value.avro-confluent.basic-auth.credentials-source' = '你的配置值',
'value.avro-confluent.basic-auth.user-info' = '你的用户名:密码'

无需全局配置，仅对当前表生效。

方案3：自定义反序列化格式 #

如果前两个方案受集群环境限制无法落地，可以手动实现自定义的avro-confluent反序列化格式，在代码中硬编码或者透传认证参数，打包后引入Flink SQL作业使用。

优先选择方案1验证：提交作业时加上上述两个动态参数，删除建表SQL中对应的两个带properties.前缀的参数，重新执行即可正常消费数据。

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