[Oracle][MySQL][SqlServer][PostgresSQL] Fix Oracle/MySQL/SqlServer/PostgresSQL CDC parser schema change event failed #2315

e-mhui · 2023-07-18T02:41:55Z

Fix: #2055 #1792 #1379 #1807 #1810 #1803 #1804 #2311

When a SchemaChangeEvent is triggered, the schemaChangeRecordValue will obtain binlog information from sourceInfo. However, only MySQL CDC has binlog information, and other CDC will report an error after triggering a SchemaChangeEvent event.

So I provided an interface for each connector to implement their own sourceinfo parsing.

/** SchemaChangeEvent Handler. */
public interface SchemaChangeEventHandler {
    Map<String, Object> parseSource(SchemaChangeEvent event);
}

Then, it's used in JdbcSourceEventDispatcher#schemaChangeRecordValue.

        private Struct schemaChangeRecordValue(SchemaChangeEvent event) throws IOException {
            Map<String, Object> source = schemaChangeEventHandler.parseSource(event);
            HistoryRecord historyRecord =
                    new HistoryRecord(
                            source,
                            event.getOffset(),
                            event.getDatabase(),
                            null,
                            event.getSchema(),
                            event.getDdl(),
                            event.getTableChanges());
        // ... other code ...

For MySQL CDC

/** MySqlSchemaChangeEventHandler to deal MySql's schema change event. */
public class MySqlSchemaChangeEventHandler implements SchemaChangeEventHandler {

    @Override
    public Map<String, Object> parseSource(SchemaChangeEvent event) {
        Map<String, Object> source = new HashMap<>();
        Struct sourceInfo = event.getSource();
        String fileName = sourceInfo.getString(BINLOG_FILENAME_OFFSET_KEY);
        Long pos = sourceInfo.getInt64(BINLOG_POSITION_OFFSET_KEY);
        Long serverId = sourceInfo.getInt64(SERVER_ID_KEY);
        source.put(SERVER_ID_KEY, serverId);
        source.put(BINLOG_FILENAME_OFFSET_KEY, fileName);
        source.put(BINLOG_POSITION_OFFSET_KEY, pos);
        return source;
    }
}

For Oracle CDC

/** OracleSchemaChangeEventHandler to deal Oracle's schema change event. */
public class OracleSchemaChangeEventHandler implements SchemaChangeEventHandler {

    @Override
    public Map<String, Object> parseSource(SchemaChangeEvent event) {
        Map<String, Object> source = new HashMap<>();
        Struct sourceInfo = event.getSource();
        String scn = sourceInfo.getString(SCN_KEY);
        source.put(SCN_KEY, scn);
        return source;
    }
}

e-mhui · 2023-07-18T02:43:35Z

@ruanhang1993 pls review it.

ruanhang1993

@e-mhui Thanks for the work.
Maybe Oracle cdc should provide a new dispatcher by self instead of modifying the dispatcher in connector-base. WDYT?
Maybe we could change some scope of methods for JdbcSourceEventDispatcher to reuse its code.

flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/utils/SourceRecordUtils.java

e-mhui · 2023-07-19T14:27:56Z

@e-mhui Thanks for the work.
Maybe Oracle cdc should provide a new dispatcher by self instead of modifying the dispatcher in connector-base. WDYT?
Maybe we could change some scope of methods for JdbcSourceEventDispatcher to reuse its code.

@ruanhang1993 Hi, I have refactored the code, please review it again.

Use the schemaChangeEventHandler.parseSource(event) interface to handle different CDC sources. The construction of the source mainly refers to the following links:

Use io.debezium.connector.*.SchemaChangeKey to determine the SchemaChangeEvent.

gtk96 · 2023-07-26T01:15:49Z

@e-mhui please run mvn spotless:apply fix code style,thanks

e-mhui · 2023-07-26T01:47:53Z

@e-mhui please run mvn spotless:apply fix code style,thanks

Thanks, it has done.

gtk96 · 2023-08-30T09:21:01Z

@e-mhui please refer https://github.com/ververica/flink-cdc-connectors/blob/master/README.md#code-style fix your code style to pass ci check

e-mhui · 2023-08-30T11:01:45Z

@e-mhui please refer https://github.com/ververica/flink-cdc-connectors/blob/master/README.md#code-style fix your code style to pass ci check

@gtk96 I had been fixed the check style, can you rerun the workflow ?

gtk96 · 2023-09-01T03:37:45Z

cc @ruanhang1993

e-mhui · 2023-10-18T08:50:11Z

@ruanhang1993 pls review it.

cainbit · 2023-10-30T00:44:17Z

@ruanhang1993 Could you please help to review this pr? Many thanks.

ruanhang1993 · 2024-02-01T03:24:45Z

@e-mhui Hi. We need to add some tests to test this bug.
Please tell me when you need to review again. Thanks.

e-mhui · 2024-02-21T11:05:30Z

@e-mhui Hi. We need to add some tests to test this bug. Please tell me when you need to review again. Thanks.

@ruanhang1993 Hi, pls review again.

klion26 · 2024-02-29T02:23:52Z

thanks @e-mhui for contributing and @ruanhang1993 for merging this!

…hange events (apache#2315)

stardustman · 2024-03-14T09:06:15Z

SchemaChangeEventHandler，这个接口在哪个版本里，我找了一下，最新版的flink-cdc-base 3.0.0 没有看到相关代码。是需要自己修改源码自己编译吗

oracle 表结构变动报错，用到oracle-cdc 3.0包，我这边也报错，请问你这边解决了吗

This PR will fix it.

I did't find handler packages in the /com/ververica/cdc/connectors/base/relational and the flink-connector-oracle-cdc also not found, how to deal with the schema changes myself using the release-3.0? thanks a lot.@e-mhui

e-mhui · 2024-03-14T09:14:22Z

@stardustman @Jetol The latest release 3.0.1 was packaged on Jan 22 and has not fixed this issue. If you want to use it in advance, you can build it yourself from the master branch. Otherwise, you will have to wait for the next release version.

package command:

git clone https://github.com/ververica/flink-cdc-connectors.git
cd flink-cdc-connectors
mvn clean install -DskipTests

stardustman · 2024-03-15T07:08:14Z

@stardustman @Jetol The latest release 3.0.1 was packaged on Jan 22 and has not fixed this issue. If you want to use it in advance, you can build it yourself from the master branch. Otherwise, you will have to wait for the next release version.

package command:
git clone https://github.com/ververica/flink-cdc-connectors.git
cd flink-cdc-connectors
mvn clean install -DskipTests

hello, I'm a new user of the flink-cdc, I have get latest package using master branch. and import the dependecyies. will the OracleSchemaChangeEventHandler work automatically or I need other configs to enable it? thanks a lot. @e-mhui

e-mhui · 2024-03-15T07:33:28Z

@stardustman @Jetol The latest release 3.0.1 was packaged on Jan 22 and has not fixed this issue. If you want to use it in advance, you can build it yourself from the master branch. Otherwise, you will have to wait for the next release version.
package command:
git clone https://github.com/ververica/flink-cdc-connectors.git
cd flink-cdc-connectors
mvn clean install -DskipTests
hello, I'm a new user of the flink-cdc, I have get latest package using master branch. and import the dependecyies. will the OracleSchemaChangeEventHandler work automatically or I need other configs to enable it? thanks a lot. @e-mhui

@stardustmanYes, it will work automatically. However, you may need to configure some parameters to capture DDL events.

debezium.database.history.store.only.captured.tables.ddl = true

You can also refer to:

stardustman · 2024-03-15T08:08:54Z

debezium.database.history.store.only.captured.tables.ddl

I read the doc docs/connectors/cdc-connectors/oracle-cdc/#connector-options. is it the config belongs to debezium should be started with "debezium" when using the flink cdc? but I find the demo code cdc-connectors/oracle-cdc/#datastream-source
this line debeziumProperties.setProperty("log.mining.strategy", "online_catalog"); of the main method, log.mining.strategy didn't start with the debezium, after read the log.mining.strategy it seems the property is belongs to debezium, and the log.mining.strategy also need to be set redo_log_catalog to capture the DDL?

 public static void main(String[] args) throws Exception {
        Properties debeziumProperties = new Properties();
        debeziumProperties.setProperty("log.mining.strategy", "online_catalog");

will the "log.mining.strategy" need to start with "debezium"? thanks a lot for bothering you many times😄. @e-mhui

e-mhui · 2024-03-15T08:33:44Z

debezium.database.history.store.only.captured.tables.ddl
I read the doc docs/connectors/cdc-connectors/oracle-cdc/#connector-options. is it the config belongs to debezium should be started with "debezium" when using the flink cdc? but I find the demo code cdc-connectors/oracle-cdc/#datastream-source this line debeziumProperties.setProperty("log.mining.strategy", "online_catalog"); of the main method, log.mining.strategy didn't start with the debezium, after read the log.mining.strategy it seems the property is belongs to debezium, and the log.mining.strategy also need to be set redo_log_catalog to capture the DDL?
 public static void main(String[] args) throws Exception {
        Properties debeziumProperties = new Properties();
        debeziumProperties.setProperty("log.mining.strategy", "online_catalog"); 
will the "log.mining.strategy" need to start with "debezium"? thanks a lot for bothering you many times😄. @e-mhui

It is a parameter of debezium, so you should add the prefix debezium..

@e-mhui

fix error the "log.mining.strategy" should be start with prefix "debezium." according to https://nightlies.apache.org/flink/flink-cdc-docs-release-3.0/zh/docs/connectors/cdc-connectors/oracle-cdc/#connector-options and the help from @e-mhui with the link apache#2315 (comment)

…event.

stardustman · 2024-03-18T08:00:06Z

SchemaChangeEventHandler，这个接口在哪个版本里，我找了一下，最新版的flink-cdc-base 3.0.0 没有看到相关代码。是需要自己修改源码自己编译吗

oracle 表结构变动报错，用到oracle-cdc 3.0包，我这边也报错，请问你这边解决了吗

你好，你是实现 DebeziumDeserializationSchema 接口自定义 Deserialization 来接收数据的么? 在这里是怎样区分 DDL 和实际表的数据的呢？有没有脱敏的 demo 可以发一下？谢谢！@Jetol @e-mhui

…hange events (apache#2315)

mdzz9527 · 2024-06-21T06:12:39Z

合并到哪个版本了呢

shangeyao · 2024-07-08T08:10:46Z

合并到哪个版本了呢

3.1

…hange events (apache#2315)

e-mhui changed the title ~~Fix Oracle CDC parser schema change event failed~~ [Oracle] Fix Oracle CDC parser schema change event failed Jul 18, 2023

ruanhang1993 reviewed Jul 19, 2023

View reviewed changes

flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/utils/SourceRecordUtils.java Outdated Show resolved Hide resolved

rbreejen mentioned this pull request Jul 27, 2023

[Bug][Oracle] Fix option to not include.schema.changes #2350

Closed

2 tasks

e-mhui force-pushed the flink-cdc-oracle-2311 branch from 262f827 to 51170a0 Compare August 18, 2023 02:16

e-mhui force-pushed the flink-cdc-oracle-2311 branch from 67ff61b to d8a4984 Compare August 30, 2023 09:26

JNSimba mentioned this pull request Oct 13, 2023

[cdc-base] IncrementalSourceRecordEmitter process more than mysql schema change event #2545

Closed

2 tasks

e-mhui mentioned this pull request Oct 18, 2023

[Oracle] Fix Oracle CDC cannot capture newly added tables during task running #2078

Closed

e-mhui force-pushed the flink-cdc-oracle-2311 branch from a58156d to f442ccc Compare October 18, 2023 09:40

leonardBang requested a review from ruanhang1993 October 24, 2023 06:35

e-mhui force-pushed the flink-cdc-oracle-2311 branch from f442ccc to db96c9c Compare November 8, 2023 02:05

github-actions bot added docs Improvements or additions to documentation values-pipeline-connector composer common runtime mongodb-cdc-connector build mysql-cdc-connector base oracle-cdc-connector labels Dec 2, 2023

gong mentioned this pull request Feb 3, 2024

oracle cdc DataException: file is not a valid field name #1792

Closed

e-mhui added 5 commits February 19, 2024 11:01

[Oracle] Add Oracle UT

f2fe6fc

[Oracle] Add Oracle UT

a7faf5d

[Oracle] Add SqlServer UT

d6b0606

[Oracle] Add SqlServer UT

89d68d7

[Oracle] Rerun workflow

310391a

gong mentioned this pull request Feb 22, 2024

Oracle ddl has bug #3073

Closed

2 tasks

e-mhui added 2 commits February 27, 2024 18:00

[Oracle] Modify SqlServer UT

ed627c0

[Oracle] Rerun workflow

f960a2a

ruanhang1993 merged commit ea7b0e8 into apache:master Feb 28, 2024
17 checks passed

ruanhang1993 mentioned this pull request Feb 28, 2024

[Bug] [Oracle] Oracle CDC capture schema change event failed #2311

Closed

2 tasks

aiwenmo pushed a commit to aiwenmo/flink-cdc-connectors that referenced this pull request Mar 6, 2024

[cdc-base] Support to parse Oracle/SqlServer/PostgresSQL CDC schema c…

0fadc1c

…hange events (apache#2315)

This was referenced Mar 15, 2024

Update oracle-cdc.md stardustman/flink-cdc#1

Merged

Update oracle-cdc.md #3151

Closed

abatorQ pushed a commit to abatorQ/flink-cdc that referenced this pull request Mar 18, 2024

bug fix : apache#2315 oracle cdc failed when ddl fires schema change …

838aaa9

…event.

joyCurry30 pushed a commit to joyCurry30/flink-cdc-connectors that referenced this pull request Mar 22, 2024

[cdc-base] Support to parse Oracle/SqlServer/PostgresSQL CDC schema c…

22512ee

…hange events (apache#2315)

wuzhenhua01 pushed a commit to wuzhenhua01/flink-cdc-connectors that referenced this pull request Aug 4, 2024

[cdc-base] Support to parse Oracle/SqlServer/PostgresSQL CDC schema c…

ca08b99

…hange events (apache#2315)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Oracle][MySQL][SqlServer][PostgresSQL] Fix Oracle/MySQL/SqlServer/PostgresSQL CDC parser schema change event failed #2315

[Oracle][MySQL][SqlServer][PostgresSQL] Fix Oracle/MySQL/SqlServer/PostgresSQL CDC parser schema change event failed #2315

e-mhui commented Jul 18, 2023 •

edited

Loading

e-mhui commented Jul 18, 2023

ruanhang1993 left a comment •

edited

Loading

e-mhui commented Jul 19, 2023 •

edited

Loading

gtk96 commented Jul 26, 2023

e-mhui commented Jul 26, 2023

gtk96 commented Aug 30, 2023

e-mhui commented Aug 30, 2023

gtk96 commented Sep 1, 2023

e-mhui commented Oct 18, 2023

cainbit commented Oct 30, 2023

ruanhang1993 commented Feb 1, 2024

e-mhui commented Feb 21, 2024

klion26 commented Feb 29, 2024

stardustman commented Mar 14, 2024 •

edited

Loading

e-mhui commented Mar 14, 2024 •

edited

Loading

stardustman commented Mar 15, 2024 •

edited

Loading

e-mhui commented Mar 15, 2024

stardustman commented Mar 15, 2024

e-mhui commented Mar 15, 2024

stardustman commented Mar 18, 2024

mdzz9527 commented Jun 21, 2024

shangeyao commented Jul 8, 2024

[Oracle][MySQL][SqlServer][PostgresSQL] Fix Oracle/MySQL/SqlServer/PostgresSQL CDC parser schema change event failed #2315

[Oracle][MySQL][SqlServer][PostgresSQL] Fix Oracle/MySQL/SqlServer/PostgresSQL CDC parser schema change event failed #2315

Conversation

e-mhui commented Jul 18, 2023 • edited Loading

e-mhui commented Jul 18, 2023

ruanhang1993 left a comment • edited Loading

Choose a reason for hiding this comment

e-mhui commented Jul 19, 2023 • edited Loading

gtk96 commented Jul 26, 2023

e-mhui commented Jul 26, 2023

gtk96 commented Aug 30, 2023

e-mhui commented Aug 30, 2023

gtk96 commented Sep 1, 2023

e-mhui commented Oct 18, 2023

cainbit commented Oct 30, 2023

ruanhang1993 commented Feb 1, 2024

e-mhui commented Feb 21, 2024

klion26 commented Feb 29, 2024

stardustman commented Mar 14, 2024 • edited Loading

e-mhui commented Mar 14, 2024 • edited Loading

stardustman commented Mar 15, 2024 • edited Loading

e-mhui commented Mar 15, 2024

stardustman commented Mar 15, 2024

e-mhui commented Mar 15, 2024

stardustman commented Mar 18, 2024

mdzz9527 commented Jun 21, 2024

shangeyao commented Jul 8, 2024

e-mhui commented Jul 18, 2023 •

edited

Loading

ruanhang1993 left a comment •

edited

Loading

e-mhui commented Jul 19, 2023 •

edited

Loading

stardustman commented Mar 14, 2024 •

edited

Loading

e-mhui commented Mar 14, 2024 •

edited

Loading

stardustman commented Mar 15, 2024 •

edited

Loading