[FLINK-39590] connection catalog manager changes

[lihaosky]

What is the purpose of the change

This pull request adds first-class support for catalog connections in the Table API. A Connection is a new top-level catalog object (alongside tables, views, functions, and models) that captures the configuration for accessing an external system, so that downstream features can reuse a centrally managed connection definition rather than embedding connection properties in every table DDL.

This change introduces the connection lifecycle on the catalog stack — interface methods, in-memory implementation, plumbing through CatalogManager and TableEnvironmentImpl, listener events, exception types, and a connection factory — without altering existing catalog object behavior.

Brief change log

• Added connection CRUD APIs (listConnections, getConnection, connectionExists, createConnection, dropConnection, alterConnection) to Catalog as default methods, preserving binary/source compatibility for existing implementations.
• Implemented connection storage in GenericInMemoryCatalog.
• Plumbed connection management (including temporary connections) through CatalogManager with overloads for ObjectIdentifier-based lookup, listing, and create/alter/drop.
• Wired CatalogManager to expose a ConnectionFactory and routed connection access from TableEnvironmentImpl.
• Added catalog listener events: CreateConnectionEvent, AlterConnectionEvent, DropConnectionEvent, all extending a new ConnectionModificationEvent.
• Added new exception types: ConnectionNotExistException, ConnectionAlreadyExistException.
• Added ConnectionFactory SPI and DefaultConnectionFactory implementation.

Verifying this change

Please make sure both new and modified tests in this PR follow the conventions for tests defined in our code quality guide.

This change added tests and can be verified as follows:

• Added CatalogTest (abstract test base, ~220 lines) covering listConnections, getConnection, connectionExists, createConnection (incl. ifNotExists semantics and ConnectionAlreadyExistException), dropConnection (incl. ifNotExists semantics and ConnectionNotExistException), and alterConnection.
• GenericInMemoryCatalogTest extends CatalogTest and exercises the new methods against the in-memory implementation (123 tests passing locally).

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): yes — Catalog is @PublicEvolving. New methods are default, so existing implementations remain source- and binary-compatible.
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? JavaDocs on the new Catalog methods, listener events, exception types, and the ConnectionFactory SPI.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Claude

[flinkbot]

CI report:

• 22a06d7 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build