[FLINK-36406] Close MetadataApplier when the job stops

Author: morozov

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

@@ -30,7 +30,7 @@
 
 /** {@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,7 @@ default boolean acceptsSchemaEvolutionType(SchemaChangeEventType schemaChangeEve
     default Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
         return Arrays.stream(SchemaChangeEventTypeFamily.ALL).collect(Collectors.toSet());
     }
+
+    @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

@@ -148,6 +148,12 @@ public void applySchemaChange(SchemaChangeEvent schemaChangeEvent)
                 });
     }
 
+    public void close() throws Exception {
+        if (catalog != null) {
+            catalog.close();
+        }
+    }
+
     private void applyCreateTable(CreateTableEvent event) throws SchemaEvolveException {
         try {
             if (!catalog.databaseExists(event.tableId().getSchemaName())) {

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java

@@ -348,6 +348,12 @@ public void close() throws IOException {
         if (schemaChangeThreadPool != null) {
             schemaChangeThreadPool.shutdown();
         }
+
+        try {
+            metadataApplier.close();
+        } catch (Exception e) {
+            throw new IOException("Failed to close metadata applier.", e);
+        }
     }
 
     private List<SchemaChangeEvent> calculateDerivedSchemaChangeEvents(SchemaChangeEvent event) {