[FLINK-39143][pipeline-connecotr][Fluss]Optimize the connection of fluss when use FlussMetaDataApplier (#4282)

Co-authored-by: Thorne <syyfffy@163.com>

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,
@@ -89,22 +92,22 @@ public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
     @Override
     public void applySchemaChange(SchemaChangeEvent schemaChangeEvent) {
         LOG.info("fluss metadata applier receive schemaChangeEvent {}", schemaChangeEvent);
+        Admin admin = getAdmin();
         if (schemaChangeEvent instanceof CreateTableEvent) {
             CreateTableEvent createTableEvent = (CreateTableEvent) schemaChangeEvent;
-            applyCreateTable(createTableEvent);
+            applyCreateTable(admin, createTableEvent);
         } else if (schemaChangeEvent instanceof DropTableEvent) {
             DropTableEvent dropTableEvent = (DropTableEvent) schemaChangeEvent;
-            applyDropTable(dropTableEvent);
+            applyDropTable(admin, dropTableEvent);
         } else {
             throw new IllegalArgumentException(
                     "fluss metadata applier only support CreateTableEvent now but receives "
                             + schemaChangeEvent);
         }
     }
 
-    private void applyCreateTable(CreateTableEvent event) {
-        try (Connection connection = ConnectionFactory.createConnection(flussClientConfig);
-                Admin admin = connection.getAdmin()) {
+    private void applyCreateTable(Admin admin, CreateTableEvent event) {
+        try {
             TableId tableId = event.tableId();
             TablePath tablePath = new TablePath(tableId.getSchemaName(), tableId.getTableName());
             String tableIdentifier = tablePath.getDatabaseName() + "." + tablePath.getTableName();
@@ -126,9 +129,8 @@ private void applyCreateTable(CreateTableEvent event) {
         }
     }
 
-    private void applyDropTable(DropTableEvent event) {
-        try (Connection connection = ConnectionFactory.createConnection(flussClientConfig);
-                Admin admin = connection.getAdmin()) {
+    private void applyDropTable(Admin admin, DropTableEvent event) {
+        try {
             TableId tableId = event.tableId();
             TablePath tablePath = new TablePath(tableId.getSchemaName(), tableId.getTableName());
             admin.dropTable(tablePath, true).get();
@@ -138,6 +140,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()