[hotfix] Fix data lake catalog connection resource is not closed in flink catalog. (#2087)

Author: zhuangchong

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

@@ -27,7 +27,7 @@
 public interface TableChange {
 
     /**
-     * A table change toadd the column with specified position.
+     * A table change to add the column with specified position.
      *
      * <p>It is equal to the following statement:
      *

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java

@@ -158,6 +158,7 @@ public void open() throws CatalogException {
     public void close() throws CatalogException {
         IOUtils.closeQuietly(admin, "fluss-admin");
         IOUtils.closeQuietly(connection, "fluss-connection");
+        IOUtils.closeQuietly(lakeFlinkCatalog, "fluss-lake-catalog");
     }
 
     @Override

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeFlinkCatalog.java

@@ -36,7 +36,7 @@
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
 
 /** A lake flink catalog to delegate the operations on lake table. */
-public class LakeFlinkCatalog {
+public class LakeFlinkCatalog implements AutoCloseable {
 
     private final String catalogName;
     private final ClassLoader classLoader;
@@ -93,6 +93,13 @@ public DataLakeFormat getLakeFormat() {
         return lakeFormat;
     }
 
+    @Override
+    public void close() throws Exception {
+        if (catalog != null) {
+            catalog.close();
+        }
+    }
+
     /**
      * Factory for creating Paimon Catalog instances.
      *