Confluent Developer Marketplace Get started
  1. Home
  2. Flink Jobs
  3. Deploy and Manage Statements in Confluent Manager for Apache Flink

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:

{
  "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.

  • 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.

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:

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