Confluent Developer Marketplace Get started
  1. Home
  2. Configure Kubernetes Clusters, Environments, Catalogs, Compute Pools, Artifacts, and Access Control for Confluent Platform for Apache Flink

Manage Kafka Catalogs and Databases for Confluent Manager for Apache Flink

Apache Flink® SQL uses the concept of catalogs and databases to connect to external storage systems. This page describes how to create and manage catalogs and databases using Confluent Manager for Apache Flink (CMF) for Flink SQL.

A core concept of SQL is the table. Tables store data, represented as rows. Users can query and modify the rows of a table by running SQL queries and Data Definition Language (DDL) statements. Most database systems store, manage, and process table data internally. In contrast, Flink SQL is solely a processing engine and not a data store. Flink accesses external data storage systems to read and write data.

Catalogs and databases bridge the gap between the SQL engine and external data storage systems, enabling users to access and manipulate data stored in various formats and locations.

CMF provides two kinds of catalogs:

Kafka catalogs

Built-in catalogs that connect to Kafka and Schema Registry. Each Kafka topic is exposed as a table, and its schema is derived from Schema Registry. You create and manage Kafka catalogs and their databases as described on this page.

Environment catalog

A writable catalog, one per environment, in which you register user-defined functions and create tables that use custom connectors with SQL DDL. CMF creates it automatically for each environment rather than through the REST calls on this page. For details, see Manage Environment Catalogs in Confluent Manager for Apache Flink.

The rest of this page describes how to create and manage Kafka catalogs and databases.

Catalogs and Databases: A catalog contains one or more databases. A catalog references a Schema Registry instance, which is used to derive table schemas from topic schemas. Each catalog can have multiple databases. A database references a Kafka cluster and contains tables that correspond to the topics in that cluster. Each topic of a Kafka cluster is represented as a TABLE in the database.

Tables: Tables are managed using SQL DDL statements such as CREATE TABLE, ALTER TABLE, and DROP TABLE. These statements allow you to create new tables, customize how Kafka topics are exposed as tables, and drop tables. For details, see Table Operations in Confluent Manager for Apache Flink.

Hierarchy:

  • CATALOG → references a Schema Registry instance

  • DATABASE → references a Kafka cluster (contained within a catalog)

  • TABLE → corresponds to a Kafka topic (contained within a database)

A catalog is a CMF global resource and potentially accessible from all CMF environments. The access of an environment to a catalog or database can be restricted with Secrets and Secret Mappings (see section below).

Create a catalog

A Kafka catalog references a Schema Registry instance, which is used to look up topic schema information for the topics of the Kafka clusters referenced by the catalog’s databases. The catalog is configured with connection properties for a Schema Registry client. These properties are used to fetch the metadata that is needed during query translation and execution.

A Kafka Catalog is configured with the following resource definition:

{
   "apiVersion": "cmf.confluent.io/v1",
   "kind": "KafkaCatalog",
   "metadata": {
     "name": "kafka-cat"
   },
   "spec": {
     "srInstance": {
       "connectionConfig": {
         "schema.registry.url": "http://schemaregistry:8081"
       },
       "connectionSecretId": "sr-conn-secret-id"
     }
   }
 }

  • The connectionConfig property is a map and can contain further configuration options as key/value pairs, within the limits described in Connection configuration limits.

  • The connectionSecretId property is optional. It can be configured to provide a placeholder against which environments can link secrets with connection credentials. For instruction on how to create a secret and link it to a catalog, see Secrets and secret mappings.

Given a catalog resource definition that is stored in a file catalog.json, the following REST call creates the catalog:

curl -v -H "Content-Type: application/json" \
-X POST http://cmf:8080/cmf/api/v1/catalogs/kafka \
-d@/<path-to>/catalog.json

Create a database

A database is created within a catalog and references a Kafka cluster. The database exposes the topics of its Kafka cluster as tables. The schema of a table is derived from the schema information of its topic that is fetched from the Schema Registry instance referenced by the catalog that the database is located in.

A database is configured with connection properties for Kafka clients. These properties are used to access metadata from Kafka and to read data from and write data to topics during query execution.

A database is configured with the following resource definition:

{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "KafkaDatabase",
  "metadata": {
    "name": "kafka-db"
  },
  "spec": {
    "kafkaCluster": {
      "connectionConfig": {
        "bootstrap.servers": "kafka-1:9092"
      },
      "connectionSecretId": "kafka-1-secret-id"
    },
    "ddlEnvironments": [
      "admin-env"
    ]
  }
}

  • The connectionConfig property is a map containing the bootstrap servers for the Kafka cluster. It is possible to add additional Kafka connection options as key/value pairs, within the limits described in Connection configuration limits.

  • The optional connectionSecretId property can be used to map environment-specific secrets containing sensitive authentication credentials. See the following section on Secrets and Secret Mappings for instructions.

  • The ddlEnvironments property lists the environments that can run table operation statements (CREATE TABLE, ALTER TABLE, DROP TABLE) against tables in this database. You cannot submit table operation statements from environments not listed in this property. If you try, the statement fails with an error. For details, see Table Operations in Confluent Manager for Apache Flink.

