Tabular Iceberg Sink Connector自动建表异常及数据写入问题

核心问题根源 #

两个异常本质是Debezium CDC消息结构未被Iceberg Sink正确解析，叠加配置中的显性错误导致：

1. 冗余无效配置干扰连接器初始化
2. Topic名称带空格导致匹配失败
3. 未指定从CDC消息的after字段提取实际数据，引发空值写入或Schema解析错误

具体修复步骤 #

1. 修正连接器配置错误 #

清理无效项并添加关键参数：

• 删除两个多了一个点的无效配置：iceberg..s3.secret-access-key、iceberg..s3a.path.style.access
• 去掉topics字段开头的空格，改为"topics": "sqls_cdc_new.dbo.Employees"
• 添加iceberg.tables.write-data-from-field: "after"，指定连接器从CDC消息的after字段提取源表数据（Debezium的CDC消息是包含before/after/op等元数据的信封结构，实际业务数据在after中）

修正后的完整配置：

{
  "connector.class": "io.tabular.iceberg.connect.IcebergSinkConnector",
  "consumer.override.sasl.jaas.config": "",
  "iceberg.catalog.catalog-impl": "org.apache.iceberg.nessie.NessieCatalog",
  "iceberg.catalog.client.region": "default",
  "iceberg.catalog.io-impl": "org.apache.iceberg.aws.s3.S3FileIO",
  "iceberg.catalog.ref": "main",
  "iceberg.catalog.s3.access-key-id": "",
  "iceberg.catalog.s3.compat": "true",
  "iceberg.catalog.s3.endpoint": "https://myminio.mydomain.com:21000",
  "iceberg.catalog.s3.path-style-access": "true",
  "iceberg.catalog.s3.secret-access-key": "",
  "iceberg.catalog.uri": "http://mynessie.mydomain.com:19120/api/v1",
  "iceberg.catalog.warehouse": "s3://testbucket",
  "iceberg.fs.defaultFS": "s3://testbucket",
  "iceberg.fs.s3a.impl": "org.apache.fs.s3a.S3AFileSystem",
  "iceberg.tables": "schema.sqlserver_kafka_cdc_test",
  "iceberg.tables.auto-create-enabled": "true",
  "iceberg.tables.evolve-schema-enabled": "true",
  "iceberg.tables.schema-case-insensitive": "true",
  "iceberg.tables.write-data-from-field": "after",
  "key.converter": "com.cloudera.dim.kafka.connect.converts.AvroConverter",
  "key.converter.passthrough.enabled": "false",
  "key.converter.schema.registry.url": "https://schemareg.mydomain.com:7790/api/v1/",
  "tasks.max": "1",
  "topics": "sqls_cdc_new.dbo.Employees",
  "value.converter": "com.cloudera.dim.kafka.connect.converts.AvroConverter",
  "value.converter.passthrough.enabled": "false",
  "value.converter.schema.registry.url": "https://schemareg.mydomain.com:7790/api/v1/",
  "secret.properties": "consumer.override.sasl.jaas.config",
  "name": "sqlserver_kafka_cdc_sink_iceberg_test"
}

2. 手动建表场景适配 #

若选择手动创建Iceberg表：

• 登录Schema Registry查看目标Topic的Avro Schema，确认after字段下的字段名、数据类型
• 确保手动创建的Iceberg表字段与after中的Schema完全匹配（开启schema-case-insensitive后大小写可兼容，但严格匹配更稳妥）
• 必须保留iceberg.tables.write-data-from-field: "after"配置，确保连接器写入实际业务数据

3. 自动建表场景验证 #

启用自动建表后：

• 连接器会基于after字段的Schema自动创建Iceberg表，此时表结构应与源表一致
• 查看Kafka Connect任务日志，确认无Schema解析、权限或存储连接错误
• 插入测试数据后，通过Nessie或Iceberg CLI查询目标表，验证数据写入情况

额外排查点 #

• 检查Kafka Connect任务日志，重点关注字段映射、Schema兼容、存储权限相关报错
• 验证Nessie Catalog与MinIO的连接有效性，确保连接器拥有S3存储的读写权限
• 使用kafka-console-consumer工具消费目标Topic，确认消息的after字段包含有效业务数据

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