diff --git a/.github/workflows/push_pr.yml b/.github/workflows/push_pr.yml index 033254d..3bc7aaf 100644 --- a/.github/workflows/push_pr.yml +++ b/.github/workflows/push_pr.yml @@ -30,10 +30,10 @@ jobs: with: module: flink-connector-oceanbase-directload - flink-connector-oceanbase-tools-cdc: + flink-connector-oceanbase-cli: uses: ./.github/workflows/test.yml with: - module: flink-connector-oceanbase-tools-cdc + module: flink-connector-oceanbase-cli flink-connector-oceanbase-e2e-tests: strategy: diff --git a/docs/sink/flink-connector-oceanbase-tools-cdc.md b/docs/sink/flink-connector-oceanbase-cli.md similarity index 69% rename from docs/sink/flink-connector-oceanbase-tools-cdc.md rename to docs/sink/flink-connector-oceanbase-cli.md index 3486310..e20130d 100644 --- a/docs/sink/flink-connector-oceanbase-tools-cdc.md +++ b/docs/sink/flink-connector-oceanbase-cli.md @@ -1,41 +1,12 @@ -# Flink Connector OceanBase By Tools CDC +# Flink Connector OceanBase CLI -English | [简体中文](flink-connector-oceanbase-tools-cdc_cn.md) +English | [简体中文](flink-connector-oceanbase-cli_cn.md) -This project is a flink command line tool that supports the synchronization of CDC tasks to oceanbase through the Flink command line, which greatly simplifies the command writing of data synchronization to oceanbase through flink. +The project is a set of CLI (command line interface) tools that supports submitting Flink jobs to migrate data from other data sources to OceanBase. ## Getting Started -You can get the release packages at [Releases Page](https://github.com/oceanbase/flink-connector-oceanbase/releases) or [Maven Central](https://central.sonatype.com/artifact/com.oceanbase/flink-connector-oceanbase-directload). - -```xml - - com.oceanbase - flink-connector-oceanbase-tools-cdc - ${project.version} - -``` - -If you want to use the latest snapshot version, you can specify by configuring the Maven snapshot repository: - -```xml - - com.oceanbase - flink-connector-oceanbase-tools-cdc - ${project.version} - - - - - sonatype-snapshots - Sonatype Snapshot Repository - https://s01.oss.sonatype.org/content/repositories/snapshots/ - - true - - - -``` +You can get the release packages at [Releases Page](https://github.com/oceanbase/flink-connector-oceanbase/releases) or [Maven Central](https://central.sonatype.com/artifact/com.oceanbase/flink-connector-oceanbase-cli)，or get the latest snapshot packages at [Sonatype Snapshot](https://s01.oss.sonatype.org/content/repositories/snapshots/com/oceanbase/flink-connector-oceanbase-cli). You can also manually build it from the source code. @@ -45,17 +16,23 @@ cd flink-connector-oceanbase mvn clean package -DskipTests ``` -### Notes: +### Using Flink CDC as Source + +#### Dependencies + +This project is based on the SQL Client JAR of [Flink CDC Source Connector](https://nightlies.apache.org/flink/flink-cdc-docs-master/docs/connectors/flink-sources/overview/). + +We do not provide Flink CDC Source Connector in the JAR package of this project, so you need to manually download the used Flink CDC SQL JAR. Note that this project requires Flink CDC to be 3.2.0 or later version. + +If you're using Flink Oracle CDC as source, you need also download the dependencies of the source connector, see the *Dependencies* chapter of [Oracle CDC Connector](https://nightlies.apache.org/flink/flink-cdc-docs-master/docs/connectors/flink-sources/oracle-cdc/#sql-client-jar). -* Currently, the project supports using Flink CDC to access multiple tables or the entire database. During synchronization, you need to add the corresponding Flink CDC dependency in the `$FLINK_HOME/lib` directory, such as flink-sql-connector-mysql-cdc-\${version}. jar, flink-sql-connector-oracle-cdc-\${version}.jar, flink-sql-connector-sqlserver-cdc-\${version}.jar. - * The dependent Flink CDC version needs to be above 3.1. If you need to use Flink CDC to synchronize MySQL and Oracle, you also need to add the relevant JDBC driver under `$FLINK_HOME/lib`. - * If you synchronize data to OceanBase, you must use oceanBase or mysql as the protocol name for the URL connection string of OceanBase. +#### Demo: Migrate from Flink MySQL CDC to OceanBase -### MySQL Synchronous OceanBase Example +##### Preparation -#### Geting Ready +Add the CLI JAR `flink-connector-oceanbase-cli-xxx.jar` and dependency JAR `flink-sql-connector-mysql-cdc-xxx.jar` to `$FLINK_HOME/lib`. -Create a table test_history_strt_sink in a MySql database test_db library, test_history_text. +Then prepare tables and data in MySQL database. ```mysql use test_db; @@ -73,18 +50,21 @@ CREATE TABLE test_history_text ( ns integer DEFAULT '0' NOT NULL, PRIMARY KEY (itemid,clock,ns) ); + +INSERT INTO test_db.test_history_text (itemid,clock,value,ns) VALUES + (1,21131,'ces1',21321); ``` -#### Build A Flink Task +##### Submit Job via CLI -##### An example of the Flink command line +Replace the following command with your real database information, and execute it to submit a Flink job. ```shell $FLINK_HOME/bin/flink run \ -Dexecution.checkpointing.interval=10s \ -Dparallelism.default=1 \ -c com.oceanbase.connector.flink.tools.cdc.CdcTools \ - lib/flink-connector-oceanbase-tools-cdc-${version}.jar \ + lib/flink-connector-oceanbase-cli-xxx.jar \ mysql-sync-database \ --database test_db \ --mysql-conf hostname=xxxx \ @@ -98,66 +78,26 @@ $FLINK_HOME/bin/flink run \ --sink-conf url=jdbc:mysql://xxxx:xxxx ``` -Replace the above database information with your real database information, and when a message similar to the following appears, the task is successfully built and submitted. +##### Check and Verify -```shell -Job has been submitted with JobID 0177b201a407045a17445aa288f0f111 -``` +Check the target OceanBase database, you should find out these two tables and one row data. -The tool automatically parses the information on the command line and creates a table, which can be queried and verified in OceanBase. - -MySQL to insert test data +You can go on insert test data to MySQL database as below: ```sql INSERT INTO test_db.test_history_str (itemid,clock,value,ns) VALUES (1,2,'ces1',1123); INSERT INTO test_db.test_history_text (itemid,clock,value,ns) VALUES - (1,21131,'ces1',21321), (2,21321,'ces2',12321); ``` Since it is a CDC task, after data is inserted in MySQL, you can query and verify the synchronized data in OceanBase. -### Parameter parsing - -This configuration is the program configuration information of flink - -```shell --Dexecution.checkpointing.interval=10s --Dparallelism.default=1 -``` - -Specify the JAR package of the program and the entry of the program - -```shell --c com.oceanbase.connector.flink.tools.cdc.CdcTools \ -lib/flink-connector-oceanbase-tools-cdc-${version}.jar \ -``` - -The name of the database - -```shell ---database test_db -``` - -This name is customized, meaning the name given to this database, and ultimately serves as the naming rule for flink tasks. - -## Configuration Items - -#### Supported data sources - -| Data source identifier | Data source | -|-------------------------|----------------------| -| mysql-sync-database | mysql datasource | -| oracle-sync-database | oracle datasource | -| postgres-sync-database | postgres datasource | -| sqlserver-sync-database | sqlserver datasource | -| db2-sync-database | db2 datasource | - -#### Configuration Items +#### Options -| Configuration Items | Comment | +| Option | Comment | |------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| {identifier} | Data source identifier, only mysql type `mysql-sync-database` is verified now. | | --job-name | Flink task name, optional. | | --database | Database name synchronized to OceanBase. | | --table-prefix | OceanBase table prefix name, such as --table-prefix ods_. | @@ -173,13 +113,13 @@ This name is customized, meaning the name given to this database, and ultimately | --ignore-default-value | Turn off synchronization of MySQL table structures by default. It is suitable for the situation when synchronizing MySQL data to oceanbase, the field has a default value, but the actual inserted data is null. | | --create-table-only | Whether to only synchronize the structure of the table. | -#### Configuration items of sink-conf +`--sink-conf` option -| Configuration Items | Default Value | Required | Comment | -|---------------------|---------------|----------|-------------------------------------------------------------------| -| url | -- | N | jdbc connection information, such as: jdbc:mysql://127.0.0.1:2881 | -| username | -- | Y | Username to access oceanbase | -| password | -- | Y | Password to access oceanbase | +| Option | Default Value | Required | Comment | +|----------|---------------|----------|-------------------------------------------------------------------| +| url | -- | N | jdbc connection information, such as: jdbc:mysql://127.0.0.1:2881 | +| username | -- | Y | Username to access oceanbase | +| password | -- | Y | Password to access oceanbase | ## Reference information diff --git a/docs/sink/flink-connector-oceanbase-tools-cdc_cn.md b/docs/sink/flink-connector-oceanbase-cli_cn.md similarity index 71% rename from docs/sink/flink-connector-oceanbase-tools-cdc_cn.md rename to docs/sink/flink-connector-oceanbase-cli_cn.md index fecd551..8601aa5 100644 --- a/docs/sink/flink-connector-oceanbase-tools-cdc_cn.md +++ b/docs/sink/flink-connector-oceanbase-cli_cn.md @@ -1,41 +1,12 @@ -# Flink Connector OceanBase By Tools CDC +# Flink Connector OceanBase CLI -[English](flink-connector-oceanbase-tools-cdc.md) | 简体中文 +[English](flink-connector-oceanbase-cli.md) | 简体中文 -本项目是一个可以支持通过Flink命令行构建cdc任务同步到oceanbase的flink命令行工具，极大的简化了通过flink同步数据到oceanbase的命令书写。 +本项目是一套 CLI（命令行界面）工具，支持提交 Flink 作业将数据从其他数据源迁移到 OceanBase。 ## 开始上手 -您可以在 [Releases 页面](https://github.com/oceanbase/flink-connector-oceanbase/releases) 或者 [Maven 中央仓库](https://central.sonatype.com/artifact/com.oceanbase/flink-connector-oceanbas-directload) 找到正式的发布版本。 - -```xml - - com.oceanbase - flink-connector-oceanbase-tools-cdc - ${project.version} - -``` - -如果你想要使用最新的快照版本，可以通过配置 Maven 快照仓库来指定： - -```xml - - com.oceanbase - flink-connector-oceanbase-tools-cdc - ${project.version} - - - - - sonatype-snapshots - Sonatype Snapshot Repository - https://s01.oss.sonatype.org/content/repositories/snapshots/ - - true - - - -``` +您可以在 [Releases 页面](https://github.com/oceanbase/flink-connector-oceanbase/releases) 或者 [Maven 中央仓库](https://central.sonatype.com/artifact/com.oceanbase/flink-connector-oceanbase-cli) 找到正式的发布版本，或者从 [Sonatype Snapshot](https://s01.oss.sonatype.org/content/repositories/snapshots/com/oceanbase/flink-connector-oceanbase-cli) 获取最新的快照版本。 您也可以通过源码构建的方式获得程序包。 @@ -45,17 +16,23 @@ cd flink-connector-oceanbase mvn clean package -DskipTests ``` -### 注意事项： +### 使用 Flink CDC 作为源端 -* 目前项目支持使用Flink CDC接入多表或整库。同步时需要在 `$FLINK_HOME/lib` 目录下添加对应的 Flink CDC 依赖，比如 flink-sql-connector-mysql-cdc-\${version}.jar，flink-sql-connector-oracle-cdc-\${version}.jar ，flink-sql-connector-sqlserver-cdc-\${version}.jar -* 依赖的 Flink CDC 版本需要在 3.1 以上，如果需使用 Flink CDC 同步 MySQL 和 Oracle，还需要在 `$FLINK_HOME/lib` 下增加相关的 JDBC 驱动。 -* 同步至oceanbase时，oceanbase的url连接串需要使用oceanbase或mysql作为协议名称。 +#### 依赖 -### MySQL同步至OceanBase示例 +本项目基于 [Flink CDC Source 连接器](https://nightlies.apache.org/flink/flink-cdc-docs-master/docs/connectors/flink-sources/overview/) 的 SQL 客户端 JAR。 + +本项目的 JAR 包中未包含 Flink CDC Source 连接器，因此您需要手动下载使用的 Flink CDC SQL JAR。请注意，本项目要求 Flink CDC 为 3.2.0 或更高版本。 + +如果您使用 Flink Oracle CDC 作为源端，您还需要下载源连接器的依赖项，请参阅 [Oracle CDC 连接器](https://nightlies.apache.org/flink/flink-cdc-docs-master/docs/connectors/flink-sources/oracle-cdc/#sql-client-jar) 的 *依赖项* 章节。 + +#### 示例：从 Flink MySQL CDC 迁移数据到 OceanBase #### 准备 -在 MySql 数据库test_db库中创建表 test_history_strt_sink，test_history_text。 +将 CLI JAR `flink-connector-oceanbase-cli-xxx.jar` 和依赖 JAR `flink-sql-connector-mysql-cdc-xxx.jar` 添加到 `$FLINK_HOME/lib`。 + +然后在 MySQL 数据库中准备表和数据。 ```mysql use test_db; @@ -73,18 +50,21 @@ CREATE TABLE test_history_text ( ns integer DEFAULT '0' NOT NULL, PRIMARY KEY (itemid,clock,ns) ); + +INSERT INTO test_db.test_history_text (itemid,clock,value,ns) VALUES + (1,21131,'ces1',21321); ``` -#### 构建Flink任务 +##### 通过 CLI 提交作业 -##### Flink命令行示例 +将以下命令替换为您的真实数据库信息，并执行它以提交 Flink 作业。 ```shell $FLINK_HOME/bin/flink run \ -Dexecution.checkpointing.interval=10s \ -Dparallelism.default=1 \ -c com.oceanbase.connector.flink.tools.cdc.CdcTools \ - lib/flink-connector-oceanbase-tools-cdc-${version}.jar \ + lib/flink-connector-oceanbase-cli-xxx.jar \ mysql-sync-database \ --database test_db \ --mysql-conf hostname=xxxx \ @@ -100,69 +80,31 @@ $FLINK_HOME/bin/flink run \ 请将以上的数据库信息替换为您真实的数据库信息，当出现类似于以下的信息时，任务构建成功并提交。 -```shell -Job has been submitted with JobID 0177b201a407045a17445aa288f0f111 -``` +##### 检查和验证 -工具会自动解析命令行的信息并进行建表，可在OceanBase 中查询验证。 +检查目标 OceanBase 数据库，你应该找到这两个表和一行数据。 -MySQl中插入测试数据 +你可以继续将测试数据插入到 MySQL 数据库，如下所示： ```sql INSERT INTO test_db.test_history_str (itemid,clock,value,ns) VALUES (1,2,'ces1',1123); INSERT INTO test_db.test_history_text (itemid,clock,value,ns) VALUES - (1,21131,'ces1',21321), (2,21321,'ces2',12321); ``` 由于是CDC任务，MySQL中插入数据后，即可在 OceanBase 中查询验证同步过来的数据。 -### 参数解析 - -该配置是flink的程序配置信息 - -```shell --Dexecution.checkpointing.interval=10s --Dparallelism.default=1 -``` - -指定程序的jar包和程序的入口 - -```shell --c com.oceanbase.connector.flink.tools.cdc.CdcTools \ -lib/flink-connector-oceanbase-tools-cdc-${version}.jar \ -``` - -数据库名称 - -```shell ---database test_db -``` - -这个名称是自定义的，意为给这个数据库取的名字，最终作为flink任务的命名规则。 - -## 配置项 - -#### 支持的数据源 - -| 数据源标识 | 数据源 | -|-------------------------|--------------| -| mysql-sync-database | mysql数据源 | -| oracle-sync-database | oracle数据源 | -| postgres-sync-database | postgres数据源 | -| sqlserver-sync-database | sqlserver数据源 | -| db2-sync-database | db2数据源 | - #### 配置项 | 配置项 | 描述 | -|------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------------------------------| +|------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| {identifier} | 数据源标识，目前经过验证的只有 MySQL 源端 `mysql-sync-database`。 | | --job-name | Flink 任务名称，非必需。 | | --database | 同步到 OceanBase 的数据库名。 | | --table-prefix | OceanBase表前缀名，例如 --table-prefix ods_。 | | --table-suffix | 同上，OceanBase表的后缀名。 | -| --including-tables | 需要同步的 MySQL 表，可以使用｜分隔多个表，并支持正则表达式。比如--including-tables table1。 | `分隔多个表，并支持正则表达式。比如--including-tables table1 | +| --including-tables | 需要同步的 MySQL 表，可以使用｜分隔多个表，并支持正则表达式。比如--including-tables table1。 | | --excluding-tables | 不需要同步的表，用法同上。 | | --mysql-conf | MySQL CDCSource 配置，其中 hostname/username/password/database-name 是必需的。同步的库表中含有非主键表时，必须设置scan.incremental.snapshot.chunk.key-column，且只能选择非空类型的一个字段。
例如：`scan.incremental.snapshot.chunk.key-column=database.table:column,database.table1:column...`，不同的库表列之间用`,`隔开。 | | --oracle-conf | Oracle CDCSource 配置，其中 hostname/username/password/database-name/schema-name 是必需的。 | @@ -173,7 +115,7 @@ lib/flink-connector-oceanbase-tools-cdc-${version}.jar \ | --ignore-default-value | 关闭同步 MySQL 表结构的默认值。适用于同步 MySQL 数据到 OceanBase 时，字段有默认值，但实际插入数据为 null 情况。 | | --create-table-only | 是否只仅仅同步表的结构。 | -#### sink-conf的配置项 +`--sink-conf` 配置项： | 配置项 | 默认值 | 是否需要 | 描述 | |----------|-----|------|-----------------------------------------| diff --git a/flink-connector-oceanbase-tools-cdc/pom.xml b/flink-connector-oceanbase-cli/pom.xml similarity index 89% rename from flink-connector-oceanbase-tools-cdc/pom.xml rename to flink-connector-oceanbase-cli/pom.xml index f8c79af..cd27011 100644 --- a/flink-connector-oceanbase-tools-cdc/pom.xml +++ b/flink-connector-oceanbase-cli/pom.xml @@ -22,9 +22,14 @@ under the License. ${revision} - flink-connector-oceanbase-tools-cdc + flink-connector-oceanbase-cli jar + + 3.2.1 + 19.3.0.0 + + com.oceanbase @@ -35,7 +40,7 @@ under the License. org.apache.flink flink-sql-connector-mysql-cdc - ${flink.sql.cdc.version} + ${flink.cdc.version} provided @@ -47,7 +52,7 @@ under the License. org.apache.flink flink-sql-connector-oracle-cdc - ${flink.sql.cdc.version} + ${flink.cdc.version} provided @@ -59,7 +64,7 @@ under the License. org.apache.flink flink-sql-connector-postgres-cdc - ${flink.sql.cdc.version} + ${flink.cdc.version} provided @@ -72,7 +77,7 @@ under the License. org.apache.flink flink-sql-connector-sqlserver-cdc - ${flink.sql.cdc.version} + ${flink.cdc.version} provided @@ -84,7 +89,7 @@ under the License. org.apache.flink flink-sql-connector-db2-cdc - ${flink.sql.cdc.version} + ${flink.cdc.version} provided @@ -112,7 +117,6 @@ under the License. org.testcontainers mysql - ${testcontainer.version} test diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/connection/OceanBaseToolsConnectProvider.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/connection/OceanBaseToolsConnectProvider.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/connection/OceanBaseToolsConnectProvider.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/connection/OceanBaseToolsConnectProvider.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchema.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchema.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchema.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchema.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/FieldSchema.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/catalog/FieldSchema.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/FieldSchema.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/catalog/FieldSchema.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSchemaFactory.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSchemaFactory.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSchemaFactory.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSchemaFactory.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSinkBuild.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSinkBuild.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSinkBuild.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSinkBuild.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseType.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseType.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseType.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseType.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseTypeMapper.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseTypeMapper.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseTypeMapper.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseTypeMapper.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/TableSchema.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/catalog/TableSchema.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/TableSchema.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/catalog/TableSchema.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/CdcTools.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/CdcTools.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/CdcTools.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/CdcTools.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/DatabaseSync.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/DatabaseSync.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/DatabaseSync.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/DatabaseSync.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/DatabaseSyncConfig.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/DatabaseSyncConfig.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/DatabaseSyncConfig.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/DatabaseSyncConfig.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/JdbcSourceSchema.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/JdbcSourceSchema.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/JdbcSourceSchema.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/JdbcSourceSchema.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/ParsingProcessFunction.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/ParsingProcessFunction.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/ParsingProcessFunction.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/ParsingProcessFunction.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/SourceConnector.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/SourceConnector.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/SourceConnector.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/SourceConnector.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/SourceSchema.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/SourceSchema.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/SourceSchema.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/SourceSchema.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2DatabaseSync.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2DatabaseSync.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2DatabaseSync.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2DatabaseSync.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2DateConverter.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2DateConverter.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2DateConverter.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2DateConverter.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2Schema.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2Schema.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2Schema.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2Schema.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2Type.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2Type.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2Type.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/db2/Db2Type.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/deserialize/OceanBaseJsonDebeziumDeserializationSchema.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/deserialize/OceanBaseJsonDebeziumDeserializationSchema.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/deserialize/OceanBaseJsonDebeziumDeserializationSchema.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/deserialize/OceanBaseJsonDebeziumDeserializationSchema.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/DateToStringConverter.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/DateToStringConverter.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/DateToStringConverter.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/DateToStringConverter.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/MysqlDatabaseSync.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/MysqlDatabaseSync.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/MysqlDatabaseSync.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/MysqlDatabaseSync.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/MysqlSchema.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/MysqlSchema.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/MysqlSchema.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/MysqlSchema.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/MysqlType.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/MysqlType.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/MysqlType.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/mysql/MysqlType.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleDatabaseSync.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleDatabaseSync.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleDatabaseSync.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleDatabaseSync.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleDateConverter.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleDateConverter.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleDateConverter.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleDateConverter.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleSchema.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleSchema.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleSchema.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleSchema.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleType.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleType.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleType.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/oracle/OracleType.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresDatabaseSync.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresDatabaseSync.java similarity index 99% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresDatabaseSync.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresDatabaseSync.java index da1f3cc..35ee3d4 100644 --- a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresDatabaseSync.java +++ b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresDatabaseSync.java @@ -186,7 +186,6 @@ public DataStreamSource buildCdcSource(StreamExecutionEnvironment env) { .deserializer(schema) .slotName(slotName) .decodingPluginName(config.get(DECODING_PLUGIN_NAME)) - .includeSchemaChanges(true) .debeziumProperties(debeziumProperties) .startupOptions(startupOptions) .splitSize(config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE)) diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresDateConverter.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresDateConverter.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresDateConverter.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresDateConverter.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresSchema.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresSchema.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresSchema.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresSchema.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresType.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresType.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresType.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/postgres/PostgresType.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerDatabaseSync.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerDatabaseSync.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerDatabaseSync.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerDatabaseSync.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerDateConverter.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerDateConverter.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerDateConverter.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerDateConverter.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerSchema.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerSchema.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerSchema.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerSchema.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerType.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerType.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerType.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/tools/cdc/sqlserver/SqlServerType.java diff --git a/flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/utils/OceanBaseToolsJdbcUtils.java b/flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/utils/OceanBaseToolsJdbcUtils.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/utils/OceanBaseToolsJdbcUtils.java rename to flink-connector-oceanbase-cli/src/main/java/com/oceanbase/connector/flink/utils/OceanBaseToolsJdbcUtils.java diff --git a/flink-connector-oceanbase-tools-cdc/src/test/java/com/oceanbase/connector/flink/CdcMysqlSyncDatabaseITCase.java b/flink-connector-oceanbase-cli/src/test/java/com/oceanbase/connector/flink/CdcMysqlSyncDatabaseITCase.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/test/java/com/oceanbase/connector/flink/CdcMysqlSyncDatabaseITCase.java rename to flink-connector-oceanbase-cli/src/test/java/com/oceanbase/connector/flink/CdcMysqlSyncDatabaseITCase.java diff --git a/flink-connector-oceanbase-tools-cdc/src/test/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchemaTest.java b/flink-connector-oceanbase-cli/src/test/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchemaTest.java similarity index 92% rename from flink-connector-oceanbase-tools-cdc/src/test/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchemaTest.java rename to flink-connector-oceanbase-cli/src/test/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchemaTest.java index 3206d93..ccd3cbb 100644 --- a/flink-connector-oceanbase-tools-cdc/src/test/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchemaTest.java +++ b/flink-connector-oceanbase-cli/src/test/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchemaTest.java @@ -35,7 +35,6 @@ import org.apache.flink.table.types.logical.FloatType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; -import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.MapType; import org.apache.flink.table.types.logical.RowType; import org.apache.flink.table.types.logical.SmallIntType; @@ -63,8 +62,7 @@ public class OceanBaseJsonSerializationSchemaTest { @Test void testCreateNotNullConverter() { - OceanBaseJsonSerializationSchemaConverterTest schema = - new OceanBaseJsonSerializationSchemaConverterTest(null); + OceanBaseJsonSerializationSchema schema = new OceanBaseJsonSerializationSchema(null); // Test Boolean assertEquals(true, schema.createNotNullConverter(new BooleanType()).convert(true)); @@ -195,18 +193,4 @@ void testCreateNotNullConverter() { "{\"field1\":\"field1\",\"field2\":123,\"field3\":true}", schema.createNotNullConverter(rowType).convert(rowData)); } - - // Create a subclass to expose the protected createNotNullConverter method - private static class OceanBaseJsonSerializationSchemaConverterTest - extends OceanBaseJsonSerializationSchema { - - public OceanBaseJsonSerializationSchemaConverterTest(TableInfo tableInfo) { - super(tableInfo); - } - - @Override - public SerializationRuntimeConverter createNotNullConverter(LogicalType type) { - return super.createNotNullConverter(type); - } - } } diff --git a/flink-connector-oceanbase-tools-cdc/src/test/java/com/oceanbase/connector/flink/tools/CdcToolsTest.java b/flink-connector-oceanbase-cli/src/test/java/com/oceanbase/connector/flink/tools/CdcToolsTest.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/test/java/com/oceanbase/connector/flink/tools/CdcToolsTest.java rename to flink-connector-oceanbase-cli/src/test/java/com/oceanbase/connector/flink/tools/CdcToolsTest.java diff --git a/flink-connector-oceanbase-tools-cdc/src/test/java/com/oceanbase/connector/flink/tools/DatabaseSyncTest.java b/flink-connector-oceanbase-cli/src/test/java/com/oceanbase/connector/flink/tools/DatabaseSyncTest.java similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/test/java/com/oceanbase/connector/flink/tools/DatabaseSyncTest.java rename to flink-connector-oceanbase-cli/src/test/java/com/oceanbase/connector/flink/tools/DatabaseSyncTest.java diff --git a/flink-connector-oceanbase-tools-cdc/src/test/resources/docker/mysql/my.cnf b/flink-connector-oceanbase-cli/src/test/resources/docker/mysql/my.cnf similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/test/resources/docker/mysql/my.cnf rename to flink-connector-oceanbase-cli/src/test/resources/docker/mysql/my.cnf diff --git a/flink-connector-oceanbase-tools-cdc/src/test/resources/log4j2-test.properties b/flink-connector-oceanbase-cli/src/test/resources/log4j2-test.properties similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/test/resources/log4j2-test.properties rename to flink-connector-oceanbase-cli/src/test/resources/log4j2-test.properties diff --git a/flink-connector-oceanbase-tools-cdc/src/test/resources/sql/mysql-cdc.sql b/flink-connector-oceanbase-cli/src/test/resources/sql/mysql-cdc.sql similarity index 100% rename from flink-connector-oceanbase-tools-cdc/src/test/resources/sql/mysql-cdc.sql rename to flink-connector-oceanbase-cli/src/test/resources/sql/mysql-cdc.sql diff --git a/pom.xml b/pom.xml index 9ed6a5e..6b57caf 100644 --- a/pom.xml +++ b/pom.xml @@ -36,16 +36,14 @@ under the License. flink-sql-connector-obkv-hbase flink-sql-connector-oceanbase flink-sql-connector-oceanbase-directload + flink-connector-oceanbase-cli flink-connector-oceanbase-e2e-tests - flink-connector-oceanbase-tools-cdc 1.5-SNAPSHOT 1.18.0 - 19.3.0.0 - 3.1.1 - 1.17.6 + 1.18 2.12 com.oceanbase.connector.flink.shaded