<a id="cmf-flink-cli"></a>

# CLI Operations for Confluent Manager for Apache Flink

Manage Flink environments and applications in Confluent Manager for Apache Flink® (CMF) by using Flink commands provided by
the Confluent CLI.

#### IMPORTANT
You must be using Confluent CLI version 4.7.0 or later to use these commands. Confluent Platform 7.8.0
provides this version of the Confluent CLI. For more information, see [Confluent CLI Compatibility](/platform/current/installation/versions-interoperability.html#confluent-cli).

You can use the Confluent CLI to manage the following CMF resources:

- [Environments](../configure/environments.md#cmf-environments)
- [Applications](../jobs/applications/overview.md#cmf-applications)
- [Compute pools](../configure/compute-pools.md#cmf-compute-pools)
- [Statements](../jobs/sql-statements/overview.md#flink-sql-statements), including the
  [interactive SQL shell](../jobs/sql-statements/use-interactive-shell.md#flink-sql-statements-interactive-cli)
- [Catalogs and databases](../configure/catalog.md#flink-sql-catalog)
- [Secrets and secret mappings](../configure/catalog.md#secret-mappings)
- [Savepoints and detached savepoints](../jobs/savepoints.md#cmf-savepoints)

This page describes the environment and application commands.

## Configure the CLI

To use the Confluent CLI to manage Flink Applications and Environments, the CLI needs to be connected
to the CMF server. There are two ways of configuring the CLI.

- Setting an environment variable
- Using a `--url` flag when you make a call

#### NOTE
You can only access the Flink on-premises commands when you are
logged out of Confluent Cloud.

To set the environment variable, specify the URL globally like the following:

```none
export CONFLUENT_CMF_URL=<URL-of-CMF-server>
```

Or each time you invoke the CLI, like the following:

```none
confluent flink environment list --url <cmf-url>
```

The following authentication and authorization parameters are available:

- Client Key path
- Client Certificate path
- Certificate Authority path

You can set them globally like the following:

```none
export CONFLUENT_CMF_CLIENT_KEY_PATH=<path/to/client/key/file>
export CONFLUENT_CMF_CLIENT_CERT_PATH=<path/to/client/cert/file>
export CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH=<path/to/certificate/authority/file>
```

Or pass them as flags each time you invoke the CLI.

```none
confluent flink environment list --url <cmf-url> --client-key-path <path/to/client/key/file>
confluent flink environment list --url <cmf-url> --client-cert-path <path/to/client/cert/file>
confluent flink environment list --url <cmf-url> --certificate-authority-path <path/to/certificate/authority/file>
```

## View available commands

You can see all of the commands available by using the `--help` option. To view the commands,
you must be logged out of Confluent Cloud.
For example:

```none
confluent flink environment --help
confluent flink application --help
```

You can also see a list of the available options on the
[https://docs.confluent.io/confluent-cli/current/](https://docs.confluent.io/confluent-cli/current/).
Make sure to choose the **On-Premises** tab.

## Manage Environments

Using the Confluent CLI, you can perform these actions:

- Create a new environment
- Update an existing environment
- Describe an environment
- Delete an environment
- List all available environments

### Create a new environment

In CMF, one Flink environment is attached to one Kubernetes namespace.
To learn more about environments, see [Environments](../configure/environments.md#cmf-environments).

There are two ways of specifying Flink application defaults in an environment.
You can pass a JSON string, or pass a path to a JSON or YAML file.

For example, to pass a JSON string, use the following command:

```none
confluent flink environment create <environment-name> --kubernetes-namespace <k8s-namespace> --defaults <json-defaults-string>
```

Example:

```none
confluent flink environment create dev-environment --kubernetes-namespace devops --defaults "{\"metadata\":{\"annotations\":{\"prometheus.io/scrape\":\"true\"}},\"spec\":{\"flinkConfiguration\":{\"taskmanager.numberOfTaskSlots\":\"4\"}}}"
```

To pass a path to a JSON or YAML file, use the following command:

```none
confluent flink environment create <environment-name> --kubernetes-namespace <k8s-namespace> --defaults <path/to/file>
```

Example:

```none
confluent flink environment create dev-environment --kubernetes-namespace devops --defaults ~/input/flink/environment/create-success-with-defaults.json
```

Example JSON input `(create-success-with-defaults.json` in the previous example):

```none
{
    "metadata": {
        "annotations": {
            "prometheus.io/scrape": "true"
        }
    },
    "spec": {
        "flinkConfiguration": {
            "taskmanager.numberOfTaskSlots": "4"
        }
    }
}
```

Note that the JSON specified is the JSON for the `flinkApplicationDefaults` key and not the
environment JSON.

Labels (`metadata.labels`) are not allowed in environment defaults. CMF
rejects defaults that contain labels.

Your output should resemble:

```none
+-----------------------+-----------------------------------------------------------------------------------------------------------------------------------+
|| Name                 || dev-environment                                                                                                                  |
|| Kubernetes Namespace || devops                                                                                                                           |
|| Created Time         || 2024-11-08 06:28:00.251622876                                                                                                    |
||                      || +0000 UTC                                                                                                                        |
|| Updated Time         || 2024-11-08 06:28:00.251623376                                                                                                    |
||                      || +0000 UTC                                                                                                                        |
|| Application Defaults || {"metadata":{"annotations":{"prometheus.io/scrape":"true"}},"spec":{"flinkConfiguration":{"taskmanager.numberOfTaskSlots":"4"}}} |
+-----------------------+-----------------------------------------------------------------------------------------------------------------------------------+
```

### Update an existing environment

You can update the defaults for Flink application in Flink environment
similar to how you create an environment.

#### IMPORTANT
You cannot update the Kubernetes namespace for an existing environment.
Also, if use the `update` command with no defaults, you will erase of the existing defaults.

To update an environment, you can either pass a JSON string or pass the path to a JSON or YAML file.

To pass a JSON string, use the following command.

```none
confluent flink environment update <environment-name> --defaults <json-defaults-string>
```

For example:

```none
confluent flink environment update dev-environment --defaults "{\"metadata\":{\"annotations\":{\"billing/origin\":\"sales-team-emea\"}}}"
```

Or to pass the path to a JSON or YAML file, use the following command.

```none
confluent flink environment update <environment-name> --defaults <path/to/json/file>
```

For example:

```none
confluent flink environment update dev-environment --defaults ~/input/flink/environment/create-success-with-defaults.json
```

A JSON file passed to this command might look like the following:

```none
{
    "metadata": {
        "annotations": {
            "billing/origin": "sales-team-emea"
        }
    }
}
```

Your output should resemble the following, regardless of whether you pass a string or a file path:

```none
+-----------------------+--------------------------------------------------------------------+
|| Name                 || dev-environment                                                   |
|| Kubernetes Namespace || devops                                                            |
|| Created Time         || 2024-11-05 11:01:52.497 +0000                                     |
||                      || UTC                                                               |
|| Updated Time         || 2024-11-05 11:32:30.458160877                                     |
||                      || +0000 UTC                                                         |
|| Application Defaults || {"metadata":{"annotations":{"billing/origin":"sales-team-emea"}}} |
+-----------------------+--------------------------------------------------------------------+
```

```none
confluent flink environment update dev-environment
```

Your output should resemble:

```none
+----------------------------+--------------------------------+
| Name                       | dev-environment                |
| Kubernetes Namespace       | devops                         |
| Created Time               | 2024-11-05 11:01:52.497 +0000  |
|                            | UTC                            |
| Updated Time               | 2024-11-05 11:34:36.823979088  |
|                            | +0000 UTC                      |
| Flink Application Defaults | null                           |
+----------------------------+--------------------------------+
```

### Describe an environment

Use the following command to get details of an existing Flink environment:

```none
confluent flink environment describe <environment-name>
```

Example command

```none
confluent flink environment describe dev-environment
```

Your output should resemble:

```none
+----------------------------+----------------------------------------------------------------------------------------------------------------------------------+
| Name                       | dev-environment                                                                                                                  |
| Kubernetes Namespace       | devops                                                                                                                           |
| Created Time               | 2024-11-08 06:28:00.251 +0000                                                                                                    |
|                            | UTC                                                                                                                              |
| Updated Time               | 2024-11-08 06:28:00.251 +0000                                                                                                    |
|                            | UTC                                                                                                                              |
| Flink Application Defaults | {"metadata":{"annotations":{"prometheus.io/scrape":"true"}},"spec":{"flinkConfiguration":{"taskmanager.numberOfTaskSlots":"4"}}} |
+----------------------------+----------------------------------------------------------------------------------------------------------------------------------+
```

### Delete an existing environment

Run the following command to delete an environment

```none
confluent flink environment delete <environment-name>
```

The optional `--force` flag skips the confirmation prompt.

```none
confluent flink environment delete <environment-name> --force
```

Your output should resemble:

```none
Deleted Flink environment "<environment-name>".
```

### List all available environments

You can use the following command to list environments:

```none
confluent flink environment list
```

Your output should resemble:

```none
Name           | Kubernetes Namespace |          Created Time          |          Updated Time
------------------+----------------------+--------------------------------+---------------------------------
dev-environment| devops               | 2024-11-05 11:01:52.497 +0000  | 2024-11-05 11:35:13.216 +0000
               |                      | UTC                            | UTC
staging        | billing              | 2024-11-05 11:38:46.969 +0000  | 2024-11-05 11:38:46.969 +0000
               |                      | UTC                            | UTC
```

The default output format for this command is human-readable.
You can get the output in JSON and YAML format too by passing the optional `--output` flag.

## Manage Applications

Using the Confluent CLI, you can perform these actions:

- Create a new application
- Update an existing application
- Describe an application
- Delete an application
- List all applications in a given environment
- Access the Flink Web UI for the application

The output format for these commands is either JSON or YAML. Human-readable output is not an option.

### Create a new application

Use the following command to create an application in an environment by passing the
path to the JSON or YAML file.

```none
confluent flink application create --environment <environment-name> <path/to/file>
```

Example command:

```none
confluent flink application create ~/resources/app_payload.json --environment dev-environment --output json
```

Example JSON:

```none
{
"apiVersion": "cmf.confluent.io/v1",
"kind": "FlinkApplication",
"metadata": {
    "name": "basic-example"
},
"spec": {
    "image": "confluentinc/cp-flink:1.19.1-cp1",
    "flinkVersion": "v1_19",
    "flinkConfiguration": {
    "taskmanager.numberOfTaskSlots": "4",
    "metrics.reporter.prom.factory.class": "org.apache.flink.metrics.prometheus.PrometheusReporterFactory",
    "metrics.reporter.prom.port": "9249-9250"
    },
    "serviceAccount": "flink",
    "jobManager": {
    "resource": {
        "memory": "1048m",
        "cpu": 1
    }
    },
    "taskManager": {
    "resource": {
        "memory": "1048m",
        "cpu": 1
    }
    },
    "job": {
    "jarURI": "local:///opt/flink/examples/streaming/StateMachineExample.jar",
    "state": "running",
    "parallelism": 2,
    "upgradeMode": "stateless"
    }
},
"status": null
}
```

Your output should resemble:

```none
{
"apiVersion": "cmf.confluent.io/v1",
"kind": "FlinkApplication",
"metadata": {
    "name": "basic-example"
},
"spec": {
    "image": "confluentinc/cp-flink:1.19.1-cp1",
    "flinkVersion": "v1_19",
    "flinkConfiguration": {
    "taskmanager.numberOfTaskSlots": "4",
    "metrics.reporter.prom.factory.class": "org.apache.flink.metrics.prometheus.PrometheusReporterFactory",
    "metrics.reporter.prom.port": "9249-9250"
    },
    "serviceAccount": "flink",
    "jobManager": {
    "resource": {
        "memory": "1048m",
        "cpu": 1
    }
    },
    "taskManager": {
    "resource": {
        "memory": "1048m",
        "cpu": 1
    }
    },
    "job": {
    "jarURI": "local:///opt/flink/examples/streaming/StateMachineExample.jar",
    "state": "running",
    "parallelism": 2,
    "upgradeMode": "stateless"
    }
},
"status": null
}
```

### Update an existing application

You can update an existing application in an environment by using this command:

```none
confluent flink application update --environment <environment-name> <path/to/file>
```

Example command

```none
confluent flink application update ~/resources/app_payload.json --environment dev-environment --output json
```

Your output should resemble your output from the create command, but with the specified properties updated.

### Describe an application

Use the following command to get details of an existing Flink application.

```none
confluent flink application describe <application-name> --environment <environment-name>
```

Example command

```none
confluent flink application describe basic-example --environment dev-environment
```

Your output should resemble:

```none
{
"apiVersion": "cmf.confluent.io/v1",
"kind": "FlinkApplication",
"metadata": {
    "name": "basic-example"
},
"spec": {
    "flinkConfiguration": {
    "metrics.reporter.prom.factory.class": "org.apache.flink.metrics.prometheus.PrometheusReporterFactory",
    "metrics.reporter.prom.port": "9249-9250",
    "taskmanager.numberOfTaskSlots": "4"
    },
    "flinkVersion": "v1_19",
    "image": "confluentinc/cp-flink:1.19.1-cp1",
    "job": {
    "jarURI": "local:///opt/flink/examples/streaming/StateMachineExample.jar",
    "parallelism": 2,
    "state": "running",
    "upgradeMode": "stateless"
    },
    ...
```

### Delete an existing application in an environment

Run the following command to delete an application.

```none
confluent flink application delete <application-name> --environment <environment-name>
```

The optional `--force` flag skips the confirmation prompt.

```none
confluent flink application delete <application-name> --environment <environment-name> --force
```

Your output should resemble:

```none
Deleted flink application "basic-example".
```

### List all application in a given environment

You can use the following command to list all applications
in a given environment:

```none
confluent flink application list --environment <environment-name>
```

Your output should resemble:

```none
    Name      |   Environment   |     Job Name      | Job Status
--------------+-----------------+-------------------+-------------
basic-example | dev-environment | State machine job | RUNNING
```

<a id="access-web-ui"></a>

### Access the Flink Web UI for the application

To access the Flink Web UI for a Flink application in an environment, use the following command:

```none
confluent flink application web-ui-forward <application-name> --environment <environment-name> --port <port-to-start-web-ui> --url <cmf-url>
```

Example command:

```none
confluent flink application web-ui-forward basic-example --environment dev-environment --port 9090 --url http://localhost:8080
```

Your output should resemble:

```none
Starting web UI at http://localhost:9090/ ... (Press Ctrl-C to stop)
```

Keep the session on and access the UI at the URL provided in the output.

#### IMPORTANT
There is a known bug in CLI v4.7.0 which requires you to pass the URL to the CMF server
explicitly for this command.

#### NOTE
CMF can’t proxy the Flink Web UI when TLS is enabled on the Flink REST endpoint (`security.ssl.rest.enabled: true`).
As a workaround, run `kubectl -n <namespace> port-forward svc/<cluster-id>-rest 8081:8081` and open `https://localhost:8081`.

## Debugging

If you experience issues with the Confluent CLI, you can use the `flag --unsafe-trace`  to get
more insights into the command that is running.
You will notice more output when you use this flag.

If you get a `not found` error response, and you are sure that the specified resource exists,
verify that the CMF server is up and running and you have connectivity to the server.

## Related content

- [Confluent CLI Flink Reference](https://docs.confluent.io/confluent-cli/current/command-reference/flink/index.html)
- [REST API](rest.md#af-rest-api)