Kafka JDBC Postgres Sink Connector配置错误排查求助

错误日志 #

org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
    at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:611)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:333)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:234)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:203)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:244)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.kafka.connect.errors.ConnectException: Sink connector 'sink-postgres-json-distributed' is configured with 'delete.enabled=false' and 'pk.mode=none' and therefore requires records with a non-null Struct value and non-null Struct schema, but found record at (topic='Json_test',partition=0,offset=0,timestamp=1695833816061) with a HashMap value and null value schema.

核心问题分析 #

当前配置存在两个关键矛盾：

1. delete.enabled=false + pk.mode=none的组合要求Kafka消息必须携带带Schema的Struct类型值，但你设置了value.converter.schemas.enable=false，导致消息无Schema，仅为HashMap类型
2. insert.mode=upsert与pk.mode=none冲突，upsert逻辑必须依赖主键判断是插入还是更新，不支持无主键模式

解决方法 #

方案一：调整连接器配置（推荐） #

修改配置文件，解决冲突并满足Schema要求：

# 启用值转换器的Schema支持
value.converter.schemas.enable=true
# 替换pk.mode=none，根据实际场景选择主键模式
# 场景1：用消息value中的字段作为主键（示例用name字段）
pk.mode=record_value
pk.fields=name
# 场景2：用Kafka元数据+消息key作为主键
pk.mode=record_key,topic,partition,offset
# 保留upsert模式，确保与主键配置匹配
insert.mode=upsert

方案二：修改源端Serializer生成带Schema的消息 #

由于你用POJO生成源值Serializer，可替换为支持Schema输出的序列化器，比如io.confluent.connect.json.JsonSchemaSerializer，确保消息携带完整的Schema信息，适配连接器对Struct类型值的要求。

方案三：改用消息key作为主键 #

如果示例中的消息key（如1981）可作为Postgres表的主键，调整配置：

pk.mode=record_key
pk.fields=id  # 对应Postgres表的主键字段名
# 保持key转换器配置不变
key.converter=org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable=false

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