Important

The Kafka and Schema Registry credentials of an environment that has access to a database must meet the following minimum permissions:

  • Catalog Schema Registry credentials must allow you to read subjects and schemas.

  • Database Kafka credentials must allow you to list topics and read from the metadata topic (_confluent_sr_catalog). In addition, the credentials must allow you to read from and write to all topics that statements in the environment need to access.

For environments listed in the ddlEnvironments property, the credentials must additionally include:

  • Catalog Schema Registry credentials must allow you to create and delete subjects and schemas.

  • Database Kafka credentials must allow you to create and delete topics, and write to the metadata topic (_confluent_sr_catalog).

Given a database resource definition that is stored in a file database.json, the following REST call creates the database in the catalog kafka-cat:

curl -v -H "Content-Type: application/json" \
-X POST http://cmf:8080/cmf/api/v1/catalogs/kafka/kafka-cat/databases \
-d@/<path-to>/database.json

Connection configuration limits

Starting with CMF 2.4.3, CMF rejects a request to create or update a catalog or a database if its connectionConfig does any of the following:

  • Sets config.providers or a config.providers.* key.

  • Uses a ${...} placeholder in a value.

  • Sets sasl.kerberos.kinit.cmd.

  • Names a class that is not one of the standard Kafka and Schema Registry security classes.

The standard classes are:

  • In sasl.jaas.config: PlainLoginModule, ScramLoginModule, OAuthBearerLoginModule, and Krb5LoginModule.

  • In a sasl.*.class key: the Kafka OAuth callback handlers OAuthBearerLoginCallbackHandler and OAuthBearerValidatorCallbackHandler.

  • In ssl.engine.factory.class: DefaultSslEngineFactory.

Any other key that names a class, such as a *.class or *.classes key, security.providers, metric.reporters, or a serializer or deserializer key, is rejected. The same rules apply to keys with the schema.registry. prefix.

To allow more classes, list them in the CMF property cmf.kafka.connection-config.additional-allowed-classes. For example, AWS MSK IAM authentication needs software.amazon.msk.auth.iam.IAMLoginModule and software.amazon.msk.auth.iam.IAMClientCallbackHandler in this list. To keep credentials out of the resource definition, put them in a secret and reference it with connectionSecretId, as described in Secrets and secret mappings. If you need Kafka config providers instead, list their classes in cmf.kafka.connection-config.allowed-config-provider-classes.

CMF does not check the data of a secret when you create it. When CMF builds the catalog for a statement, it merges the mapped secret into the connection configuration, removes any key that breaks these rules, and logs a warning. A credential that depends on a removed key, such as a custom login module, then stops working.

CMF also logs a warning for a key outside the standard connection keys: bootstrap.servers, schema.registry.url, security.protocol, basic.auth.credentials.source, basic.auth.user.info, and keys that start with sasl., ssl., or bearer.auth.. Such keys still work in CMF 2.4.3. To stop the warning, list the keys in cmf.kafka.connection-config.additional-allowed-keys.

Secrets and secret mappings

Catalogs and databases are configured with a connectionConfig to access a Schema Registry instance or a Kafka cluster, respectively. Production setups of these systems require user authentication. Schema Registry and Kafka access credentials are sensitive information that must be securely stored and not be shared.

CMF features the concepts of secrets to securely store access credentials and secret mappings to link secrets per environment to catalogs and databases. This mechanism allows CMF administrators to configure for each environment different access credentials (with different permissions) for catalogs and databases.

  • A secret stores a map of connection properties in the CMF database. The data is encrypted only if encryption is enabled. For details, see Data Encryption in Confluent Manager for Apache Flink.

  • Catalogs and databases can define a connectionSecretId in their spec. The connectionSecretId is a placeholder against which an environment can map a secret.

  • A secret mapping is created in the context of an environment and links a secret to a connectionSecretId.

The Schema Registry connection properties of a catalog for an environment are derived by combining the properties of the public connectionConfig of the catalog resource with the properties of the secret that is linked for the environment to the catalog’s connectionSecretId. If a catalog has no connectionSecretId defined, every environment only uses the properties of the public connectionConfig. This implies, that all environments use the same connection properties for a catalog without connectionSecretId. If a catalog has a connectionSecretId defined, but an environment does not link a secret to this connectionSecretId, then the catalog and all of its databases are not accessible from the environment. The same mechanism is used to derive the Kafka connection properties of a database for an environment.

Important

