[FLINK-36406] Close MetadataApplier when the job stops

[morozov]

This pull request implements closing MetadataApplier when the job stops. This allows MetadataApplier implementations to release the resources that they may use to apply schema changes (e.g. a JDBC connection).

Choice of the interface

1. SchemaRegistryRequestHandler (which will close MetadataApplier) implements Closeable, so it would make sense to declare MetadataApplier as Closeable as well.
2. PaimonMetadataApplier internally instantiates a Paimon Catalog, which it ideally should close, and which is Autocloseable.

To me, it makes more sense to declare MetadataApplier as AutoCloseable. The reason is that Closeable is an IO-specific interface, which is declared in java.io and throws IOException in close(). AutoCloseable is more generic and seems to be more suitable for MetadataApplier, whose implementations doesn't necessarily perform IO.

Testing

I didn't find existing tests focusing on SchemaRegistryRequestHandler or SchemaRegistry, so I dind't implement a unit test. Testing of the changes in SchemaRegistryRequestHandler can be done by implementing some logging in ValuesMetadataApplier#close and running any test that uses it (e.g. FlinkPipelineComposerITCase)

Additional scope

Just as a demonstration of where this new functionality can be applied, I implemented close() in PaimonMetadataApplier. I didn't test it and am not sure that this is the right thing to do. I can revert this part, if necessary.

Further considerations

The problem of closing resources seems also relevant for the MetadataAccessor interface. Specifically, MySqlMetadataAccessor disconnects from the database after each call but some other implementations may require maintaining a persistent connection.

[yuxiqian]

Thanks for @morozov's great work! Just left some minor comments.

[yuxiqian]

Adding a JavaDocs for this method would be nice.

[yuxiqian]

I've tested locally with a dummy MetadataApplier that always throws exception but gets no output. Seems OperatorCoordinators will be closed by calling OperatorCoordinatorHandler#disposeAllOperatorCoordinators, where exceptions thrown here are ignored without being logged:

// DefaultOperatorCoordinatorHandler.java
@Override
public void disposeAllOperatorCoordinators() {
    coordinatorMap.values().forEach(IOUtils::closeQuietly);
}

// IOUtils.java
public static void closeQuietly(AutoCloseable closeable) {
    try {
        if (closeable != null) {
            closeable.close();
        }
    } catch (Throwable ignored) {
    }
}

Maybe we can add some logging message if metadata appliers are not closed as expected?

[morozov]

Good catch 👍 Thanks for the detailed explanation.

[yuxiqian]

Looks good! cc @ruanhang1993 @lvyanquan