[oracle]#1244 Oracle URL format is inconsistent with the debezium default format

Author: dujie

flink-connector-oracle-cdc/src/main/java/com/ververica/cdc/connectors/oracle/OracleSource.java

@@ -170,6 +170,17 @@ public DebeziumSourceFunction<T> build() {
                 props.putAll(dbzProperties);
             }
 
+            String hostname = props.getProperty("database.hostname");
+            String port = props.getProperty("database.port");
+            String dbname = props.getProperty("database.dbname");
+
+            String url;
+            if (!props.containsKey("database.url")) {
+                url = "jdbc:oracle:thin:@" + hostname + ":" + port + ":" + dbname;
+                // debezium default URL format is SID rather than serviceName， so we should set url
+                props.setProperty("database.url", url);
+            }
+
             return new DebeziumSourceFunction<>(
                     deserializer, props, specificOffset, new OracleValidator(props));
         }

flink-connector-oracle-cdc/src/main/java/com/ververica/cdc/connectors/oracle/OracleValidator.java

@@ -65,12 +65,17 @@ public void validate() {
 
     public static Connection openConnection(Properties properties) throws SQLException {
         DriverManager.registerDriver(new oracle.jdbc.OracleDriver());
-        String hostname = properties.getProperty("database.hostname");
-        String port = properties.getProperty("database.port");
-        String dbname = properties.getProperty("database.dbname");
+        String url;
+        if (properties.containsKey("database.url")) {
+            url = properties.getProperty("database.url");
+        } else {
+            String hostname = properties.getProperty("database.hostname");
+            String port = properties.getProperty("database.port");
+            String dbname = properties.getProperty("database.dbname");
+            url = "jdbc:oracle:thin:@" + hostname + ":" + port + ":" + dbname;
+        }
         String userName = properties.getProperty("database.user");
         String userpwd = properties.getProperty("database.password");
-        return DriverManager.getConnection(
-                "jdbc:oracle:thin:@" + hostname + ":" + port + ":" + dbname, userName, userpwd);
+        return DriverManager.getConnection(url, userName, userpwd);
     }
 }