address comments

Author: LiebingYu

fluss-common/src/main/java/org/apache/fluss/metadata/DatabaseChange.java

@@ -175,7 +175,7 @@ public int hashCode() {
 
         @Override
         public String toString() {
-            return "SetComment{" + "comment='" + comment + '\'' + '}';
+            return "UpdateComment{" + "comment='" + comment + '\'' + '}';
         }
     }
 }

fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java

@@ -150,16 +150,17 @@ public void alterDatabaseProperties(
                 throw new DatabaseNotExistException("Database " + databaseName + " not exists.");
             }
 
-            DatabaseInfo databaseInfo = getDatabase(databaseName);
-            DatabaseDescriptor currentDescriptor = databaseInfo.getDatabaseDescriptor();
+            DatabaseRegistration databaseRegistration = getDatabaseRegistration(databaseName);
+            DatabaseDescriptor currentDescriptor = databaseRegistration.toDatabaseDescriptor();
 
             // Create updated descriptor
             DatabaseDescriptor newDescriptor =
                     getUpdatedDatabaseDescriptor(currentDescriptor, databasePropertyChanges);
 
             if (newDescriptor != null) {
                 // Update the database in ZooKeeper
-                DatabaseRegistration updatedRegistration = DatabaseRegistration.of(newDescriptor);
+                DatabaseRegistration updatedRegistration =
+                        databaseRegistration.newProperties(newDescriptor);
                 zookeeperClient.updateDatabase(databaseName, updatedRegistration);
                 LOG.info("Successfully altered database properties for database: {}", databaseName);
             } else {
@@ -189,16 +190,13 @@ private DatabaseDescriptor getUpdatedDatabaseDescriptor(
         // reset properties
         newCustomProperties.keySet().removeAll(changes.customPropertiesToReset);
 
-        boolean hasCommentChange =
-                !currentDescriptor.getComment().equals(Optional.ofNullable(changes.commentToSet));
-
         if (newCustomProperties.equals(currentDescriptor.getCustomProperties())
-                && !hasCommentChange) {
+                && !changes.isCommentChanged()) {
             return null;
         }
 
         String newComment =
-                hasCommentChange
+                changes.isCommentChanged()
                         ? changes.commentToSet
                         : currentDescriptor.getComment().orElse(null);
 
@@ -209,7 +207,15 @@ private DatabaseDescriptor getUpdatedDatabaseDescriptor(
     }
 
     public DatabaseInfo getDatabase(String databaseName) throws DatabaseNotExistException {
+        DatabaseRegistration databaseReg = getDatabaseRegistration(databaseName);
+        return new DatabaseInfo(
+                databaseName,
+                databaseReg.toDatabaseDescriptor(),
+                databaseReg.createdTime,
+                databaseReg.modifiedTime);
+    }
 
+    public DatabaseRegistration getDatabaseRegistration(String databaseName) {
         Optional<DatabaseRegistration> optionalDB;
         try {
             optionalDB = zookeeperClient.getDatabase(databaseName);
@@ -221,13 +227,7 @@ public DatabaseInfo getDatabase(String databaseName) throws DatabaseNotExistExce
         if (!optionalDB.isPresent()) {
             throw new DatabaseNotExistException("Database '" + databaseName + "' does not exist.");
         }
-
-        DatabaseRegistration databaseReg = optionalDB.get();
-        return new DatabaseInfo(
-                databaseName,
-                databaseReg.toDatabaseDescriptor(),
-                databaseReg.createdTime,
-                databaseReg.modifiedTime);
+        return optionalDB.get();
     }
 
     public boolean databaseExists(String databaseName) {

fluss-server/src/main/java/org/apache/fluss/server/entity/DatabasePropertyChanges.java

@@ -29,15 +29,24 @@ public class DatabasePropertyChanges {
 
     public final Map<String, String> customPropertiesToSet;
     public final Set<String> customPropertiesToReset;
+
     public final String commentToSet;
 
+    private final boolean commentChanged;
+
     protected DatabasePropertyChanges(
             Map<String, String> customPropertiesToSet,
             Set<String> customPropertiesToReset,
-            @Nullable String commentToSet) {
+            @Nullable String commentToSet,
+            boolean commentChanged) {
         this.customPropertiesToSet = customPropertiesToSet;
         this.customPropertiesToReset = customPropertiesToReset;
         this.commentToSet = commentToSet;
+        this.commentChanged = commentChanged;
+    }
+
+    public boolean isCommentChanged() {
+        return commentChanged;
     }
 
     public static DatabasePropertyChanges.Builder builder() {
@@ -50,6 +59,7 @@ public static class Builder {
         private final Set<String> customPropertiesToReset = new HashSet<>();
 
         private String commentToSet = null;
+        private boolean commentChanged = false;
 
         public void setCustomProperty(String key, String value) {
             customPropertiesToSet.put(key, value);
@@ -59,13 +69,14 @@ public void resetCustomProperty(String key) {
             customPropertiesToReset.add(key);
         }
 
-        public void setComment(String comment) {
+        public void setComment(@Nullable String comment) {
             this.commentToSet = comment;
+            this.commentChanged = true;
         }
 
         public DatabasePropertyChanges build() {
             return new DatabasePropertyChanges(
-                    customPropertiesToSet, customPropertiesToReset, commentToSet);
+                    customPropertiesToSet, customPropertiesToReset, commentToSet, commentChanged);
         }
     }
 }

fluss-server/src/main/java/org/apache/fluss/server/zk/data/DatabaseRegistration.java

@@ -54,6 +54,14 @@ public DatabaseDescriptor toDatabaseDescriptor() {
         return builder.build();
     }
 
+    public DatabaseRegistration newProperties(DatabaseDescriptor databaseDescriptor) {
+        return new DatabaseRegistration(
+                databaseDescriptor.getComment().orElse(null),
+                databaseDescriptor.getCustomProperties(),
+                createdTime,
+                System.currentTimeMillis());
+    }
+
     public static DatabaseRegistration of(DatabaseDescriptor databaseDescriptor) {
         final long currentMillis = System.currentTimeMillis();
         return new DatabaseRegistration(