[FLINK-3640] Close MetadataApplier when the job stops

apache · Sep 29, 2024 · 5761577 · 5761577
1 parent 4b13c49
commit 5761577
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 1 deletion.
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/MetadataApplier.java b/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 {}
 }
diff --git a/...imon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java b/...imon/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())) {

diff --git a/...g/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java b/...g/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) {