Within an environment, CMF uses the same Schema Registry and Kafka connection properties to translate and execute all statements, regardless of the user who submits the statement. Hence, data access of a user is controlled by their access to CMF environments and the connection configuration of these environments. RBAC cannot isolate these credentials between statements that share an environment. For more information, see Environment isolation.

Secrets and secret mappings enable very flexible configuration of environment connection credentials. An environment can map a secret to multiple different connectionSecretIds, reusing the secret for multiple catalogs or databases. Multiple catalogs or databases can share the same connectionSecretId. A secret that is mapped to a shared connectionSecretId will be used for all catalogs or databases with that connectionSecretId.

The following examples show how to configure secrets for a catalog. The configuration for databases is analogous.

Create a secret resource with connection options for Schema Registry authentication:

{
    "apiVersion": "cmf.confluent.io/v1",
    "kind": "Secret",
    "metadata": {
      "name": "sr-secret"
    },
    "spec": {
      "data": {
          "basic.auth.user.info": "admin:admin",
          "basic.auth.credentials.source": "USER_INFO"
      }
    }
}

  • The data property is a map of key/value pairs. Starting with CMF 2.4.3, the keys are subject to the limits described in Connection configuration limits. To create a secret for Kafka connection credentials, put the respective configuration options into the data map.

The secret is created via the REST API:

curl -v -H "Content-Type: application/json" \
-X POST http://cmf:8080/cmf/api/v1/secrets -d@/<path-to>/sr-secret.json

The following resource definition maps the secret "sr-secret" to the connectionSecretId "sr-conn-secret-id":

{
    "apiVersion": "cmf.confluent.io/v1",
    "kind": "EnvironmentSecretMapping",
    "metadata": {
      "name": "sr-conn-secret-id"
    },
    "spec": {
      "secretName": "sr-secret"
    }
}

The mapping is created for an environment env-1 with the following REST request:

curl -v -H "Content-Type: application/json" \
-X POST http://cmf:8080/cmf/api/v1/environments/env-1/secret-mappings \
-d@/<path-to>/sr-secret-mapping.json

With this mapping, statements created in environment env-1 will use the following properties to configure the Schema Registry client of catalog kafka-cat:

// from the plain "connectionConfig" of catalog `kafka-cat`
"schema.registry.url": "http://schemaregistry:8081",
// from the secret "sr-secret" that is mapped to `kafka-cat`'s connectionSecretId 'sr-conn-secret-id'
"basic.auth.user.info": "admin:admin",
"basic.auth.credentials.source": "USER_INFO"

Follow the same steps to create a secret for Kafka credentials and map them to the connectionSecretId of a database.

Update a catalog

You can update a catalog’s configuration by submitting a PUT request with the updated specification. For example, you can update the catalog’s name or Schema Registry connection configuration. You cannot change the connectionSecretId of the Schema Registry instance.

To update the kafka-cat catalog with a new resource file updated-catalog.json, run the following REST request:

curl -v -H "Content-Type: application/json" \
-X PUT "http://cmf:8080/cmf/api/v1/catalogs/kafka/kafka-cat" \
-d@/<path-to>/updated-catalog.json

Update a database

You can update a database’s configuration by submitting a PUT request with the updated specification. For example, you can update the database name, the Kafka cluster configuration or the list of environments with table operation permissions. You cannot change the connectionSecretId of the Kafka cluster. The update must include ddlEnvironments, which can be an empty list.

To update the kafka-db database in the kafka-cat catalog with a new resource file updated-database.json, run the following REST request:

curl -v -H "Content-Type: application/json" \
-X PUT "http://cmf:8080/cmf/api/v1/catalogs/kafka/kafka-cat/databases/kafka-db" \
-d@/<path-to>/updated-database.json

Delete a catalog

A catalog can be deleted via the Confluent CLI or the REST API.

Important

Only catalogs without databases can be deleted. Before deleting a catalog, you must delete all databases associated with it.

Delete a catalog kafka-cat with the Confluent CLI:

confluent flink catalog delete kafka-cat

Delete a catalog kafka-cat with the REST API:

curl -v -H "Content-Type: application/json" \
-X DELETE http://cmf:8080/cmf/api/v1/catalogs/kafka/kafka-cat

Delete a database

Important

Deleting a database removes only the database definition from CMF. Table metadata in the _confluent_sr_catalog topic and the schemas in Schema Registry are not deleted, and neither are the actual topics in the Kafka cluster.

Delete a database kafka-db in a catalog kafka-cat with the REST API:

curl -v -H "Content-Type: application/json" \
-X DELETE http://cmf:8080/cmf/api/v1/catalogs/kafka/kafka-cat/databases/kafka-db

Limitations

You cannot register your own catalog implementations. CMF provides the built-in KafkaCatalog, an optional example catalog enabled with the cmf.sql.examples-catalog.enabled configuration flag, and a per-environment environment catalog, enabled with the cmf.sql.environmentCatalog.enabled flag, for user-defined functions and tables that use custom connectors.