Optimize the connection of fluss when use FlussMetaDataApplier

Author: ThorneANN

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java

@@ -58,6 +58,9 @@ public class FlussMetaDataApplier implements MetadataApplier {
     private Set<SchemaChangeEventType> enabledEventTypes =
             new HashSet<>(Arrays.asList(CREATE_TABLE, DROP_TABLE));
 
+    private transient Connection connection;
+    private transient Admin admin;
+
     public FlussMetaDataApplier(
             Configuration flussClientConfig,
             Map<String, String> tableProperties,
@@ -103,8 +106,8 @@ public void applySchemaChange(SchemaChangeEvent schemaChangeEvent) {
     }
 
     private void applyCreateTable(CreateTableEvent event) {
-        try (Connection connection = ConnectionFactory.createConnection(flussClientConfig);
-                Admin admin = connection.getAdmin()) {
+        try {
+            Admin admin = getAdmin();
             TableId tableId = event.tableId();
             TablePath tablePath = new TablePath(tableId.getSchemaName(), tableId.getTableName());
             String tableIdentifier = tablePath.getDatabaseName() + "." + tablePath.getTableName();
@@ -127,8 +130,8 @@ private void applyCreateTable(CreateTableEvent event) {
     }
 
     private void applyDropTable(DropTableEvent event) {
-        try (Connection connection = ConnectionFactory.createConnection(flussClientConfig);
-                Admin admin = connection.getAdmin()) {
+        try {
+            Admin admin = getAdmin();
             TableId tableId = event.tableId();
             TablePath tablePath = new TablePath(tableId.getSchemaName(), tableId.getTableName());
             admin.dropTable(tablePath, true).get();
@@ -138,6 +141,24 @@ private void applyDropTable(DropTableEvent event) {
         }
     }
 
+    private Admin getAdmin() {
+        if (connection == null) {
+            connection = ConnectionFactory.createConnection(flussClientConfig);
+            admin = connection.getAdmin();
+        }
+        return admin;
+    }
+
+    @Override
+    public void close() throws Exception {
+        if (admin != null) {
+            admin.close();
+        }
+        if (connection != null) {
+            connection.close();
+        }
+    }
+
     private void sanityCheck(TableDescriptor inferredFlussTable, TableInfo currentTableInfo) {
         List<String> inferredPrimaryKeyColumnNames =
                 inferredFlussTable.getSchema().getPrimaryKeyColumnNames().stream()