Manage Flink Environments Using Confluent for Kubernetes

A Apache Flink® environment is a set of configurations that Flink applications and SQL statements share. After you create an environment, you can deploy Flink applications or run Flink SQL statements against it.

Create a Flink environment

A Flink environment is a set of configurations that Flink applications and SQL statements use.

The FlinkEnvironment specification supports the computePoolDefaults, kubernetesClusterName, secrets, and statementDefaults fields, and reports status information.

Create a FlinkEnvironment CR using the following specification, and deploy it with the kubectl apply -f command.
```
apiVersion: platform.confluent.io/v1beta1
kind: FlinkEnvironment
metadata:
  name:
  namespace:
spec:
  kubernetesNamespace:      --- [1]
  flinkApplicationDefaults: --- [2]
    metadata:               --- [3]
    spec:                   --- [4]
      flinkConfiguration:
  cmfRestClassRef:          --- [5]
    name:
    namespace:
```
- [1] The namespace of the Flink cluster.
  
  Typically, you would install the FlinkEnvironment CR in the CFK namespace (metadata.namespace), but the Flink cluster would be in another namespace (spec.kubernetesNamespace), for example, default.
- [2] Configurations for the Flink cluster to specify the deployment-wide default application settings.
- [3] Kubernetes API metadata.
- [4] Spec of the FlinkApplicationSpec type.
- [5] The reference to the REST Class you created in Create a CMF REST Class.
  
  You can install FlinkEnvironment CR and the CMF REST class in different namespaces.
  
  If omitted, the CMFRestClass of the name default in the same namespace is used.
Note

The FlinkEnvironment’s metadata.namespace is the home namespace for every Flink CR that references this environment. The FlinkApplication (and any other Flink CR that references this environment through spec.flinkEnvironment) must be in the same namespace as the FlinkEnvironment CR. CFK stamps an environment-rooted owner reference on those CRs for cascade deletion, and Kubernetes owner references cannot cross namespaces. Only the Flink cluster (spec.kubernetesNamespace) and the CMF REST class (cmfRestClassRef) can live in other namespaces.

An example FlinkEnvironment CR:
```
apiVersion: platform.confluent.io/v1beta1
kind: FlinkEnvironment
metadata:
  name: flink-env1
  namespace: operator
spec:
  kubernetesNamespace: default
  flinkApplicationDefaults:
    metadata:
      labels:
        "acmecorp.com/owned-by": "analytics-team"
    spec:
      flinkConfiguration:
        taskmanager.numberOfTaskSlots: "2"
        rest.profiling.enabled: "true"
  cmfRestClassRef:
    name: default
    namespace: operator
```

Check the status.

kubectl get flinkEnvironment -n <namespace> -oyaml

Migrate cross-namespace Flink resources

Starting in CFK 3.3.0, a FlinkApplication and the other Flink CRs must be in the same namespace as the FlinkEnvironment that they reference through spec.flinkEnvironment. CFK stamps an environment-rooted owner reference on these CRs so that deleting the FlinkEnvironment cascades to them, and Kubernetes owner references cannot cross namespaces. Only the Flink cluster (spec.kubernetesNamespace) and the CMF REST class (cmfRestClassRef) can point to a different namespace.

If you deployed the FlinkEnvironment and FlinkApplication in different namespaces on an earlier release, each FlinkApplication reports status.cmfSync: Failed after you upgrade to CFK 3.3.0. The FlinkApplication’s underlying Flink job keeps running, so you can complete the following migration without downtime.

Part A: Restore the applications. For each namespace that hosts a FlinkApplication, or any other Flink CR:

Create a FlinkEnvironment CR in that namespace with the same name and an identical spec as the original: the same spec.kubernetesNamespace, the same flinkApplicationDefaults, and a cmfRestClassRef that is reachable from this namespace. Because the name matches, this is an idempotent update of the same CMF environment. The FlinkApplication’s running Flink job is not disturbed.
Verify that the FlinkEnvironment is created and the applications recover:
```
kubectl get flinkenvironment <name> -n <app-namespace>
kubectl get flinkapplication <app> -n <app-namespace> -o yaml
```
The FlinkEnvironment shows a Created state. Each FlinkApplication shows status.cmfSync no longer Failed and gains a metadata.ownerReferences entry that points at the new FlinkEnvironment.

Part B: Remove the redundant original FlinkEnvironment. Do this only if the original FlinkEnvironment’s namespace no longer hosts any Flink CRs.

Warning

Do not run a plain kubectl delete flinkenvironment on a healthy FlinkEnvironment. Its finalizer deletes the shared CMF environment and cascade-deletes every FlinkApplication in that namespace, destroying their running Flink jobs. Use the following procedure instead, which removes the CR but preserves the CMF environment and its jobs.

Block reconciliation so the operator does not act on the deletion:

kubectl annotate flinkenvironment <name> -n <old-namespace> \
  platform.confluent.io/block-reconcile=true --overwrite

Delete the CR. With --wait=false, the command returns immediately while the CR enters Terminating state. Because reconciliation is blocked, the operator does not delete the CMF environment and does not cascade-delete the applications:
```
kubectl delete flinkenvironment <name> -n <old-namespace> --wait=false
```

Remove the finalizer so the CR is deleted. The CMF environment and all running Flink jobs are preserved:

kubectl patch flinkenvironment <name> -n <old-namespace> \
  --type=merge -p '{"metadata":{"finalizers":[]}}'