首次配置Confluent Kafka JDBC Sink Connector至Oracle异常排查请求

Hey there, let's break down your problem step by step—since your connector is running, offsets are increasing, but you can't see the table or data, there are a few key areas to check:

First: Confirm the Table Actually Exists (It Probably Does!) #

Your logs clearly state 使用Oracle方言，表"member_cbdt_sink1"已存在, so the table is being created. The most likely reason you can't see it is an Oracle schema/permission quirk:

• When you query for the table, are you using the same connection.user account that the connector uses? Oracle creates tables in the user's default schema by default. If you're logged in as a different user, you need to specify the schema: SELECT * FROM YOUR_CONNECT_USER.member_cbdt_sink1;
• Log directly into Oracle with the connector's user and run:

  SELECT table_name FROM user_tables WHERE table_name = 'MEMBER_CBDT_SINK1';
  

  (Oracle defaults to uppercase table names unless you explicitly use quotes, so match that case.)
• Double-check that the connector's user has CREATE TABLE permission—even though the log says the table exists, a missing privilege might have created it in an unexpected schema.

Next: Figure Out Why Data Isn't Loading #

Offsets are going up, so the connector is consuming messages—but they're not making it to the table. Here's what to check:

1. Avro Schema & Table Column Mismatch #

You're using AvroConverter, so make sure your produced messages match the table structure the connector created:

• The log shows your table has columns like first_name(CLOB), height(BINARY_FLOAT), and all non-nullable except automated_email. Verify your Avro schema has exactly these fields with compatible types. If a required field is missing or has the wrong type, the connector might silently drop messages (or fail without obvious errors).
• Turn on DEBUG logging for the JDBC connector: set log4j.logger.io.confluent.connect.jdbc=DEBUG in your Connect worker's log config. This will show you detailed field mapping and any write failures you're missing.

2. Oracle Transaction Visibility #

Oracle uses transaction isolation levels that might prevent you from seeing uncommitted data. Try:

• Logging into Oracle with the connector's user and running COMMIT; manually—sometimes connectors hold transactions longer than expected, and this will flush any pending writes.
• Check if your connector is configured for auto-commit (it should be by default, but it's worth confirming).

3. Missing Write Mode Configuration #

Your current config doesn't set insert.mode, which defaults to insert. If there's any kind of duplicate key issue (even without a primary key) or schema mismatch, writes could fail silently. Add this to your config to be explicit:

"insert.mode": "insert"

If you expect updates later, you can use upsert once you define a primary key.

4. Strict Column Constraints #

All columns except automated_email are marked as non-nullable in the table. If your Avro messages have null values for any of these required fields, the write will fail—but the error might be buried in the logs. Enable error logging to catch this:

"errors.log.enable": "true",
"errors.log.include.messages": "true"

Recommended Config Tweaks to Fix & Prevent Issues #

Update your connector config with these additions to get better visibility and handle edge cases:

{
  "name": "ora_sink_task",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "connection.url": "jdbc:oracle:thin:@host:port/servicename",
    "connection.user": "user",
    "connection.password": "password",
    "topics": "connecttest",
    "tasks.max": "1",
    "table.name.format": "member_cbdt_sink1",
    "value.converter":"io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url":"http://localhost:8081",
    "auto.create": "true",
    // New additions
    "insert.mode": "insert",
    "auto.evolve": "true", // Auto-update table if Avro schema changes
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "errors.deadletterqueue.topic.name": "dlq-ora-sink", // Send failed messages to a dead-letter queue
    "errors.deadletterqueue.topic.replication.factor": "1"
  }
}

Quick Validation Steps #

1. Confirm the table exists via the connector's Oracle user (run the user_tables query above).
2. Enable DEBUG logging to see exactly what's happening during writes.
3. Check the dead-letter queue (if you added it) for any failed messages—this will tell you exactly which records are failing and why.
4. Use kafka-avro-console-consumer to inspect the messages in connecttest and verify their schema matches the table's columns.

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