Loading...

Details

Type: Bug
Resolution: Done
Priority: Blocker
Fix Version/s: 1.4.0.Alpha2
Affects Version/s: 1.3.0.Final
Component/s: oracle-connector
Labels:
None

Blocked:
False
Ready:
False
Release Note Text:
Undefined
Steps to Reproduce:
Hide

I'm Using Oracle 11g.

Create a partition table in oracle:

CREATE TABLE U_REPORT.CDC_TABLE ( ID NUMBER(20) not null, FULL_DATE DATE, CODE VARCHAR2(6), PRIMARY KEY (ID) ) PARTITION by RANGE(FULL_DATE) ( PARTITION P_2020 VALUES LESS THAN (TO_DATE('2021-01-01 00:00:00','yyyy-mm-dd hh24:mi:ss')) TABLESPACE TS_BI_REPORT_DATA, PARTITION P_2021 VALUES LESS THAN (TO_DATE('2022-01-01 00:00:00','yyyy-mm-dd hh24:mi:ss')) TABLESPACE TS_BI_REPORT_DATA );

I create a test code like below:

import io.debezium.engine.ChangeEvent; import io.debezium.engine.DebeziumEngine; import io.debezium.engine.RecordChangeEvent; import io.debezium.engine.format.Json; import java.io.IOException; import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import io.debezium.engine.format.Json; public class DBZOracleTest { private DebeziumEngine<ChangeEvent<String, String>> engine; public static void main(String[] args) throws IOException, InterruptedException { DBZOracleTest test = new DBZOracleTest(); test.run(); } public void run() throws InterruptedException, IOException { // Create the engine with this configuration ... props.setProperty("name", "engine"); props.setProperty("offset.storage", "org.apache.kafka.connect.storage.MemoryOffsetBackingStore"); props.setProperty("database.out.server.name", "dbzxout"); props.setProperty("database.hostname", "127.0.0.1"); props.setProperty("database.port", "1521"); props.setProperty("database.user", "logminer"); props.setProperty("database.password", "ys1hl9t#"); props.setProperty("database.dbname", "orcl"); props.setProperty("database.schema","U_REPORT"); props.setProperty("connector.class", "io.debezium.connector.oracle.OracleConnector"); props.setProperty("database.server.name", "CDC"); props.setProperty("table.include.list", "U_REPORT.CDC_TABLE"); props.setProperty("database.connection.adapter", "logminer"); props.setProperty("database.database.connection.adapter", "logminer"); props.setProperty("database.history", "io.debezium.relational.history.MemoryDatabaseHistory"); props.setProperty("database.history.file.filename", "/tmp/dbhistory.dat"); engine = DebeziumEngine.create(Json.class).using(props) .using((success, message, error) -> { System.out.println(message); System.out.println(error); }) .notifying(record -> { Object value = ((RecordChangeEvent) record).record(); System.out.println(value); }).build(); ExecutorService executor = Executors.newSingleThreadExecutor(); executor.execute(engine); } }

Run the code and try create/update/delete the "CDC_TABLE" table in database, no record print.
Show
I'm Using Oracle 11g. Create a partition table in oracle: CREATE TABLE U_REPORT.CDC_TABLE ( ID NUMBER(20) not null , FULL_DATE DATE, CODE VARCHAR2(6), PRIMARY KEY (ID) ) PARTITION by RANGE(FULL_DATE) ( PARTITION P_2020 VALUES LESS THAN (TO_DATE( '2021-01-01 00:00:00' , 'yyyy-mm-dd hh24:mi:ss' )) TABLESPACE TS_BI_REPORT_DATA, PARTITION P_2021 VALUES LESS THAN (TO_DATE( '2022-01-01 00:00:00' , 'yyyy-mm-dd hh24:mi:ss' )) TABLESPACE TS_BI_REPORT_DATA ); I create a test code like below: import io.debezium.engine.ChangeEvent; import io.debezium.engine.DebeziumEngine; import io.debezium.engine.RecordChangeEvent; import io.debezium.engine.format.Json; import java.io.IOException; import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import io.debezium.engine.format.Json; public class DBZOracleTest { private DebeziumEngine<ChangeEvent< String , String >> engine; public static void main( String [] args) throws IOException, InterruptedException { DBZOracleTest test = new DBZOracleTest(); test.run(); } public void run() throws InterruptedException, IOException { // Create the engine with this configuration ... props.setProperty( "name" , "engine" ); props.setProperty( "offset.storage" , "org.apache.kafka.connect.storage.MemoryOffsetBackingStore" ); props.setProperty( "database.out.server.name" , "dbzxout" ); props.setProperty( "database.hostname" , "127.0.0.1" ); props.setProperty( "database.port" , "1521" ); props.setProperty( "database.user" , "logminer" ); props.setProperty( "database.password" , "ys1hl9t#" ); props.setProperty( "database.dbname" , "orcl" ); props.setProperty( "database.schema" , "U_REPORT" ); props.setProperty( "connector.class" , "io.debezium.connector.oracle.OracleConnector" ); props.setProperty( "database.server.name" , "CDC" ); props.setProperty( "table.include.list" , "U_REPORT.CDC_TABLE" ); props.setProperty( "database.connection.adapter" , "logminer" ); props.setProperty( "database.database.connection.adapter" , "logminer" ); props.setProperty( "database.history" , "io.debezium.relational.history.MemoryDatabaseHistory" ); props.setProperty( "database.history.file.filename" , "/tmp/dbhistory.dat" ); engine = DebeziumEngine.create(Json.class).using(props) .using((success, message, error) -> { System .out.println(message); System .out.println(error); }) .notifying(record -> { Object value = ((RecordChangeEvent) record).record(); System .out.println(value); }).build(); ExecutorService executor = Executors.newSingleThreadExecutor(); executor.execute(engine); } } Run the code and try create/update/delete the "CDC_TABLE" table in database, no record print.

SFDC Cases Counter:
SFDC Cases Links:

Description

I'm trying Oracle Connector with logminer and found partition tables' redolog is never read by the engine while non-partition tables appears normally.

After debugging I found the sql for searching the redo-sql is like below:

SELECT SCN,
 SQL_REDO,
 OPERATION_CODE,
 TIMESTAMP,
 XID,
 CSF,
 TABLE_NAME,
 SEG_NAME,
 SEG_OWNER,
 OPERATION,
 USERNAME
 FROM V$LOGMNR_CONTENTS
 WHERE OPERATION_CODE in (1, 2, 3, 5)
 AND SEG_OWNER = 'U_REPORT'
 AND table_name IN ('CDC_TABLE')
AND SCN >= ? AND SCN < ? 
---- Notice this clause
AND SEG_NAME IN ('CDC_TABLE')
----
 OR (OPERATION_CODE IN (5, 7, 34, 36) AND
 USERNAME NOT IN ('SYS', 'SYSTEM', 'LOGMINER'))
 ORDER BY SCN

Notice the 'SEG_NAME IN' clause, it's produced by the below code:

// io.debezium.connector.oracle.logminer.SqlUtils
private static String buildTableInPredicate(List<String> tables) {
        if (tables.size() == 0 || tables.size() > 1000) {
            LOGGER.warn(" Cannot apply {} whitelisted tables condition", tables.size());
            return "";
        }        StringJoiner tableNames = new StringJoiner(",");
        tables.forEach(table -> tableNames.add("'" + table + "'"));
        return " AND table_name IN (" + tableNames + ") AND SEG_NAME IN (" + tableNames + ") ";
    }

In this code, the SEG_NAME is always set to equal tableName, it's OK when the table is non-partition, but a partition tables' SEG_NAME is like 'tableName'.'partitionName' and will never meet this condition. This is the reason for my issue.

I talked with my DBA and he think the 'SEG_NAME' condition may be unnecessary. So I delete the condition and it can run on partition table for me and I‘v created a [PR|https://github.com/debezium/debezium-incubator/pull/205]. But I'm still wondering your concern about adding this clause.

Oracle LogMiner doesn't support partition tables

Details

Description

Attachments

Activity

People

Dates