Manage Environment Catalogs in Confluent Manager for Apache Flink

Starting with version 2.4.0, Confluent Manager for Apache Flink® (CMF) provides an environment catalog for every environment. The environment catalog is a writable Flink SQL catalog that stores metadata you create with Data Definition Language (DDL) statements, such as user-defined functions and tables that use custom connectors.

The built-in Kafka catalogs cannot store this metadata. The catalogs expose Kafka topics as tables and derive their schemas from Schema Registry, but they do not accept CREATE FUNCTION, and they cannot store tables that are backed by systems other than Kafka. The environment catalog fills that gap. You register functions and custom-connector tables in it. Because the catalog is scoped to a single environment, the objects you create are visible only within that environment.

Note

The environment catalog stores metadata only. You deliver the function and connector code by uploading it to artifact management and referencing it from your DDL.

How it works

When the environment catalog is enabled, CMF adds one catalog to every environment automatically. You do not create or delete it.

  • The catalog name is _env_ followed by the environment name. For example, environment prod has the catalog _env_prod. The _env_ prefix keeps the name from colliding with your Kafka catalogs.

  • Each environment catalog includes a default database that CMF creates automatically and that you cannot delete. You can create additional databases with CREATE DATABASE and drop them with DROP DATABASE.

  • CMF stores the catalog’s objects (functions and tables) in its own database. No Schema Registry or Kafka cluster is required.

The objects you can create in the environment catalog are:

Enable the environment catalog

The environment catalog is an opt-in feature and is disabled by default. Enable it in your Helm values file:

cmf:
  sql:
    environmentCatalog:
      enabled: true

While the feature is disabled, the environment catalog is not registered, and the catalog endpoint returns HTTP 404.

To register and use functions or custom connectors, also enable artifact management, which stores the JAR files that your DDL references. See User-Defined Functions in Confluent Manager for Apache Flink and Custom Connectors and Formats in Confluent Manager for Apache Flink.

Use the environment catalog

To create or use objects in the environment catalog, set the statement’s current catalog and database to the environment catalog and its default database.

With the Confluent CLI, pass the --catalog and --database flags:

confluent flink statement create show-functions \
  --url http://cmf:8080 \
  --environment prod \
  --compute-pool pool \
  --catalog _env_prod \
  --database default \
  --sql "SHOW USER FUNCTIONS;"

With the REST API, set the current catalog and database in the statement’s properties:

{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Statement",
  "metadata": {
    "name": "show-functions"
  },
  "spec": {
    "statement": "SHOW USER FUNCTIONS;",
    "computePoolName": "pool",
    "properties": {
      "sql.current-catalog": "_env_prod",
      "sql.current-database": "default"
    }
  }
}

Within the environment catalog you can run function DDL (CREATE FUNCTION, DROP FUNCTION), table DDL (CREATE TABLE, ALTER TABLE, DROP TABLE), database DDL (CREATE DATABASE, DROP DATABASE), and queries (SELECT, INSERT INTO) that reference those objects.

View the environment catalog

CMF exposes a read-only view of an environment catalog and its databases through the REST API:

curl -v -X GET http://cmf:8080/cmf/api/v1/environments/prod/catalog

The response contains the catalog name and its databases. It does not list the functions in the catalog; to list functions, run a SHOW USER FUNCTIONS; statement against the catalog. The endpoint returns HTTP 404 when the environment catalog feature is disabled.

Limitations

  • CREATE TABLE AS (CTAS) and CREATE OR REPLACE TABLE AS (RTAS) statements are not supported in the environment catalog.

  • UPDATE and DELETE statements are not supported, in line with Flink SQL, which does not support these DML statements in streaming mode. Use INSERT INTO to write rows and SELECT to read them.

  • CREATE TEMPORARY FUNCTION and CREATE TEMPORARY SYSTEM FUNCTION are not supported. Register functions with CREATE FUNCTION instead.

  • The environment catalog and its default database cannot be renamed or deleted.

  • User-defined functions must be implemented in Java. See User-Defined Functions in Confluent Manager for Apache Flink.

Related content