<a id="flink-sql-statements-manage"></a>

# Manage Statements in Confluent Manager for Apache Flink

There are different ways to interact with a CMF Statement resource. You can rescale, stop, resume,
and delete statements.

#### IMPORTANT
The examples in the following topics assume that CMF was installed with
the examples catalog enabled (`cmf.sql.examples-catalog.enabled=true`).

## Rescale statements

The rescaling operation changes the execution parallelism of a Statement.
This operation can only be applied to Statements in a non-terminal state such as
PENDING, RUNNING, STOPPED, and FAILING.

### REST API

The following example shows how to rescale a Statement via the REST API, with a PUT request to
update the resource with an adjusted `spec.parallelism` value.

```bash
curl -v -H "Content-Type: application/json" \
-X PUT http://localhost:8080/cmf/api/v1/environments/env-1/statements/stmt-1 \
-d @/path/to/stmt-1.json
```

### Confluent CLI

The following example shows how to rescale a Statement using the Confluent CLI.
For a full list of options, see the [https://docs.confluent.io/confluent-cli/current/confluent flink statement rescale <command-reference/flink/statement/confluent_flink_statement_rescale.html>](https://docs.confluent.io/confluent-cli/current/confluent flink statement rescale <command-reference/flink/statement/confluent_flink_statement_rescale.html>) reference.

```bash
confluent --environment env-1 flink statement rescale stmt-1 --parallelism 2
```

## Stop statements

A running statement can be stopped to pause it.
While it is stopped, it is not consuming any Kubernetes resources.
The stop operation can only be applied to Statements in a non-terminal state such as
`PENDING`, `RUNNING`, and `FAILING`. Whether a statement keeps its state
across a stop and resume depends on its upgrade mode, set in
`spec.upgradeMode`. With `savepoint`, Flink takes a savepoint when a
running statement is stopped so that the statement can later resume without
any data loss. With `stateless`, no state is kept. On a dedicated compute
pool, you can also set `last-state`, which resumes the statement from its
most recent checkpoint without taking a new savepoint. This mode needs
`state.checkpoints.dir` set on the compute pool, directly or through the
environment’s `computePoolDefaults`. If it is not set, the Kubernetes
Operator rejects the statement’s deployment. Shared compute pools don’t
support `last-state`. Starting with CMF 2.4.3, CMF rejects a
`last-state` statement on a shared compute pool with `HTTP 400`.

If the statement does not set `spec.upgradeMode`, CMF uses the
`job.upgradeMode` that a dedicated compute pool sets in its cluster spec.
If neither sets an upgrade mode, CMF uses `savepoint` only when the
compute pool, directly or through the environment’s `computePoolDefaults`,
sets both `state.checkpoints.dir` and `state.savepoints.dir`. Otherwise,
CMF uses `stateless`. Setting these directories in the statement’s
`flinkConfiguration` has no effect.

### REST API

To stop a Statement with the REST API, use a PUT request to update the
resource with a `spec.stopped` value of `true`.

```bash
curl -v -H "Content-Type: application/json" \
-X PUT http://localhost:8080/cmf/api/v1/environments/env-1/statements/stmt-1 \
-d @/path/to/stmt-1.json
```

### Confluent CLI

You can use the `stop` option with the|confluent-cli| to stop a statement:

```bash
confluent --environment env-1 flink statement stop stmt-1
```

## Resume statements

A stopped statement can be resumed. If the statement was running before and its
upgrade mode is `savepoint`, the resume operation uses the savepoint that was
taken during the stop operation to repopulate the Statement’s internal state
so that the statement continues processing without data loss. With
`last-state`, the statement resumes from its most recent checkpoint. With
`stateless`, the statement resumes with empty state.
The resume operation can only be applied to Statements in the `STOPPED` state.

### REST API

To resume a Statement with the REST API, use a PUT request to update the resource
with a `spec.stopped` value of `false`.

```bash
curl -v -H "Content-Type: application/json" \
-X PUT http://localhost:8080/cmf/api/v1/environments/env-1/statements/stmt-1 \
-d @/path/to/stmt-1.json
```

### Confluent CLI

You can use the `resume` option with the Confluent CLI to resume a statement:

```bash
confluent --environment env-1 flink statement resume stmt-1
```

## List exceptions

Like all Flink applications, SQL statements can fail. Failures are categorized into two types:

**Compilation failures**: These are immediately reported by CMF in the status field of the
response to a Statement creation request. The phase field will be set to `FAILED`, and the detail
field will contain the error message.

**Execution failures**:

* For statements that query metadata and are directly executed by CMF,
  failures are reported in the same manner as compilation failures.
* For `SELECT` and `INSERT INTO` statements that are executed on a Flink cluster, CMF stores the
  exceptions of the ten most recent execution failures. These can be retrieved from a dedicated
  REST endpoint or using a Confluent CLI command.

### REST API

The following GET REST request fetches the execution exceptions of Statement `stmt-1`.

```bash
curl -v -H "Content-Type: application/json" \
-X GET http://localhost:8080/cmf/api/v1/environments/env-1/statements/stmt-1/exceptions
```

### Confluent CLI

The following CLI command fetches the statement exceptions of Statement `stmt-1`.

```bash
confluent --environment env-1 flink statement exception list stmt-1
```

## Fetch results

CMF executes `SELECT` statements on a Flink cluster.
The executing Flink job of a `SELECT` statement buffers its results in memory.
The result rows can be retrieved from CMF via a REST endpoint.

The results of a `SELECT` statement are fetched with individual requests,
with each request returning a batch of result rows. Since statements can ingest
unbounded data (for example from a Kafka topic), the Flink job continuously computes possibly
unbounded results. If results of a `SELECT` statement are not fetched, the Flink job’s in-memory buffer
fills up and it will eventually pause processing.
The job will resume once results have been fetched and the result buffer has free space.

Result fetching REST requests use a `page-token` query parameter to iterate
over the batches of result rows. The first REST request must not include a
`page-token`, but all following requests must. The REST response provides a
`nextPageToken` in the `metadata.annotations` field that should be used for
the next request to fetch results. The `page-token` mechanism ensures that
result rows are sequentially returned.
After result rows are returned from the REST endpoint,
they might be discarded. This mean that you may not be able to fetch the results again.

#### NOTE
CMF’s result fetching mechanism is designed to support the development process and exploratory queries.
It should not be used in production use cases.

The following example shows the first REST result request without
`page-token` query parameter:

```bash
curl -v -H "Content-Type: application/json" \
-X GET http://localhost:8080/cmf/api/v1/environments/env-1/statements/stmt-1/results
```

Response to first result request with `nextPageToken` in the `metadata.annotations` field:

```json
{
    "apiVersion": "cmf.confluent.io/v1",
    "kind": "StatementResult",
    "metadata": {
        "creationTimestamp": "2025-07-29T12:36:39.862204842Z",
        "annotations": {
            "nextPageToken": "ODQ1N3w0YmYzM2FhZS1hMGRhLTQzOGQtODQ5NS03YTg0NTAyODE2YmJ8WHZlUExDd0RYajF0dFhWWlZxLURDdTRDRnFMWWtfZkV5NUlBb3h5blhVRQ"
        }
    },
    "results": {
        "data": [
            {
                "op": 0,
                "row": [
                    "47464f2b882507954a972df0162a148dea301138a575e90fd081f93bb37b8c1a"
                ]
            }
        ]
    }
}
```

The following example shows the second REST request with `page-token` query parameter:

```bash
curl -v -H "Content-Type: application/json" \
-X GET http://localhost:8084/cmf/api/v1/environments/test/statements/stmt/results\?page-token\=ODQ1N3w0YmYzM2FhZS1hMGRhLTQzOGQtODQ5NS03YTg0NTAyODE2YmJ8WHZlUExDd0RYajF0dFhWWlZxLURDdTRDRnFMWWtfZkV5NUlBb3h5blhVRQ
```

#### NOTE
The CLI SQL Shell automatically fetches results of SELECT queries and visualizes their results.

## Automatic resource cleanup for INSERT statements

When an `INSERT INTO` statement reaches the `COMPLETED` phase, CMF automatically deletes the
associated Flink deployment resources from Kubernetes. This behavior is enabled by default. For
configuration details, see [Configure SQL Statements for Confluent Manager for Apache Flink](../../installation/sql-statements.md#cmf-configure-sql).

## Delete statements

You can also delete a statement using the REST API or the Confluent CLI.

### REST API

The following example shows how you can delete a Statement using the REST API:

```bash
curl -v -H "Content-Type: application/json" \
-X DELETE http://localhost:8080/cmf/api/v1/environments/env-1/statements/stmt-1
```

### Confluent CLI

The following example shows how you can use the Confluent CLI to delete a statement:

```bash
confluent --environment env-1 flink statement delete stmt-1
```

The deletion of a statement deletes all Kubernetes resources.

## Access the Flink web UI

The statement must be in the `RUNNING` phase to access the web UI.

### Confluent CLI

Use the Confluent CLI to forward the web UI to a local port:

```bash
confluent --environment env-1 flink statement web-ui-forward stmt-1
```

This starts a local proxy and prints the URL. Use `--port` to specify a port; otherwise
a random port is assigned.

### REST API

Access the web UI directly via the CMF REST API:

```http
GET http://cmf:8080/cmf/api/v1/environments/{env}/statements/{statement-name}/flink-web-ui/{path}
```

For shared pools, users with ClusterAdmin, SystemAdmin, or super.user permissions can also
access a full-cluster view of the shared pool that shows all statements running on it.
See [Manage Compute Pools in Confluent Manager for Apache Flink](../../configure/compute-pools.md#cmf-compute-pools) for details.

<a id="cmf-manage-statements-disconnected-cluster"></a>

## Behavior on disconnected or decommissioned clusters

When the Kubernetes cluster that an environment is bound to is `DISCONNECTED` or
`DECOMMISSIONED`, the behavior of statement operations changes:

- Describe and list operations return the last known data cached in CMF. The
  response includes a `status.warning` field indicating that the data might be
  stale and does not reflect the actual state on Kubernetes.
- List exceptions returns the exceptions stored in CMF. Fetching results is
  not served from the cache, because CMF reads result rows from the running
  Flink job.
- Write operations (create, rescale, stop, resume, delete) are rejected with `HTTP 503`.

To remove a statement from CMF’s database when the cluster is unreachable, use `?force=true`
on the delete request. A forced delete skips the cluster connectivity check and
still tries to delete the statement’s Kubernetes resources, but it removes the
statement from CMF even if that deletion fails. In that case, the
`FlinkDeployment` or `FlinkSessionJob` custom resource can remain on the
cluster as an orphan. A forced delete is still rejected while savepoints are
attached to the statement, so delete those savepoints first.

```bash
curl -X DELETE \
  "http://cmf:8080/cmf/api/v1/environments/env-1/statements/stmt-1?force=true"
```

For more information on cluster states and recovery, see [Manage Multiple Kubernetes Clusters in Confluent Manager for Apache Flink](../../configure/kubernetes-clusters.md#cmf-kubernetes-clusters).

## Related content

* [Create Statements with Confluent Manager for Apache Flink](create-statements.md#flink-sql-statements-create)
* [Deploy and Manage Statements in Confluent Manager for Apache Flink](overview.md#flink-sql-overview)
* [Savepoint Management in Confluent Manager for Apache Flink](../savepoints.md#cmf-savepoints)
* [Manage Multiple Kubernetes Clusters in Confluent Manager for Apache Flink](../../configure/kubernetes-clusters.md#cmf-kubernetes-clusters)