[inlong] branch master updated: [INLONG-8558][Sort] Use database name in upper case at the OracleTableSourceFactory (#8559)

[dockerzhang]

This is an automated email from the ASF dual-hosted git repository.

dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git


The following commit(s) were added to refs/heads/master by this push:
     new 27144fdcce [INLONG-8558][Sort] Use database name in upper case at the 
OracleTableSourceFactory (#8559)
27144fdcce is described below

commit 27144fdcced51620aadfd04038db643681735431
Author: emhui <111486498+e-m...@users.noreply.github.com>
AuthorDate: Thu Jul 20 18:59:21 2023 +0800

    [INLONG-8558][Sort] Use database name in upper case at the 
OracleTableSourceFactory (#8559)
---
 .../sort/cdc/oracle/source/utils/OracleConnectionUtils.java       | 2 +-
 .../inlong/sort/cdc/oracle/table/OracleTableSourceFactory.java    | 8 ++++++++
 2 files changed, 9 insertions(+), 1 deletion(-)

diff --git 
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/utils/OracleConnectionUtils.java
 
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/utils/OracleConnectionUtils.java
index 860676396b..f592fed582 100644
--- 
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/utils/OracleConnectionUtils.java
+++ 
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/utils/OracleConnectionUtils.java
@@ -100,7 +100,7 @@ public class OracleConnectionUtils {
                             String schemaName = rs.getString(1);
                             String tableName = rs.getString(2);
                             TableId tableId =
-                                    new 
TableId(jdbcConnection.database().toUpperCase(), schemaName, tableName);
+                                    new TableId(jdbcConnection.database(), 
schemaName, tableName);
                             tableIdSet.add(tableId);
                         }
                     });
diff --git 
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleTableSourceFactory.java
 
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleTableSourceFactory.java
index 892e9df6ca..a7328b2e6f 100644
--- 
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleTableSourceFactory.java
+++ 
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleTableSourceFactory.java
@@ -53,6 +53,7 @@ import static 
com.ververica.cdc.connectors.base.options.SourceOptions.SPLIT_KEY_
 import static 
com.ververica.cdc.connectors.base.utils.ObjectUtils.doubleCompare;
 import static 
com.ververica.cdc.connectors.oracle.source.config.OracleSourceOptions.PORT;
 import static 
com.ververica.cdc.debezium.table.DebeziumOptions.getDebeziumProperties;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 import static org.apache.flink.util.Preconditions.checkState;
 import static org.apache.inlong.sort.base.Constants.AUDIT_KEYS;
 import static org.apache.inlong.sort.base.Constants.INLONG_AUDIT;
@@ -77,6 +78,13 @@ public class OracleTableSourceFactory implements 
DynamicTableSourceFactory {
         String username = config.get(USERNAME);
         String password = config.get(PASSWORD);
         String databaseName = config.get(DATABASE_NAME);
+        checkNotNull(databaseName);
+        // During the incremental phase, Debezium uses the uppercase database 
name.
+        // However, during the snapshot phase, the database name is 
user-configurable.
+        // To avoid inconsistencies between the database names in the snapshot 
and incremental phases,
+        // it is necessary to convert the database name to uppercase when 
constructing the Oracle Source.
+        // For more details, please refer to: 
https://github.com/apache/inlong/pull/7865.
+        databaseName = databaseName.toUpperCase();
         String tableName = config.get(TABLE_NAME);
         String schemaName = config.get(SCHEMA_NAME);
         int port = config.get(PORT);