[FLINK-36406][cdc-runtime] Close MetadataApplier when the job stops

This closes  #3623

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/MetadataApplier.java

@@ -30,7 +30,7 @@ import java.util.stream.Collectors;
 
 /** {@code MetadataApplier} is used to apply metadata changes to external systems. */
 @PublicEvolving
-public interface MetadataApplier extends Serializable {
+public interface MetadataApplier extends Serializable, AutoCloseable {
 
     /** Apply the given {@link SchemaChangeEvent} to external systems. */
     void applySchemaChange(SchemaChangeEvent schemaChangeEvent) throws SchemaEvolveException;
@@ -50,4 +50,8 @@ public interface MetadataApplier extends Serializable {
     default Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
         return Arrays.stream(SchemaChangeEventTypeFamily.ALL).collect(Collectors.toSet());
     }
+
+    /** Closes the metadata applier and its underlying resources. */
+    @Override
+    default void close() throws Exception {}
 }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java

@@ -153,6 +153,13 @@ public class PaimonMetadataApplier implements MetadataApplier {
                 });
     }
 
+    @Override
+    public void close() throws Exception {
+        if (catalog != null) {
+            catalog.close();
+        }
+    }
+
     private void applyCreateTable(CreateTableEvent event) throws SchemaEvolveException {
         try {
             if (!catalog.databaseExists(event.tableId().getSchemaName())) {