[improve]Improve the clarity and detail of the database synchronizati…

…on logs (#467)
apache · Aug 14, 2024 · 63b1290 · 63b1290
1 parent eb90905
commit 63b1290
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 3 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java
@@ -118,7 +118,9 @@ public void build() throws Exception {
         DorisSystem dorisSystem = new DorisSystem(options);
 
         List<SourceSchema> schemaList = getSchemaList();
-        Preconditions.checkState(!schemaList.isEmpty(), "No tables to be synchronized.");
+        Preconditions.checkState(
+                !schemaList.isEmpty(),
+                "No tables to be synchronized. Please make sure whether the tables that need to be synchronized exist in the corresponding database or schema.");
 
         if (!StringUtils.isNullOrWhitespaceOnly(database)
                 && !dorisSystem.databaseExists(database)) {

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/JdbcSourceSchema.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/JdbcSourceSchema.java
@@ -17,7 +17,11 @@
 
 package org.apache.doris.flink.tools.cdc;
 
+import org.apache.flink.util.Preconditions;
+
 import org.apache.doris.flink.catalog.doris.FieldSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.sql.DatabaseMetaData;
 import java.sql.ResultSet;
@@ -31,6 +35,7 @@
  * databases.
  */
 public abstract class JdbcSourceSchema extends SourceSchema {
+    private static final Logger LOG = LoggerFactory.getLogger(JdbcSourceSchema.class);
 
     public JdbcSourceSchema(
             DatabaseMetaData metaData,
@@ -48,7 +53,7 @@ public LinkedHashMap<String, FieldSchema> getColumnInfo(
             DatabaseMetaData metaData, String databaseName, String schemaName, String tableName)
             throws SQLException {
         LinkedHashMap<String, FieldSchema> fields = new LinkedHashMap<>();
-        //
+        LOG.debug("Starting to get column info for table: {}", tableName);
         try (ResultSet rs = metaData.getColumns(databaseName, schemaName, tableName, null)) {
             while (rs.next()) {
                 String fieldName = rs.getString("COLUMN_NAME");
@@ -63,10 +68,17 @@ public LinkedHashMap<String, FieldSchema> getColumnInfo(
                 if (rs.wasNull()) {
                     scale = null;
                 }
-                String dorisTypeStr = convertToDorisType(fieldType, precision, scale);
+                String dorisTypeStr = null;
+                try {
+                    dorisTypeStr = convertToDorisType(fieldType, precision, scale);
+                } catch (UnsupportedOperationException e) {
+                    throw new UnsupportedOperationException(e + " in table: " + tableName);
+                }
                 fields.put(fieldName, new FieldSchema(fieldName, dorisTypeStr, comment));
             }
         }
+        Preconditions.checkArgument(!fields.isEmpty(), "The column info of {} is empty", tableName);
+        LOG.debug("Successfully retrieved column info for table: {}", tableName);
         return fields;
     }