Kafka Connect对接Google BigQuery时报schema必须为struct错误的解决方法

前置操作说明 #

1. 已创建的测试主题testmd #

创建命令执行结果：

$KAFKA_HOME/bin/kafka-topics.sh --create --topic testmd --replication-factor 3 --partitions 3 --zookeeper rhes75:2181,rhes564:2181,rhes76:2181
Created topic testmd.

2. Kafka Connect独立模式核心配置 #

connect-standalone.properties文件核心内容：

bootstrap.servers=rhes75:9092,rhes75:9093,rhes75:9094,rhes564:9092,rhes564:9093,rhes564:9094,rhes76:9092,rhes76:9093,rhes76:9094
key.converter=org.apache.kafka.connect.storage.StringConverter
#key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true
offset.storage.file.filename=/tmp/connect_bq.offsets
offset.flush.interval.ms=10000
plugin.path=/d4T/hduser/bigquery-kafka-connect-sink/share/kafka/plugins

3. BigQuery Sink连接器配置 #

连接器属性文件核心内容：

name=bigquery-sink
connector.class=com.wepay.kafka.connect.bigquery.BigQuerySinkConnector
defaultDataset=test
project=project_name
topics=testmd
autoCreateTables=false
gcsBucketName=tmp_storage_bucket
keySource=FILE
keyfile=xyz.json
schemaRetriever=com.wepay.kafka.connect.bigquery.retrieve.IdentitySchemaRetriever
threadPoolSize=10
enableBatchLoad=false
autoCreateBucket=true

4. 主题消息验证 #

通过控制台消费验证testmd主题的消息内容（使用StringDeserializer）：

$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $bootstrapservers --from-beginning --topic testmd --property print.key=true
4b1201a3-a12c-429a-84ab-a56625c42410     {"schema": { "type": "struct", "fields": [ { "field": "rowkey", "type": "string", "optional": true}],"optional": false,"name": "test.md"}, "payload": {"rowkey": "4b1201a3-a12c-429a-84ab-a56625c42410"}}

异常现象 #

• 启动连接器后持续抛出解析异常：

  com.wepay.kafka.connect.bigquery.exception.ConversionConnectException: Top-level Kafka Connect schema must be of type 'struct'
  

• 查看消费者组状态时发现，connect-bigquery-sink消费者组无活跃成员，无对应topic的offset提交记录

  $KAFKA_HOME/bin/kafka-consumer-groups.sh --describe --all-groups
  Consumer group 'connect-bigquery-sink' has no active members.
  GROUP                  TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID          HOST            CLIENT-ID
  console-consumer-24314 testmd          0          -               0               -               consumer-xxx         /50.140.197.220 consumer-xxx
  console-consumer-24314 testmd          1          -               1               -               consumer-xxx         /50.140.197.220 consumer-xxx
  console-consumer-24314 testmd          2          -               0               -               consumer-xxx         /50.140.197.220 consumer-xxx
  

根因与解决方案 #

1. 根因定位 #

• 报错触发逻辑：BigQuery Sink要求传入的消息value顶层Schema必须为Struct类型，解析过程中发现Schema类型不匹配，抛出异常
• 消费者组无活跃成员的本质：连接器任务在启动初始化阶段就因解析异常被Connect worker终止，未成功加入消费者组发起拉取请求
• 核心认知偏差：控制台消费看到的带schema/payload的消息，是经StringDeserializer解码后的字符串文本，并非Connect JsonConverter可直接解析的标准序列化结构

2. 具体解决步骤 #

（1）校验生产端消息格式合规性

• 常见格式问题：生产端直接将带schema/payload的JSON字符串以String类型写入topic，形成双重转义结构，导致Connect JsonConverter解析时识别到的value是String类型而非带Schema的Struct
• 验证方法：直接查看消息的字节内容，确认不存在外层字符串包裹、内层转义JSON的问题
• 适配方案：
  • 生产端统一使用和Connect侧一致的org.apache.kafka.connect.json.JsonConverter（开启schemas.enable=true）序列化消息，确保写入的是可直接被JSON解析的标准{"schema":..., "payload":...}结构
  • 若生产端无法改造为标准序列化格式，将Connect全局的value.converter临时调整为StringConverter，搭配可解析JSON字符串的自定义转换器，将字符串转为Struct类型后再传入BigQuery Sink

（2）校验Connect配置生效状态

• 检查connect-standalone.properties的配置加载路径，确认无文件路径错误、参数拼写问题：
  • 确认value.converter=org.apache.kafka.connect.json.JsonConverter、value.converter.schemas.enable=true参数已被worker正确读取
  • 检查连接器专属配置文件中是否存在覆盖全局converter配置的参数，导致实际解析逻辑和预期不符
• 验证方法：启动Connect worker时输出调试日志，确认启动阶段加载的converter参数值；临时启动测试连接器直接消费testmd主题，打印解析后Record的value Schema类型，确认解析结果

（3）修正消费者组状态查询命令

• 此前执行的消费者组查询命令存在输入缺失，--bootstrap-server参数漏写，导致查询结果不完整
• 正确的查询命令格式为：

  $KAFKA_HOME/bin/kafka-consumer-groups.sh --bootstrap-server rhes75:9092,rhes75:9093,rhes75:9094,rhes564:9092,rhes564:9093,rhes564:9094,rhes76:9092,rhes76:9093,rhes76:9094 --describe --group connect-bigquery-sink
  

  修正命令后可获取连接器任务的实际offset提交记录、报错详情，精准确认异常触发的环节

（4）调整BigQuery Sink的Schema兼容规则

• 当前配置中IdentitySchemaRetriever、allBQFieldsNullable=false参数对消息Schema的校验规则较严格，非Struct类型的消息会直接触发异常
• 临时适配方案：
  • 开启allowNewBigQueryFields=true、allowBigQueryRequiredFieldRelaxation=true参数，放宽Schema一致性校验要求
  • 清理testmd主题中不符合Struct格式的历史测试消息，避免旧消息触发解析异常

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