Manage Global Flink Resources Using Confluent for Kubernetes

The secrets, catalogs, and databases that Apache Flink® SQL uses are shared across environments and statements. Create these resources, then reference them from the Flink SQL statements that you run in each environment.

Note

Flink SQL support in CFK is a preview feature in CFK 3.3.0. Do not use preview features in production.

These global resources form a dependency chain. Apply them in the following order, and delete them in the reverse order so that each CMF-side resource is removed before the resource it depends on:

FlinkSecret
  -> FlinkEnvironmentSecretMapping
    -> FlinkKafkaCatalog
      -> FlinkKafkaDatabase

Warning

A FlinkEnvironmentSecretMapping name must equal the connectionSecretId or connectionSecretRef name that a catalog or database references. CMF keys an environment’s secrets by the mapping name. This is CMF runtime behavior, not a CFK-side validation. If the mapping is missing, CMF silently ignores the catalog rather than returning an error.

Create a FlinkSecret

CMF needs credentials to reach Kafka and Schema Registry. A FlinkSecret syncs a same-namespace Kubernetes Secret to CMF under the FlinkSecret name. The catalog and database then reference the secret by that name.

The FlinkSecret indirection contract works as follows:

The Kubernetes Secret is the source of truth. The FlinkSecret does not store credential values. It references a secret in the same namespace through spec.secretRef.
CFK tracks the backing secret’s resourceVersion. When you rotate the secret, CFK re-syncs the new values to CMF automatically.
When you delete the FlinkSecret, CFK removes the synced secret from CMF. Deleting the FlinkSecret does not delete the backing Kubernetes Secret.

Because the FlinkSecret consumes a standard Kubernetes Secret, you can produce that secret with any external secret manager. For details, see Use an external secret manager.

Create the backing Kubernetes Secret and the FlinkSecret CR, and deploy them with the kubectl apply -f command.

apiVersion: v1
kind: Secret
metadata:
  name: flink-connection-credentials
  namespace: operator
type: Opaque
stringData:
  schema.registry.basic.auth.user.info: "sr-user:sr-password"
  sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username='kafka-user' password='kafka-password';"
---
apiVersion: platform.confluent.io/v1beta1
kind: FlinkSecret
metadata:
  name: flink-connection-secret
  namespace: operator
spec:
  secretRef: flink-connection-credentials   --- [1]
  cmfRestClassRef:                           --- [2]
    name: default
    namespace: operator

[1] The name of the Kubernetes Secret to sync. The Secret must be in the same namespace as the FlinkSecret.
[2] The reference to the CMFRestClass. If omitted, the CMFRestClass named default in the same namespace is used.

Check the status:

kubectl get flinksecret flink-connection-secret -n operator -oyaml

Expect cfkInternalState: CREATED and cmfSync.status: Created.

Create a FlinkEnvironmentSecretMapping

A FlinkEnvironmentSecretMapping makes a synced secret usable by the catalogs and databases in an environment. Without a mapping, CMF ignores any catalog whose connectionSecretId is unmapped.

Create the FlinkEnvironmentSecretMapping CR and deploy it with the kubectl apply -f command.

apiVersion: platform.confluent.io/v1beta1
kind: FlinkEnvironmentSecretMapping
metadata:
  name: flink-connection-secret   --- [1]
  namespace: operator
spec:
  flinkEnvironment: flink-env1     --- [2]
  secretRef:
    name: flink-connection-secret  --- [3]
  cmfRestClassRef:
    name: default
    namespace: operator

[1] The mapping name. This name must equal the connectionSecretId or connectionSecretRef name that the catalog and database reference. CMF keys an environment’s secrets by the mapping name.
[2] The Flink environment to expose the secret to.
[3] The FlinkSecret to expose.

Check the status:

kubectl get flinkenvironmentsecretmapping flink-connection-secret -n operator -oyaml

Expect cfkInternalState: CREATED and cmfSync.status: Created.

Create a FlinkKafkaCatalog

A FlinkKafkaCatalog provides the Schema Registry instance that backs its databases. The connectionSecretId references the secret mapping name so that CMF can authenticate to Schema Registry. You add Kafka backed databases separately through FlinkKafkaDatabase, not in the catalog spec.

Create the FlinkKafkaCatalog CR and deploy it with the kubectl apply -f command.

apiVersion: platform.confluent.io/v1beta1
kind: FlinkKafkaCatalog
metadata:
  name: kafka-catalog
  namespace: operator
spec:
  flinkEnvironment: flink-env1
  srInstance:                               --- [1]
    connectionConfig:
      schema.registry.url: http://schemaregistry.operator.svc.cluster.local:8081
    connectionSecretId: flink-connection-secret   --- [2]
  cmfRestClassRef:
    name: default
    namespace: operator

[1] The Schema Registry instance that backs the catalog’s databases.
[2] The secret mapping name from Create a FlinkEnvironmentSecretMapping, used to authenticate to Schema Registry.

Check the status:

kubectl get flinkkafkacatalog kafka-catalog -n operator -oyaml

Create a FlinkKafkaDatabase

A FlinkKafkaDatabase belongs to a catalog and binds to a specific Kafka cluster. This SQL namespace is where your statements read from and write to. Existing Kafka topics surface as tables, and CREATE TABLE adds new ones.

Create the FlinkKafkaDatabase CR and deploy it with the kubectl apply -f command.

apiVersion: platform.confluent.io/v1beta1
kind: FlinkKafkaDatabase
metadata:
  name: clickstream
  namespace: operator
spec:
  flinkEnvironment: flink-env1
  catalogRef:                               --- [1]
    name: kafka-catalog
  kafkaCluster:                             --- [2]
    connectionConfig:
      bootstrap.servers: kafka.operator.svc.cluster.local:9071
  connectionSecretRef:                      --- [3]
    name: flink-connection-secret
  ddlEnvironments:                          --- [4]
    - flink-env1
  cmfRestClassRef:
    name: default
    namespace: operator

[1] The catalog that the database belongs to.
[2] The Kafka cluster that backs the database.
[3] The FlinkEnvironmentSecretMapping name that provides the Kafka credentials to CMF.
[4] The environments allowed to run Data Definition Language (DDL) statements against this database.

Check the status:

kubectl get flinkkafkadatabase clickstream -n operator -oyaml

Use an external secret manager

Because the FlinkSecret consumes a standard Kubernetes Secret, you can produce the backing secret with any tool that emits one. The FlinkSecret CR is unchanged. The example repository includes manifests for the following:

External Secrets Operator (ESO)
Sealed Secrets
HashiCorp Vault, through the Vault Secrets Operator (VSO)

To use one, install that operator and use its manifest to produce the flink-connection-credentials Secret. Then apply the FlinkSecret as described in Create a FlinkSecret.