Oracle JDBC Source Connector数据同步序列化反序列化问题求助

Let's break down each of your issues and walk through actionable fixes:

1. Avro Deserialization Error: Cannot deserialize type int64 as type float64 #

This type mismatch happens when the schema registered for your topic defines a field as int64 (long), but your consumer expects a float64 (double/float). Here's how to resolve it:

Root Cause #

Your connector uses numeric.mapping="best_fit", which tries to map Oracle numeric types to the smallest possible Kafka Connect type. If your abc column in Oracle is a NUMBER type with large precision/scale, it might be mapped to int64 instead of a floating-point type, causing the consumer to throw an error.

Fixes #

• Adjust numeric.mapping: Try setting numeric.mapping="none" to disable automatic type shrinking. This will map Oracle NUMBER types to Kafka Connect's Decimal logical type (stored as bytes in Avro), avoiding integer/float mismatches entirely.
• Explicitly cast in your query: Modify your SELECT statement to force the desired data type, e.g.:

  SELECT CAST(abc AS NUMBER(18,2)) FROM xyz
  

• Validate the registered schema: Cross-check the Schema Registry entry you provided—ensure the abc field (if present) is mapped to a type your consumer expects.

2. JSON Deserialization Error: Invalid UTF-32 character #

This error signals invalid non-UTF-8 data being serialized into JSON, almost always caused by binary Oracle types (like RAW, BLOB, or LONG RAW) being treated as plain text.

Root Cause #

The JDBC Connector is converting binary data directly into UTF-8 strings, which creates invalid character sequences that JSON parsers can't handle.

Fixes #

• Handle binary fields in your query: Convert binary columns to valid strings using Oracle built-in functions, e.g.:

  -- For RAW columns
  SELECT UTL_RAW.CAST_TO_VARCHAR2(abc) AS abc FROM xyz
  
  -- For BLOB columns
  SELECT DBMS_LOB.CONVERTTOCHARACTERSET(abc, 'UTF8') AS abc FROM xyz
  

• Add character encoding to JDBC URL: Ensure your connection string enforces UTF-8 alignment with JSON requirements:

  jdbc:oracle:thin:@<server>:<Port>/<Schema>?useUnicode=true&characterEncoding=UTF-8
  

• Tweak JSON converter config: If using JsonConverter, disable schema inclusion if your consumer doesn't need it (this can avoid unnecessary type conflicts):

  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false"
  

3. KSQL Error: Subject not found #

KSQL can't locate the Avro schema because of a mismatch between the expected subject name and what exists in the Schema Registry.

Root Cause #

KSQL expects Schema Registry subjects to follow the pattern <kafka_topic_name>-value. Your registered subject is RAW-LITERAL_VALUES-value, which means your Kafka topic is likely RAW-LITERAL_VALUES—but KSQL may be looking for a different topic name based on your connector configuration.

Fixes #

• Explicitly specify the topic in KSQL: When creating your stream/table, point it directly to the correct topic name from your Schema Registry:

  CREATE STREAM LITERAL_STREAM 
  WITH (
    KAFKA_TOPIC='RAW-LITERAL_VALUES',
    VALUE_FORMAT='AVRO'
  );
  

• Verify connector topic naming: If you used table.whitelist, confirm the connector's topic.prefix generates the right topic name. For example, a table named XYZ with topic.prefix=LITERAL_VALUES would create a topic LITERAL_VALUES_XYZ, with a corresponding Schema Registry subject LITERAL_VALUES_XYZ-value.
• Check Schema Registry connectivity: Ensure KSQL is configured to point to the exact Schema Registry instance where your RAW-LITERAL_VALUES-value subject exists.

Your Connector Config (For Reference) #

{ 
  "name": "LITERAL_VALUES", 
  "config": { 
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", 
    "key.serializer": "io.confluent.kafka.serializers.KafkaAvroSerializer", 
    "value.serializer": "io.confluent.kafka.serializers.KafkaAvroSerializer", 
    "connection.user": "<user>", 
    "connection.password": "<Password>", 
    "tasks.max": "1", 
    "connection.url": "jdbc:oracle:thin:@<server>:<Port>/<Schema>", 
    "mode": "bulk", 
    "topic.prefix": "LITERAL_VALUES", 
    "batch.max.rows":1000, 
    "numeric.mapping":"best_fit", 
    "query":"SELECT abc from xyz" 
  } 
}

Registered Schema (For Reference) #

{ 
  "subject": "RAW-LITERAL_VALUES-value", 
  "version": 1, 
  "id": 16, 
  "schema": "{\"type\":\"record\",\"name\":\"LITERAL_VALUES\",\"fields\":[{\"name\":\"LITERAL_ID\",\"type\":[\"null\",{\"type\":\"bytes\",\"scale\":127,\"precision\":64,\"connect.version\":1,\"connect.parameters\":{\"scale\":\"127\"},\"connect.name\":\"org.apache.kafka.connect.data.Decimal\",\"logicalType\":\"decimal\"}],\"default\":null},{\"name\":\"LITERAL_NAME\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"LITERAL_VALUE\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"SOURCE_SYSTEM_ID\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"SOURCE_SYSTEM_INSTANCE_ID\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"EFF_STRT_DT\",\"type\":[\"null\",{\"type\":\"long\",\"connect.version\":1,\"connect.name\":\"org.apache.kafka.connect.data.Timestamp\",\"logicalType\":\"timestamp-millis\"}],\"default\":null},{\"name\":\"EFF_END_DT\",\"type\":[\"null\",{\"type\":\"long\",\"connect.version\":1,\"connect.name\":\"org.apache.kafka.connect.data.Timestamp\",\"logicalType\":\"timestamp-millis\"}],\"default\":null},{\"name\":\"STRT_DT\",\"type\":[\"null\",{\"type\":\"long\",\"connect.version\":1,\"connect.name\":\"org.apache.kafka.connect.data.Timestamp\",\"logicalType\":\"timestamp-millis\"}],\"default\":null},{\"name\":\"END_DT\",\"type\":[\"null\",{\"type\":\"long\",\"connect.version\":1,\"connect.name\":\"org.apache.kafka.connect.data.Timestamp\",\"logicalType\":\"timestamp-millis\"}],\"default\":null},{\"name\":\"CRTD_BY\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"CRTD_DT\",\"type\":[\"null\",{\"type\":\"long\",\"connect.version\":1,\"connect.name\":\"org.apache.kafka.connect.data.Timestamp\",\"logicalType\":\"timestamp-millis\"}],\"default\":null},{\"name\":\"LST_UPD_BY\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"LST_UPD_DT\",\"type\":[\"null\",{\"type\":\"long\",\"connect.version\":1,\"connect.name\":\"org.apache.kafka.connect.data.Timestamp\",\"logicalType\":\"timestamp-millis\"}],\"default\":null}],\"connect.name\":\"LITERAL_VALUES\"}" 
}

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