[INLONG-10704][Sort] Sort Add Oceanbase Support

[xxsc0529]

Fixes #10704

Motivation

Added support for Oceanbase data sources to adapt to the Oceanbase ecosystem. close #10704

Modifications

It mainly adds oceanbase data source support and uses the oceanbase-client driver to connect.

It mainly adds oceanbase data source support and uses the oceanbase-client driver to connect. And write oceanbase's ExtractNode logic and LoadNode logic. The specific implementation logic is the same as mysql.Added support for oceanbase on the front-end page. Since the jdbc-inlong processing logic in the flink15 connection lacks the jdbc dialect, an error is reported. This is modified here.

Verifying this change

(Please pick either of the following options)

• This change is a trivial rework/code cleanup without any test coverage.

• This change is already covered by existing tests, such as:
  (please describe tests)
  Here's how the page test works
  
  
  
  

• This change added tests and can be verified as follows:
  Please use the following test classes for testing org.apache.inlong.sort.parser.MySqlExtractNodeToOceanBaseLoadNodeTest
  (example:)

  • Added integration tests for end-to-end deployment with large payloads (10MB)
  • Extended integration test for recovery after broker failure

Documentation

• Does this pull request introduce a new feature? (yes / no) yes
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented) Data source types have been added to the README.md
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create a follow-up issue for adding the documentation

[XiaoYou201]

This factory should named OceanBaseDialectFactory?

[xxsc0529]

No, it was found in the test that MySQL and Oceanbase both need this dialect to find the corresponding factory that conforms to the JDBC protocol

[EMsnap]

please add test result with real data from any source to oceanbase

[xxsc0529]

please add test result with real data from any source to oceanbase

What kind of test data is needed and where to put it

[XiaoYou201]

please add test result with real data from any source to oceanbase

What kind of test data is needed and where to put it

You can refer to inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/Mysql2StarRocksTest.java

or package your code to manager container. Then start a job to verify the oceanbase connector work properly.If all right, plz
provide some screenshots~

[xxsc0529]

please add test result with real data from any source to oceanbase

What kind of test data is needed and where to put it

You can refer to inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/Mysql2StarRocksTest.java

or package your code to manager container. Then start a job to verify the oceanbase connector work properly.If all right, plz provide some screenshots~

The logic of Oceanbase and MySQL is the same, using MySQL-CDC, and the following is tested in the form of a remote debug project jar package through local source code

This step shows that the pre-configuration logic is passable, and the corresponding CDC can be found and flinksql can be generated，Below is the generated flinksql
FlinkSqlParseResult(tableEnv=org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl@42554211, createTableSqls=[CREATE TABLE table_mysqtooceanbase(
id INT,
user_name STRING)
WITH (
'inlong.metric.labels' = 'groupId=mysqtooceanbase&streamId=mysqtooceanbase&nodeId=mysqtooceanbase',
'topic' = 'mysqtooceanbase.mysqtooceanbase',
'properties.bootstrap.servers' = '127.0.0.1:9092',
'connector' = 'kafka-inlong',
'inlong-msg.debezium-json.schema-include' = 'false',
'inlong-msg.debezium-json.encode.decimal-as-plain-number' = 'true',
'inlong-msg.inner.format' = 'debezium-json',
'format' = 'inlong-msg',
'inlong-msg.debezium-json.timestamp-format.standard' = 'SQL',
'inlong-msg.debezium-json.ignore-parse-errors' = 'true',
'inlong-msg.debezium-json.map-null-key.literal' = 'null',
'inlong-msg.debezium-json.map-null-key.mode' = 'DROP',
'inlong-msg.ignore-parse-errors' = 'false',
'scan.startup.mode' = 'earliest-offset',
'properties.group.id' = 'kafka-local_mysqtooceanbase.mysqtooceanbase_consumer_group'
), CREATE TABLE table_oceanbase(
PRIMARY KEY (id) NOT ENFORCED,
id INT,
user_name STRING)
WITH (
'inlong.metric.labels' = 'groupId=mysqtooceanbase&streamId=mysqtooceanbase&nodeId=oceanbase',
'connector' = 'jdbc-inlong',
'url' = 'jdbc:mysql://Confidentiality:4848/test',
'username' = 'Confidentiality@tt1',
'password' = '123456',
'table-name' = 't_ds_user'
)], loadSqls=[INSERT INTO table_oceanbase
SELECT
id AS id,
user_name AS user_name
FROM table_mysqtooceanbase ])
Go to flink-ui to find that the task has also been submitted

[XiaoYou201]

LGTM~