Configure Environments in Confluent Manager for Apache Flink
An Flink environment in Confluent Manager for Apache Flink® (CMF) groups a set of Flink applications together and
maps them to a specific namespace in a Kubernetes cluster. The combination of
kubernetesClusterName and kubernetesNamespace defines where Flink resources for that
environment are deployed. This topic describes how to create, update, and delete Flink
environments in CMF.
Flink environments fulfill two main roles:
Isolation: Access control for individual team members is managed at the environment-level (logical isolation). Flink applications that belong to an environment are deployed to the same Kubernetes namespace (physical isolation). RBAC isolates which applications and statements each team can access within a shared environment, but statements that run in the same environment use the same mapped credentials for a given catalog or database. For what an environment can and cannot isolate, see Environment isolation.
Shared configuration: Flink configuration at the environment level has precedence over Flink configuration for individual Flink applications. This can be used to enforce configuration options and avoid redundant configuration. A common use case is setting a common observability configuration, or checkpoint storage destination for all Flink clusters in an environment.
For example, you might configure dev and prod environments or regional environments for your organization.
Flink configuration options, such as for the storage location of Flink checkpoints, are expected to differ between different environments, such as dev and prod. Instead of requiring individual Flink applications to configure these options and keep track of which environment they are running in, setting these options at the environment-level makes management easier and separates concerns between platform operators and developers nicely.
Tools for managing Flink environments
You can use these tools to create and manage environments:
Bind an environment to a Kubernetes cluster
Each environment deploys to a specific Kubernetes cluster. Set kubernetesClusterName
when creating an environment to choose the target cluster. If omitted, CMF deploys the
environment on the cluster where it is installed (the hub cluster). The cluster must already
be registered with CMF. For more information, see Manage Multiple Kubernetes Clusters in Confluent Manager for Apache Flink.
Note
You cannot change kubernetesClusterName or kubernetesNamespace after creation.
Creating an environment on a spoke cluster requires Use permission on the target
FlinkKubernetesCluster resource, in addition to the usual Edit permission on
FlinkEnvironment. No extra permission is required when creating an environment on the
default hub cluster. For role details, see Configure Access Control for Confluent Manager for Apache Flink.
Note
If CMF is installed with namespace-scoped RBAC, meaning the watchNamespaces Helm value lists
one or more namespaces, the environment’s kubernetesNamespace must be one of those
namespaces. Otherwise, creating the environment fails with a permissions error that names
the namespace. For more information, see Restrict CMF to specific namespaces.
Note
If CMF is installed with manageEnvironmentRbac set to false, pre-provision the
environment’s ServiceAccount, Roles, and RoleBindings in its
kubernetesNamespace before you create the environment. For more information, see
Pre-provision environment RBAC.
Create an environment
You can create an environment using the REST API or the Confluent CLI. First you create a JSON file that defines the environment and then you use the REST API or the Confluent CLI to create the environment.
Following is an example of a environment payload that can be used to create an environment using the Confluent REST API:
{
"name": "test-dev",
"kubernetesNamespace": "dev-shared",
"kubernetesClusterName": "my-cluster",
"flinkApplicationDefaults": {
"spec": {
"flinkConfiguration": {
"taskmanager.numberOfTaskSlots": "2"
},
"jobManager": {
"resource": {
"memory": "1024m",
"cpu": 0.5
}
},
"taskManager": {
"resource": {
"memory": "2048m",
"cpu": "1.5"
}
}
}
},
"computePoolDefaults": {
"clusterSpec": {
"image": "confluentinc/cp-flink-sql:1.19.1-cp8",
"flinkVersion": "v1_19",
"jobManager": {
"resource": {
"memory": "1024m",
"cpu": "1500m"
}
},
"taskManager": {
"resource": {
"memory": "2048m",
"cpu": "400m"
}
}
}
},
"statementDefaults": {
"detached": {
"flinkConfiguration": {
"state.backend.type": "rocksdb"
}
}
}
}
Save the JSON definition above to a file, for example, env-dev.json and use the following
curl command to create an environment using the REST API:
curl -v -H "Content-Type: application/json" -X POST http://cmf:8080/cmf/api/v1/environments -d @/path/to/env-dev.json
Note
The Confluent CLI creates environments on the default cluster only. To create an environment on a spoke cluster, use the REST API.
When you create an environment with the Confluent CLI, Flink application defaults, compute pool defaults, and statement defaults in different files, and pass those files to the Confluent CLI. Following are examples of the required JSON files:
env-dev-flink-application-defaults.json
{
"spec": {
"flinkConfiguration": {
"taskmanager.numberOfTaskSlots": "2"
},
"jobManager": {
"resource": {
"memory": "1024m",
"cpu": 0.5
}
},
"taskManager": {
"resource": {
"memory": "2048m",
"cpu": "1.5"
}
}
}
}
env-dev-compute-pool-defaults.json
{
"clusterSpec": {
"image": "confluentinc/cp-flink-sql:1.19.1-cp8",
"flinkVersion": "v1_19",
"jobManager": {
"resource": {
"memory": "1024m",
"cpu": "1500m"
}
},
"taskManager": {
"resource": {
"memory": "2048m",
"cpu": "400m"
}
}
}
}
env-dev-statement-defaults.json
{
"detached": {
"flinkConfiguration": {
"state.backend.type": "rocksdb"
}
}
}
Save the JSON definitions to files, for example, env-dev-flink-application-defaults.json, env-dev-compute-pool-defaults.json, and env-dev-statement-defaults.json respectively.
The following command creates an environment using the Confluent CLI.
confluent flink environment create test-dev --defaults /path/to/env-dev-flink-application-defaults.json --statement-defaults /path/to/env-dev-statement-defaults.json --compute-pool-defaults /path/to/env-dev-compute-pool-defaults.json --kubernetes-namespace dev-shared
Set a default savepoint schedule for the environment
To apply the same periodic savepoint schedule to every Flink application and
detached statement in an environment, set a default schedule on the
Environment payload. Set flinkApplicationDefaults.spec.savepointSchedule
for applications and statementDefaults.detached.savepointSchedule for
detached statements. To use one schedule for both, set it in both places.
Interactive statements do not inherit an environment-level schedule, and a
schedule under statementDefaults.interactive is rejected. Both fields take
the same shape as spec.savepointSchedule on an individual resource. For
example, the following defaults trigger a savepoint every six hours and keep
the five most recent for every application and detached statement in the
environment:
{
"name": "test-dev",
"kubernetesNamespace": "dev-shared",
"flinkApplicationDefaults": {
"spec": {
"savepointSchedule": {
"cronExpression": "0 0 */6 * * *",
"retention": { "keepLast": 5 }
}
}
},
"statementDefaults": {
"detached": {
"savepointSchedule": {
"cronExpression": "0 0 */6 * * *",
"retention": { "keepLast": 5 }
}
}
}
}
Resources can still set their own spec.savepointSchedule to supply values
for fields the environment default leaves unset. For the full list of schedule
fields, precedence rules between the environment default and the resource, and
how re-materialization works when you change the default, see
Schedule periodic savepoints.
Update an environment
You can update an existing environment to change its configuration, for example, to modify the default checkpointing interval, you create a file with the change and pass this file to the Confluent REST API or the Confluent CLI to update the environment.
To change the checkpointing interval to 90s, you would modify the env-dev.json file. The following is an example of the modified file:
{
"name": "test-dev",
"kubernetesNamespace": "dev-shared",
"kubernetesClusterName": "my-cluster",
"flinkApplicationDefaults": {
"spec": {
"flinkConfiguration": {
"taskmanager.numberOfTaskSlots": "2",
"execution.checkpointing.interval": "90s"
},
"jobManager": {
"resource": {
"memory": "1024m",
"cpu": 0.5
}
},
"taskManager": {
"resource": {
"memory": "2048m",
"cpu": "1.5"
}
}
}
},
"computePoolDefaults": {
"clusterSpec": {
"image": "confluentinc/cp-flink-sql:1.19.1-cp8",
"flinkVersion": "v1_19",
"jobManager": {
"resource": {
"memory": "1024m",
"cpu": "1500m"
}
},
"taskManager": {
"resource": {
"memory": "2048m",
"cpu": "400m"
}
}
}
},
"statementDefaults": {
"detached": {
"flinkConfiguration": {
"state.backend.type": "rocksdb"
}
}
}
}
The following curl command updates the test-dev environment using the REST API and the env-dev.json file.
Updates use the same endpoint as create. CMF identifies the environment
to update by the name field in the file.
curl -v -H "Content-Type: application/json" -X POST http://cmf:8080/cmf/api/v1/environments -d @/path/to/env-dev.json
Note
The Confluent CLI manages environments on the default cluster only. To manage an environment on a spoke cluster, use the REST API.
To change the checkpointing interval to 90s, you would modify the application defaults in the env-dev-flink-application-defaults.json file:
env-dev-flink-application-defaults.json
{
"spec": {
"flinkConfiguration": {
"taskmanager.numberOfTaskSlots": "2",
"execution.checkpointing.interval": "90s"
},
"jobManager": {
"resource": {
"memory": "1024m",
"cpu": 0.5
}
},
"taskManager": {
"resource": {
"memory": "2048m",
"cpu": "1.5"
}
}
}
}
The following command updates the test-dev environment using the Confluent CLI. Note the inclusion of other default files as well. If you do not include them, those defaults will be removed from the environment.
confluent flink environment update test-dev --defaults /path/to/env-dev-flink-application-defaults.json --statement-defaults /path/to/env-dev-statement-defaults.json --compute-pool-defaults /path/to/env-dev-compute-pool-defaults.json
Delete an environment
While an environment holds any of the following, every request to delete it
fails with a 422 error that names what is left, before anything is deleted.
You have to delete these yourself first:
Artifacts. For more information, see Delete an artifact.
Databases you created in the environment catalog, and functions, models, tables, and views in its
defaultdatabase. Drop views withDROP VIEWand tables withDROP TABLE. Neither statement drops the other kind of object. For more information, see Manage Environment Catalogs in Confluent Manager for Apache Flink.
For applications, statements, and compute pools, you have two choices. You
can delete them yourself and then delete the empty environment, or add
cascade=true to the request so that CMF deletes the environment together
with its applications, statements, compute pools, savepoints, and secret
mappings. The following table compares the requests:
Request |
Use when |
Result |
|---|---|---|
|
The environment has no applications, statements, or compute pools. |
|
|
The environment’s Kubernetes cluster is connected. |
|
|
The Kubernetes cluster is |
|
The CMF UI always deletes an environment together with everything in it, and offers the last two requests as Delete in the background and Force delete. If the Kubernetes cluster is disconnected or decommissioned, only Force delete is available.
Delete an empty environment
The following commands delete the test-dev environment after you have
deleted its applications, statements, and compute pools:
Note
Deleting an environment doesn’t delete its role bindings. If you create an environment with the same name, the old role bindings apply to it again. To remove them in the CMF UI, see Delete a role binding. To remove them with the Confluent CLI, see confluent iam rbac role-binding delete. For details, see RBAC scopes.
curl -v -X DELETE http://cmf:8080/cmf/api/v1/environments/test-dev
confluent flink environment delete test-dev
If the environment still has applications, statements, or compute pools, the
request fails with a 422 error. If the environment’s Kubernetes cluster is
DISCONNECTED or DECOMMISSIONED, the request fails with a 503
error. Add ?force=true to the REST request to delete the environment from
CMF anyway. CMF then leaves the environment’s Kubernetes objects, such as
its service account and secret, in the namespace, and you have to delete them
yourself.
Delete an environment and everything in it
Add cascade=true to delete the environment together with everything in it:
curl -X DELETE "http://cmf:8080/cmf/api/v1/environments/test-dev?cascade=true"
CMF accepts the request with a 202 response that lists the resources it
deletes:
{
"toBeDeleted": {
"applications": ["orders-app"],
"statements": ["orders-enrich"],
"computePools": ["pool-1"],
"savepoints": [
{
"name": "orders-app-savepoint-1",
"application": "orders-app",
"path": "s3://flink-state/savepoints/savepoint-3f2a1b"
}
],
"secretMappings": ["kafka-credentials"]
}
}
The deletion then runs in the background. CMF deletes each resource when
nothing that depends on it is left: the savepoints first, then the
applications and statements, then the compute pools, and finally the
environment with its Kubernetes service account and secret. Until the
environment is gone, you can still read it, and its
metadata.deletionRequestedTimestamp shows when the deletion was requested.
Any request to create or update the environment, or an application,
statement, compute pool, savepoint, artifact, or secret mapping in it, fails
with a 409 error.
The request needs the environment’s Kubernetes cluster and fails with a
503 error if the cluster is DISCONNECTED or DECOMMISSIONED. In that
case, use force delete.
What happens to a savepoint’s files depends on the state of its job when CMF deletes the savepoint:
If the job is running, CMF deletes the savepoint and its files.
If the job isn’t running, including a job that failed to deploy, or its application or statement is already gone, CMF removes the savepoint but keeps its files, and logs their location.
If the savepoint is still being taken, or its job is starting, restarting, or stopping, CMF waits and deletes the savepoint later.
Savepoints that running jobs take while the deletion runs are deleted too, although the response doesn’t list them. To keep the files of a running job’s savepoints, suspend its application or statement and wait until its job has stopped before you delete the environment.
If deleting a resource fails, CMF retries it, first after 30 seconds and then with the wait doubling each time up to 10 minutes. The deletion also continues after CMF restarts. To retry right away, send the same request again. The response then lists what is still left.
Finish a deletion that is stuck
A background deletion stays pending if Kubernetes never confirms that a
resource’s objects are deleted, for example because a finalizer on them never
completes. If the environment has been deleting for longer than expected, and
sending the request again keeps listing the same resources, send it with
skipCleanup=true:
curl -X DELETE "http://cmf:8080/cmf/api/v1/environments/test-dev?cascade=true&skipCleanup=true"
CMF then clears the Kubernetes finalizers that hold up the resources already being deleted. It removes savepoints that are already being deleted without waiting for their files to be deleted, and removes the savepoints it was waiting on without deleting their files. Skipping the cleanup means that a statement’s Flink job can keep running and that savepoint files stay in storage. CMF logs where the files are.
In the CMF UI, deleting the environment again offers this as
Skip remaining cleanup and finish deletion now. skipCleanup works only
together with cascade=true and without force=true. Otherwise, the
request fails with a 400 error.
Force delete an environment and everything in it
If the Kubernetes cluster is DISCONNECTED or DECOMMISSIONED, the
namespace was deleted outside CMF, or CMF can no longer access it, a
background deletion can’t finish. Add force=true as well:
curl -X DELETE "http://cmf:8080/cmf/api/v1/environments/test-dev?cascade=true&force=true"
CMF removes everything in the environment, and then the environment, before it responds. Each resource is removed from CMF whether or not its Kubernetes objects can be deleted, which CMF tries on a best-effort basis. CMF doesn’t delete savepoint files. A savepoint that was already being deleted before the request might have lost its files already.
The 200 response lists what was deleted under deleted, in the same
format as toBeDeleted. The savepoint paths tell you where the kept files
are. notCleanedUp lists the deleted resources whose Kubernetes objects
might still exist, because they couldn’t be deleted or were already being
deleted. In each entry, kind and name identify the deleted CMF
resource, namespace is where its Kubernetes objects are, and reason
names the objects, such as the FlinkDeployment of an application:
{
"deleted": {
"applications": ["orders-app"],
"statements": ["orders-enrich"],
"computePools": ["pool-1"],
"savepoints": [
{
"name": "orders-app-savepoint-1",
"application": "orders-app",
"path": "s3://flink-state/savepoints/savepoint-3f2a1b"
}
],
"secretMappings": ["kafka-credentials"]
},
"notCleanedUp": [
{
"kind": "FlinkApplication",
"name": "orders-app",
"namespace": "test-dev",
"reason": "Its FlinkDeployment was already being deleted, so it might still exist."
}
]
}
Delete the Kubernetes objects named in reason yourself if you still use
the cluster.
A force delete runs the same checks as a background deletion: you need the
same permissions, and user data still blocks it. If a resource can’t be
removed, the request fails with a 422 error and the environment stays
marked for deletion. Send the request again to remove the rest with force.
Until you do, CMF can go on deleting what is left without force, for
example after it restarts. That deletion needs the cluster, and it deletes
the files of savepoints whose job is running.
Permissions
Deleting an environment requires the Remove permission on the environment.
With cascade=true, you also need the Remove permission on every
application, statement, and compute pool in the environment, and on the
application or statement of each savepoint. If any of these permissions is
missing, the request fails and CMF deletes nothing. For the roles that grant
these permissions, see Role permissions.