Case sensitive error when deleting records

[Oanh Nguyễn]

Hi Debezium team,

We use Oracle Source Connector and JDBC sink connector debezium 2.6 to capture change from Oracle Database 19c to PGDB 15.16.

We got this error when sink deleted events to PostgreSQL eventhough insert or update events are successful. (With or without tombstones). 

Caused by: java.lang.NullPointerException
at io.debezium.connector.jdbc.dialect.postgres.PostgresDatabaseDialect.getQueryBindingWithValueCast(PostgresDatabaseDialect.java:123)
at io.debezium.connector.jdbc.type.AbstractType.getQueryBinding(AbstractType.java:37)

Please help us resolve this issue.
------------------------------------------------------------------
Here is structure of table in oracle:
CREATE TABLE CORE.BS_PRODUCT
(
  PRODUCTNO           VARCHAR2(3 CHAR)          NOT NULL,
  PRONAME             VARCHAR2(20 CHAR)         NOT NULL,
  CURR_CODE           VARCHAR2(3 CHAR)          NOT NULL,
  ACCRUAL_FLAG        VARCHAR2(1 CHAR),
  DAYS_DATE           VARCHAR2(3 CHAR),
  DAYS_YEAR           VARCHAR2(3 CHAR),
  PRODUCT_TYPE        VARCHAR2(3 CHAR),
  PRODUCT_MESS_GROUP  VARCHAR2(3 CHAR),
  UNIT                NUMBER,
  UNIT_DESC           VARCHAR2(10 CHAR),
  TYPE_YIELD_PRICE    VARCHAR2(3 CHAR),
  TYPE_INTEREST       VARCHAR2(3 CHAR),
  TYPE_COST1          VARCHAR2(3 CHAR),
  TYPE_COST2          VARCHAR2(3 CHAR),
  TYPE_COST3          VARCHAR2(3 CHAR),
  NETCOST_INTEREST    VARCHAR2(1 CHAR),
  NETCOST_COST1       VARCHAR2(1 CHAR),
  NETCOST_COST2       VARCHAR2(1 CHAR),
  NETCOST_COST3       VARCHAR2(1 CHAR),
  CUSTODIAN_NO        VARCHAR2(12 CHAR),
  TRADING_TYPE        VARCHAR2(1 CHAR),
  DEALER_VERIFY       VARCHAR2(1 CHAR),
  SETTLE_VERIFY       VARCHAR2(1 CHAR),
  ISSUANCE            VARCHAR2(1 CHAR),
  DP_VOSTRO_ACCT      VARCHAR2(20 CHAR)
  SUPPLEMENTAL LOG DATA (ALL) COLUMNS
)
CREATE UNIQUE INDEX CORE.BPR_PK ON CORE.BS_PRODUCT
(PRODUCTNO)
------------------------------------------------------------------
Here is structure of table in postgres:
CREATE TABLE IF NOT EXISTS customer.bs_product
(
    productno character varying(3) COLLATE pg_catalog."default" NOT NULL,
    proname character varying(20) COLLATE pg_catalog."default",
    curr_code character varying(3) COLLATE pg_catalog."default",
    accrual_flag character varying(1) COLLATE pg_catalog."default",
    days_date character varying(3) COLLATE pg_catalog."default",
    days_year character varying(3) COLLATE pg_catalog."default",
    product_type character varying(3) COLLATE pg_catalog."default",
    product_mess_group character varying(3) COLLATE pg_catalog."default",
    unit numeric,
    unit_desc character varying(10) COLLATE pg_catalog."default",
    type_yield_price character varying(3) COLLATE pg_catalog."default",
    type_interest character varying(3) COLLATE pg_catalog."default",
    type_cost1 character varying(3) COLLATE pg_catalog."default",
    type_cost2 character varying(3) COLLATE pg_catalog."default",
    type_cost3 character varying(3) COLLATE pg_catalog."default",
    netcost_interest character varying(1) COLLATE pg_catalog."default",
    netcost_cost1 character varying(1) COLLATE pg_catalog."default",
    netcost_cost2 character varying(1) COLLATE pg_catalog."default",
    netcost_cost3 character varying(1) COLLATE pg_catalog."default",
    custodian_no character varying(12) COLLATE pg_catalog."default",
    trading_type character varying(1) COLLATE pg_catalog."default",
    dealer_verify character varying(1) COLLATE pg_catalog."default",
    settle_verify character varying(1) COLLATE pg_catalog."default",
    issuance character varying(1) COLLATE pg_catalog."default",
    dp_vostro_acct character varying(20) COLLATE pg_catalog."default",
    CONSTRAINT bs_product_pkey PRIMARY KEY (productno)
)
------------------------------------------------------------------
Here is source and sink connector:
{
"name": "core-test9-BS_PRODUCT-debezium-oracle-connector",
"connector.class": "io.debezium.connector.oracle.OracleConnector",
"snapshot.mode.configuration.based.snapshot.data": "true",
"snapshot.mode": "initial",
"database.user": "abc",
"database.password": "xxx",
"database.dbname": "coren4",
"database.url": "jdbc:oracle:thin:@//xx.xx.xx.xx:1521/coren4",
"auto.create.topics.enable": "true",
"topic.creation.default.partitions": "3",
"topic.creation.default.replication.factor": "3",
"tasks.max": "1",
"schema.include.list": "core",
"schema.history.internal.kafka.bootstrap.servers": "kafka.streaming-kafka-onprem.svc.cluster.local:9092",
"schema.history.internal.store.only.captured.tables.ddl": "true",
"schema.history.internal.store.only.captured.databases.ddl": "true",
"schema.history.internal.kafka.topic": "schema-changes-test9-core.inventory",
"topic.prefix": "core-test9",
"table.include.list": "core.bs_product",
"log.mining.strategy": "redo_log_catalog",
"log.mining.transaction.retention.ms": "0",
"log.mining.batch.size.default": "200000",
"log.mining.batch.size.min": "10000",
"log.mining.batch.size.max": "10000000",
"log.mining.scn.gap.detection.gap.size.min": "10000000",
"log.mining.scn.gap.detection.time.interval.max.ms": "20000",
"heartbeat.interval.ms": "1000",
"lob.enabled": "true",
"tombstones.on.delete": "true",
"errors.max.retries": "-1",
"snapshot.database.errors.max.retries": "5"
}
{
"name": "jdbc-postgres-core-test9-BS_PRODUCT-sink-connector",
"connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
"key.converter": "io.confluent.connect.avro.AvroConverter",
"key.converter.schema.registry.url": "http://schema-registry-cp-schema-registry:8081",
"tasks.max": "1",
"connection.url": "jdbc:postgresql://xx.xx.xx.xx:5432/customer",
"connection.username": "abc",
"connection.password": "xx",
"insert.mode": "upsert",
"delete.enabled": "true",
"auto.create": "true",
"primary.key.mode": "record_key",
"quote.identifiers": "false",
"schema.evolution": "basic",
"database.time_zone": "UTC",
"topics": "core-test9.CORE.BS_PRODUCT",
"table.name.format": "customer.bs_product"
}

