flink-cdc oracle incremental-snapshot log.mining.strategy set log.mining.strategy" to "redo_log_catalog" not get data?

[stardustman]

I have read the doc incremental-snapshot-based-datastream-experimental , and run the example code. It works. but I also need to get the ddl changes, so I change the debeziumProperties.setProperty("log.mining.strategy", "online_catalog");
to debeziumProperties.setProperty("log.mining.strategy", "redo_log_catalog");
after I run the modified code, there is nothing print in the console.
I set a breakpoint at JsonDebeziumDeserializationSchema#deserialize method, the method not invoked at all.
why? how to resolve this?

[yuxiqian]

Hi @stardustman, when specifying the deserializer as JsonDebeziumDeserializationSchema, you may pass an optional argument includeSchema which is false by default. Enabling this should allow it capturing schema change events.

However, there’s a known bug in version 3.0.1 and lower, which blocks Oracle database from capturing schema information. Patch #2315 has been merged, but it won’t be available until release 3.1. You may try compiling Flink CDC at release-3.1 branch and see if this problem persists.

[stardustman]

I not explain clearly. I have compiled the master branch code to capture ddl changes last month and success to capture the ddl changes, according the https://debezium.io/documentation/reference/1.9/connectors/oracle.html#oracle-property-log-mining-strategy the log.mining.strategy" needs to be set "redo_log_catalog", use the online_catalog will encounter the error like this

[ERROR ] 2024-03-20 14:53:07.779 io.debezium.connector.oracle.logminer.LogMinerHelper logError 117 debezium-oracleconnector-oracle_logminer-change-event-source-coordinator Mining session stopped due to the {}
io.debezium.DebeziumException: Oracle failed to re-construct redo SQL 'insert into "YY"."SYS_USER_ROLE"("COL 1","COL 2","COL 3") values (HEXTORAW('c207'),HEXTORAW('c208'),HEXTORAW('61646d696e'));'

by my test.
now I want to use the Incremental snapshot to short the time which full snapshot read needs.
Before I change the debeziumProperties.setProperty("log.mining.strategy", "online_catalog");
to debeziumProperties.setProperty("log.mining.strategy", "redo_log_catalog"); the example code works well. and can get the data from capture table.
after I change the debeziumProperties.setProperty("log.mining.strategy", "online_catalog");
to debeziumProperties.setProperty("log.mining.strategy", "redo_log_catalog"); to prepare get ddl changes. but I didn't make any ddl changes. but the example code can't get the data from capture table.
it seems to need get latest code to compile again? @yuxiqian

[yuxiqian]

Hi @stardustman, incremental snapshot framework isn't implemented based on Debezium backend so Debezium properties including log.mining.strategy are not be available.

[stardustman]

what's the meaning of ' incremental snapshot framework isn't implemented based on Debezium backend'? I find the debezium doc https://debezium.io/documentation/reference/1.9/connectors/oracle.html#oracle-incremental-snapshots, does the flink-cdc oracle reimplement the incremental-snapshots? if I want to use incremental-snapshots at full snapshot phase , and after the full snapshot phase ends, need to get the ddl changes. how to do it? @yuxiqian

[stardustman]

Before change the debeziumProperties.setProperty("log.mining.strategy", "online_catalog");
to debeziumProperties.setProperty("log.mining.strategy", "redo_log_catalog");
the data from target table call stack is:

