[cdc-connector][oracle] Use database name in upper case at the snapsh…

…ot phase to fix fetching two different db name for the same table at the snapshot phase and incremental phase(apache#2088)
joyCurry30 · Mar 22, 2024 · 5836663 · 5836663
1 parent 22512ee
commit 5836663
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 1 deletion.
diff --git a/...cdc/src/main/java/com/ververica/cdc/connectors/oracle/table/OracleTableSourceFactory.java b/...cdc/src/main/java/com/ververica/cdc/connectors/oracle/table/OracleTableSourceFactory.java
@@ -56,6 +56,7 @@
 import static com.ververica.cdc.connectors.oracle.source.config.OracleSourceOptions.SCHEMA_NAME;
 import static com.ververica.cdc.connectors.oracle.source.config.OracleSourceOptions.URL;
 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;
 
 /** Factory for creating configured instance of {@link OracleTableSource}. */
@@ -75,6 +76,16 @@ public DynamicTableSource createDynamicTableSource(Context context) {
         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/ververica/flink-cdc-connectors/pull/2088.
+        databaseName = databaseName.toUpperCase();
         String tableName = config.get(TABLE_NAME);
         String schemaName = config.get(SCHEMA_NAME);
         int port = config.get(PORT);

diff --git a/...src/test/java/com/ververica/cdc/connectors/oracle/table/OracleTableSourceFactoryTest.java b/...src/test/java/com/ververica/cdc/connectors/oracle/table/OracleTableSourceFactoryTest.java
@@ -83,7 +83,7 @@ public class OracleTableSourceFactoryTest {
     private static final String MY_LOCALHOST = "localhost";
     private static final String MY_USERNAME = "flinkuser";
     private static final String MY_PASSWORD = "flinkpw";
-    private static final String MY_DATABASE = "myDB";
+    private static final String MY_DATABASE = "MYDB";
     private static final String MY_TABLE = "myTable";
     private static final String MY_SCHEMA = "mySchema";
     private static final Properties PROPERTIES = new Properties();