use 'compatible mode' instead of 'database mode'

Author: whhe

flink-connector-oceanbase-cdc/src/main/java/com/ververica/cdc/connectors/oceanbase/source/OceanBaseConnection.java

@@ -40,7 +40,7 @@ public class OceanBaseConnection extends JdbcConnection {
     private static final String OB_URL_PATTERN =
             "jdbc:oceanbase://${hostname}:${port}/?connectTimeout=${connectTimeout}&serverTimezone=${serverTimezone}";
 
-    private String databaseMode;
+    private String compatibleMode;
 
     public OceanBaseConnection(
             String hostname,
@@ -108,16 +108,16 @@ private static JdbcConnection.ConnectionFactory factory(
     }
 
     /**
-     * Get the database mode of this connection, should be 'MySQL' or 'Oracle'.
+     * Get the compatible mode of this connection, should be 'MySQL' or 'Oracle'.
      *
-     * @return The database mode.
+     * @return The compatible mode.
      * @throws SQLException If a database access error occurs.
      */
-    public String getDatabaseMode() throws SQLException {
-        if (databaseMode == null) {
-            databaseMode = connection().getMetaData().getDatabaseProductName();
+    public String getCompatibleMode() throws SQLException {
+        if (compatibleMode == null) {
+            compatibleMode = connection().getMetaData().getDatabaseProductName();
         }
-        return databaseMode;
+        return compatibleMode;
     }
 
     /**
@@ -131,7 +131,7 @@ public String getDatabaseMode() throws SQLException {
     public List<String> getTables(String dbPattern, String tbPattern) throws SQLException {
         List<String> result = new ArrayList<>();
         DatabaseMetaData metaData = connection().getMetaData();
-        switch (getDatabaseMode().toLowerCase()) {
+        switch (getCompatibleMode().toLowerCase()) {
             case "mysql":
                 List<String> dbNames = getResultList(metaData.getCatalogs(), "TABLE_CAT");
                 dbNames =
@@ -165,7 +165,8 @@ public List<String> getTables(String dbPattern, String tbPattern) throws SQLExce
                 }
                 break;
             default:
-                throw new FlinkRuntimeException("Unsupported database mode: " + getDatabaseMode());
+                throw new FlinkRuntimeException(
+                        "Unsupported database mode: " + getCompatibleMode());
         }
         return result;
     }

flink-connector-oceanbase-cdc/src/main/java/com/ververica/cdc/connectors/oceanbase/source/OceanBaseRichSourceFunction.java

@@ -227,7 +227,9 @@ private void initTableWhiteList() {
                 && StringUtils.isNotBlank(databaseName)
                 && StringUtils.isNotBlank(tableName)) {
             try {
-                LOG.info("Connection database mode: {}", getSnapshotConnection().getDatabaseMode());
+                LOG.info(
+                        "Connection database mode: {}",
+                        getSnapshotConnection().getCompatibleMode());
                 List<String> tables = getSnapshotConnection().getTables(databaseName, tableName);
                 LOG.info("Pattern matched tables: {}", tables);
                 localTableSet.addAll(tables);
@@ -263,7 +265,7 @@ private void readSnapshotRecordsByTable(String databaseName, String tableName) {
                 new OceanBaseRecord.SourceInfo(
                         tenantName, databaseName, tableName, resolvedTimestamp);
         try {
-            String databaseMode = getSnapshotConnection().getDatabaseMode();
+            String databaseMode = getSnapshotConnection().getCompatibleMode();
             String fullName;
             if ("mysql".equalsIgnoreCase(databaseMode)) {
                 fullName = String.format("`%s`.`%s`", databaseName, tableName);