[jiri.p...@gmail.com]

Hi,

can you please take a look at https://debezium.io/documentation/reference/2.7/connectors/jdbc.html#jdbc-quoting-case-sensitivity ?

Jiri

[jiri.p...@gmail.com]

Did you also have a chance to try Debezium 2.7?

Jiri

[Oanh Nguyễn]

Hi Jiri,

I have try 2.7 also. Here what I want to do which CDC: Capture data change from oracle table (Which is upper-case table and column schema) to postgres (Which I want to pre-defined table and column name in lower-case)

Vào lúc 12:50:11 UTC+7 ngày Thứ Ba, 2 tháng 7, 2024, jiri.p...@gmail.com đã viết:

[Chris Cranford]

Hi -

Please implement a ColumnNamingStrategy and specify the fully qualified class name using `column.naming.strategy`[1].  In this case, the implementation of the ColumnNamingStrategy interface would simply lower case the field names to match your desired needs for PostgreSQL.  This will not only allow you to avoid the use of quoted.identifiers, but it will guarantee that the JDBC sink code will find and use the columns correctly.

Chris

[1]: https://debezium.io/documentation/reference/2.7/connectors/jdbc.html#jdbc-property-column-naming-strategy

--
You received this message because you are subscribed to the Google Groups "debezium" group.
To unsubscribe from this group and stop receiving emails from it, send an email to debezium+u...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/54a79b98-a3db-46f2-a6cc-c5b18233ec2en%40googlegroups.com.

[Oanh Nguyễn]

Hi Chris,

Thank you for your guidance, I have fixed the error of not being able to delete.

Besides that, I noticed a performance problem from the beginning of Oracle DB exporting data to Kafka. When inserting data into the Oracle database occurs, it takes 3 to 5 minutes for Kafka to receive the message.

If you have any experience improving Oracle DB input performance, please share it with me.

Vào lúc 19:39:12 UTC+7 ngày Thứ Ba, 2 tháng 7, 2024, Chris Cranford đã viết: