<a id="cmf-sql-flink-versions"></a>

# Flink Versions for Statements in Confluent Manager for Apache Flink

Starting with version 2.5.0, Confluent Manager for Apache Flink® (CMF) runs Flink SQL statements on
Flink 1.19 or Flink 2.3. Flink 2.3 adds SQL features that Flink 1.19 does not
have, such as process table functions and batch statements. Each statement runs
on the Flink version of its compute pool, so statements on both versions can run
side by side, and you can move workloads to Flink 2.3 one at a time.

## Choose a Flink version

The compute pool sets the Flink version: set `flinkVersion` in its
`clusterSpec` to `v2_3`, and set `image` to a `cp-flink-sql` image for
Flink 2.3. A statement has no version setting of its own. It runs on the version
of the compute pool that you submit it to.

The following dedicated compute pool runs statements on Flink 2.3:

```json
{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "ComputePool",
  "metadata": {
    "name": "pool-2-3"
  },
  "spec": {
    "type": "DEDICATED",
    "clusterSpec": {
      "flinkVersion": "v2_3",
      "image": "confluentinc/cp-flink-sql:2.3-cmf2.5-cp1",
      "taskManager": {
        "resource": {
          "cpu": 1.0,
          "memory": "1024m"
        }
      },
      "jobManager": {
        "resource": {
          "cpu": 0.5,
          "memory": "1024m"
        }
      }
    }
  }
}
```

You create it the same way as any other compute pool. For details, see
[Manage Compute Pools in Confluent Manager for Apache Flink](../../configure/compute-pools.md#cmf-compute-pools).

- Use an image for the same Flink version as `flinkVersion`. CMF does not
  check that the two match.
- The `computePoolDefaults` of an environment take precedence over the
  `clusterSpec` of its compute pools. If they set `flinkVersion` or
  `image`, every compute pool in the environment uses those values, so create
  Flink 2.3 compute pools in an environment whose defaults do not set these
  fields. For details, see [Configure Environments in Confluent Manager for Apache Flink](../../configure/environments.md#cmf-environments).

<a id="cmf-sql-image-tags"></a>

## Image tags

Starting with CMF 2.5.0, the tag of a `cp-flink-sql` image names the CMF
version that the image is for. Tags have the form
`<flink-version>-cmf<cmf-version>-cp<build>`:

- `<flink-version>` is the Flink version of the image, `1.19` or `2.3`.
- `<cmf-version>` is the CMF minor version, such as `2.5`.
- `<build>` counts the builds of the image for that CMF minor version,
  starting at 1.

For example, `1.19-cmf2.5-cp1` and `2.3-cmf2.5-cp1` are the first Flink 1.19
and Flink 2.3 images for CMF 2.5. Every CMF 2.5.x release works with every
`cmf2.5` image.

Before CMF 2.5.0, tags had the form `1.19-cp<build>`, such as
`1.19-cp12`, and the build number counted across CMF versions.

## Compile statements for Flink 2.3

CMF compiles each statement for the Flink version of its compute pool. By
default, `cmf.statementService.sharedCompilers` in the Helm values has a
compiler for Flink 1.19 and a compiler for Flink 2.3, so Flink 2.3 compute pools
work without further setup. If you set `cmf.statementService.sharedCompilers`,
your list replaces the default list, so keep a compiler for Flink 2.3 in it.

## Features that need Flink 2.3

The following features work only on compute pools that use Flink 2.3:

- [Process table functions](process-table-functions.md#cmf-sql-ptf)
- Batch statements
- [Topic-subject associations](../../configure/catalog.md#cmf-catalog-subject-associations) for Kafka databases
- Built-in functions that Flink added after version 1.19, such as `BTRIM`

Other features, such as user-defined functions, custom connectors, and the
examples catalog, work on both versions, with the following differences:

- A user-defined function that you compile for Java 11 runs on both versions,
  unless it uses an API that Flink 2.0 removed. Compute pools that use Flink 2.3
  run Java 17, so a function for them can also target Java 17.
- Build a custom connector or format JAR for the Flink version of the compute
  pool that uses it.

## Move statements to Flink 2.3

A statement cannot move to another compute pool. While a compute pool has
statements that can still be resumed, CMF also rejects a change of its Flink
version, because their compiled plans and savepoints might not load on another
version.

To move a workload to Flink 2.3, create a compute pool for Flink 2.3 and create
new statements on it. The new statements start without the state of the old
ones.

## Related content

- [Manage Compute Pools in Confluent Manager for Apache Flink](../../configure/compute-pools.md#cmf-compute-pools)
- [Configure Environments in Confluent Manager for Apache Flink](../../configure/environments.md#cmf-environments)
- [User-Defined Functions in Confluent Manager for Apache Flink](user-defined-functions.md#cmf-sql-udf)
- [Features & Support for Statements in Confluent Manager for Apache Flink](features-support.md#flink-sql-features-support)