From 205e48f53f24086419c01dd8652748347ecd295f Mon Sep 17 00:00:00 2001
From: fengli <ldliulsy@163.com>
Date: Fri, 31 Dec 2021 23:25:04 +0800
Subject: [PATCH] [HUDI-3132] Minor fixes for HoodieCatalog

close apache/hudi#4486
---
 .../java/org/apache/hudi/table/catalog/HoodieCatalog.java    | 4 +++-
 .../org/apache/hudi/table/catalog/TestHoodieCatalog.java     | 5 +++++
 2 files changed, 8 insertions(+), 1 deletion(-)

diff --git a/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieCatalog.java b/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieCatalog.java
index 3ad3deab8..fddaaba66 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieCatalog.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieCatalog.java
@@ -305,7 +305,9 @@ public class HoodieCatalog extends AbstractCatalog {
     try {
       StreamerUtil.initTableIfNotExists(conf);
       // prepare the non-table-options properties
-      options.put(TableOptionProperties.COMMENT, resolvedTable.getComment());
+      if (!StringUtils.isNullOrWhitespaceOnly(resolvedTable.getComment())) {
+        options.put(TableOptionProperties.COMMENT, resolvedTable.getComment());
+      }
       TableOptionProperties.createProperties(tablePathStr, hadoopConf, options);
     } catch (IOException e) {
       throw new CatalogException(String.format("Initialize table path %s exception.", tablePathStr), e);
diff --git a/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalog.java b/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalog.java
index 18a2369ff..67d3f1899 100644
--- a/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalog.java
+++ b/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalog.java
@@ -37,6 +37,7 @@ import org.apache.flink.table.catalog.ResolvedCatalogTable;
 import org.apache.flink.table.catalog.ResolvedSchema;
 import org.apache.flink.table.catalog.UniqueConstraint;
 import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
 import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
 import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
 import org.apache.flink.table.catalog.exceptions.TableNotExistException;
@@ -158,6 +159,10 @@ public class TestHoodieCatalog {
     assertTrue(catalog.listDatabases().contains("db1"));
     assertEquals(expected.getProperties(), actual.getProperties());
 
+    // create exist database
+    assertThrows(DatabaseAlreadyExistException.class,
+        () -> catalog.createDatabase("db1", expected, false));
+
     // drop exist database
     catalog.dropDatabase("db1", true);
     assertFalse(catalog.listDatabases().contains("db1"));