deserialize(SourceRecord, Collector):73, JsonDebeziumDeserializationSchema (org.apache.flink.cdc.debezium), JsonDebeziumDeserializationSchema.java
emitElement(SourceRecord, SourceOutput):156, IncrementalSourceRecordEmitter (org.apache.flink.cdc.connectors.base.source.reader), IncrementalSourceRecordEmitter.java
processElement(SourceRecord, SourceOutput, SourceSplitState):118, IncrementalSourceRecordEmitter (org.apache.flink.cdc.connectors.base.source.reader), IncrementalSourceRecordEmitter.java
emitRecord(SourceRecords, SourceOutput, SourceSplitState):88, IncrementalSourceRecordEmitter (org.apache.flink.cdc.connectors.base.source.reader), IncrementalSourceRecordEmitter.java
emitRecord(Object, SourceOutput, Object):57, IncrementalSourceRecordEmitter (org.apache.flink.cdc.connectors.base.source.reader), IncrementalSourceRecordEmitter.java
pollNext(ReaderOutput):160, SourceReaderBase (org.apache.flink.connector.base.source.reader), SourceReaderBase.java
emitNext(PushingAsyncDataInput$DataOutput):419, SourceOperator (org.apache.flink.streaming.api.operators), SourceOperator.java
emitNext(PushingAsyncDataInput$DataOutput):68, StreamTaskSourceInput (org.apache.flink.streaming.runtime.io), StreamTaskSourceInput.java
processInput():65, StreamOneInputProcessor (org.apache.flink.streaming.runtime.io), StreamOneInputProcessor.java
processInput(MailboxDefaultAction$Controller):562, StreamTask (org.apache.flink.streaming.runtime.tasks), StreamTask.java
runDefaultAction(MailboxDefaultAction$Controller):-1, 272917927 (org.apache.flink.streaming.runtime.tasks.StreamTask$$Lambda$1165), Unknown Source
runMailboxLoop():231, MailboxProcessor (org.apache.flink.streaming.runtime.tasks.mailbox), MailboxProcessor.java
runMailboxLoop():858, StreamTask (org.apache.flink.streaming.runtime.tasks), StreamTask.java
invoke():807, StreamTask (org.apache.flink.streaming.runtime.tasks), StreamTask.java
run():-1, 559471872 (org.apache.flink.runtime.taskmanager.Task$$Lambda$1426), Unknown Source
runWithSystemExitMonitoring(RunnableWithException):953, Task (org.apache.flink.runtime.taskmanager), Task.java
restoreAndInvoke(TaskInvokable):932, Task (org.apache.flink.runtime.taskmanager), Task.java
doRun():746, Task (org.apache.flink.runtime.taskmanager), Task.java
run():562, Task (org.apache.flink.runtime.taskmanager), Task.java
run():748, Thread (java.lang), Thread.java

and data print out in the console.
After change the debeziumProperties.setProperty("log.mining.strategy", "online_catalog");
to debeziumProperties.setProperty("log.mining.strategy", "redo_log_catalog");
this is nothing print out in the console.
acutally the data has read from table, like this

[DEBUG] 2024-05-09 10:06:34.471 io.debezium.connector.base.ChangeEventQueue doEnqueue 196 debezium-snapshot-reader-0 Enqueuing source record 'DataChangeEvent [record=SourceRecord{sourcePartition={server=oracle_logminer}, sourceOffset={commit_scn=0:1:, transaction_id=null, snapshot_scn=null, scn=0}} ConnectRecord{topic='oracle_logminer.C__YY.PINFO', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{ID=3, 
[DEBUG] 2024-05-09 10:06:34.471 io.debezium.connector.base.ChangeEventQueue doEnqueue 196 debezium-snapshot-reader-3 Enqueuing source record 'DataChangeEvent [record=SourceRecord{sourcePartition={server=oracle_logminer}, sourceOffset={commit_scn=0:1:, transaction_id=null, snapshot_scn=null, scn=0}} ConnectRecord{topic='oracle_logminer.C__YY.PINFO', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{ID=4,
[DEBUG] 2024-05-09 10:06:34.471 io.debezium.connector.base.ChangeEventQueue doEnqueue 196 debezium-snapshot-reader-1 Enqueuing source record 'DataChangeEvent [record=SourceRecord{sourcePartition={server=oracle_logminer}, sourceOffset={commit_scn=0:1:, transaction_id=null, snapshot_scn=null, scn=0}} ConnectRecord{topic='oracle_logminer.C__YY.PINFO', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{ID=1,

I set a breakpoint at JsonDebeziumDeserializationSchema#deserialize method, the method not invoked at all.
@yuxiqian

[stardustman]

I test again.
set the debeziumProperties.setProperty("log.mining.strategy", "redo_log_catalog"); this time get data. but there is a long latency to get the data, there is only 40 rows in the table. split size is 20, 2 parallelism, only get 3 rows after the code running about 5 mins. the io.debezium.connector.base.ChangeEventQueue doEnqueue 196 debezium-snapshot-reader-0 Enqueuing source record has log 38 rows in the table.
set the debeziumProperties.setProperty("log.mining.strategy", "online_catalog"); not has so long latency to get data from table.