Kafka JDBC Sink Connector写入Oracle报错：目标表缺失且自动创建禁用

问题场景 #

尝试将Kafka主题kafka2.INVENTORY.PRODUCT的数据通过JDBC Sink Connector写入Oracle数据库tinventory模式下的product_target表，使用用户jdbcsinkuser连接，执行时触发如下错误：

ERROR [sink-to-target|task-0] WorkerSinkTask{id=sink-to-target-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted. Error: 表"tinventory"."PRODUCT_TARGET"不存在且自动创建已禁用 (org.apache.kafka.connect.runtime.WorkerSinkTask:609)
org.apache.kafka.connect.errors.ConnectException: 表"tinventory"."PRODUCT_TARGET"不存在且自动创建已禁用

当前连接器配置：

{
    "name": "sink-to-target",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "tasks.max": "1",
        "topics": "kafka2.INVENTORY.PRODUCT",
        "table.name.format": "PRODUCT_TARGET",
        "connection.url": "jdbc:oracle:thin:@localhost:1521/orclpdb1",
        "connection.user": "jdbcsinkuser",
        "connection.password": "jdbc",
        "transforms": "unwrap,InsertField",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "pk.fields": "PRODUCTID",
        "insert.mode":"upsert",
        "delete.enabled":true,
        "delete.retention.ms":100,
        "pk.mode": "record_key",
        "transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
        "transforms.InsertField.static.field": "SOURCE",
        "transforms.InsertField.static.value": "LEGACY-SYS"
    }
}

问题原因 #

1. 目标表tinventory.PRODUCT_TARGET实际不存在于Oracle数据库中
2. 连接器配置未开启自动建表功能，且table.name.format未指定完整的模式+表名，导致连接器无法定位目标表

解决方法 #

方法一：手动创建目标表 #

在Oracle中执行建表语句，确保表结构匹配Kafka消息字段，包含PRODUCTID主键和SOURCE字段，示例：

CREATE TABLE tinventory.PRODUCT_TARGET (
    PRODUCTID NUMBER PRIMARY KEY, -- 根据实际数据类型调整
    -- 补充其他消息字段，例如：
    PRODUCT_NAME VARCHAR2(100),
    PRICE NUMBER(10,2),
    SOURCE VARCHAR2(50)
);

同时确保jdbcsinkuser拥有该表的INSERT、UPDATE、DELETE权限。

方法二：开启连接器自动建表 #

修改连接器配置，添加两个关键参数：

• auto.create=true：允许连接器自动创建表
• table.name.format=tinventory.PRODUCT_TARGET：指定包含模式的完整表名

修改后的配置片段：

{
    "name": "sink-to-target",
    "config": {
        // ... 保留原有配置 ...
        "table.name.format": "tinventory.PRODUCT_TARGET",
        "auto.create": "true",
        // ... 保留原有配置 ...
    }
}

注意：此方法需要jdbcsinkuser拥有Oracle的CREATE TABLE权限，且自动建表的字段类型可能需要后续调整以匹配业务需求。

额外检查项 #

• 确认tinventory模式在Oracle中存在，且jdbcsinkuser有权限访问该模式
• 若使用upsert模式，必须保证目标表的主键与配置中pk.fields的PRODUCTID一致

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