# 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](access-control.md#cmf-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](access-control.md#cmf-resource-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: - [REST APIs for Confluent Manager for Apache Flink](../clients-api/rest.md#af-rest-api) - [Confluent CLI](https://docs.confluent.io/confluent-cli/current/command-reference/flink/index.html) - [Confluent for Kubernetes](../clients-api/flink-cfk.md#cmf-cfk) ## Authorization For role-based access control (RBAC), the Flink environment is used as the scope to [control access](/platform/current/security/authorization/rbac/authorization-acl-with-mds.html). You can grant individual users access to specific environments to read/manage Flink applications. ## 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](kubernetes-clusters.md#cmf-kubernetes-clusters). #### 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](access-control.md#cmf-access-control). #### 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](../installation/helm.md#cmf-namespace-scoped-rbac). #### 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](../installation/helm.md#cmf-manage-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. ### REST API Following is an example of a environment payload that can be used to create an environment using the Confluent REST API: ```json { "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: ```bash curl -v -H "Content-Type: application/json" -X POST http://cmf:8080/cmf/api/v1/environments -d @/path/to/env-dev.json ``` ### Confluent CLI #### 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` ```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` ```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` ```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. ```bash 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 statement in an environment, set `savepointScheduleDefaults` on the Environment payload. The field takes the same shape as `spec.savepointSchedule` on an individual resource. For example, the following default triggers a savepoint every six hours and keeps the five most recent for every resource in the environment: ```json { "name": "test-dev", "kubernetesNamespace": "dev-shared", "savepointScheduleDefaults": { "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](../jobs/savepoints.md#cmf-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. ### REST API To change the checkpointing interval to 90s, you would modify the `env-dev.json` file. The following is an example of the modified file: ```json { "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. Note the inclusion of the environment’s name in the URL. ```bash curl -v -H "Content-Type: application/json" -X POST http://cmf:8080/cmf/api/v1/environments/test-dev -d @/path/to/env-dev.json ``` ### Confluent CLI #### 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` ```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. ```bash 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 An environment can only be deleted if it contains no Flink applications and statements. You must first delete all Flink applications and statements contained in an environment before you can delete the environment. ### REST API The following curl command deletes the test-dev environment using the REST API: ```bash curl -v -X DELETE http://cmf:8080/cmf/api/v1/environments/test-dev ``` ### Confluent CLI The following command deletes the test-dev environment using the Confluent CLI: ```bash confluent flink environment delete test-dev ``` ## Related content - [Deploy and Manage Confluent Manager for Apache Flink Applications](../jobs/applications/overview.md#cmf-applications) - [Application Instances in Confluent Manager for Apache Flink](../jobs/applications/application-instances.md#cmf-application-instances) - [Confluent Manager for Apache Flink](../concepts/cmf.md#cmf)