<a id="co-flink-global-resources"></a>

# 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:

```text
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.

<a id="co-flink-sql-secret"></a>

## 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](#co-flink-sql-external-secrets).

1. Create the backing Kubernetes Secret and the FlinkSecret CR, and deploy them
   with the `kubectl apply -f` command.
   ```yaml
   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.
2. Check the status:
   ```bash
   kubectl get flinksecret flink-connection-secret -n operator -oyaml
   ```

   Expect `cfkInternalState: CREATED` and `cmfSync.status: Created`.

<a id="co-flink-sql-secret-mapping"></a>

## 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.

1. Create the FlinkEnvironmentSecretMapping CR and deploy it with the
   `kubectl apply -f` command.
   ```yaml
   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.
2. Check the status:
   ```bash
   kubectl get flinkenvironmentsecretmapping flink-connection-secret -n operator -oyaml
   ```

   Expect `cfkInternalState: CREATED` and `cmfSync.status: Created`.

<a id="co-flink-sql-catalog"></a>

## 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.

1. Create the FlinkKafkaCatalog CR and deploy it with the `kubectl apply -f`
   command.
   ```yaml
   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](#co-flink-sql-secret-mapping), used to authenticate to Schema Registry.
2. Check the status:
   ```bash
   kubectl get flinkkafkacatalog kafka-catalog -n operator -oyaml
   ```

<a id="co-flink-sql-database"></a>

## 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.

1. Create the FlinkKafkaDatabase CR and deploy it with the `kubectl apply -f`
   command.
   ```yaml
   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.
2. Check the status:
   ```bash
   kubectl get flinkkafkadatabase clickstream -n operator -oyaml
   ```

<a id="co-flink-sql-external-secrets"></a>

## 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](#co-flink-sql-secret).