Manage Flink Applications Using Confluent for Kubernetes

A Apache Flink® application is a user program that creates one or more Flink jobs to process data. Use a FlinkApplication to run a packaged Flink program, such as a JAR or container image that contains compiled DataStream or Table API code.

The FlinkApplication specification supports the startFromSavepoint and job.autoscalerResetNonce fields.

Before you create a Flink application, complete the following:

To run Flink SQL statements instead of a packaged program, see Manage Flink SQL Statements Using Confluent for Kubernetes.

Create a Flink application

To create a Flink application resource in CFK:

Create a FlinkApplication CR using the following spec and deploy the resource using the kubectl apply -f command.

apiVersion: platform.confluent.io/v1beta1
kind: FlinkApplication
metadata:
spec:
  cmfRestClassRef:
     name:                  --- [1]
     namespace:             --- [2]
  image:                    --- [3]
  flinkEnvironment:         --- [4]
  image:
  flinkVersion:
  flinkConfiguration:       --- [5]
  serviceAccount:           --- [6]
  jobManager:               --- [7]
  taskManager:              --- [8]
  job:                      --- [9]

[1] The reference to the REST Class you created in Create a CMF REST Class.
If omitted, the CMFRestClass of the name default in the same namespace is used.
[2] The namespace of this FlinkApplication CR.
The namespace of the Flink cluster is determined by FlinkEnvironment.spec.kubernetesNamespace.
[3] The CMF image.
[4] The reference to the FlinkEnvironment CR you created in Create a Flink environment.
[5] Flink configurations.
[6] The service account that runs Flink.
[7] FlinkJobManager
[8] FlinkTaskManager
[9] FlinkJob

An example FlinkApplication CR:

apiVersion: platform.confluent.io/v1beta1
kind: FlinkApplication
metadata:
  name: my-app1
  namespace: default
spec:
  flinkEnvironment: flink-env1
  image: confluentinc/cp-flink:1.19.1-cp1
  flinkVersion: v1_19
  flinkConfiguration:
    "taskmanager.numberOfTaskSlots": "2"
    "metrics.reporter.prom.factory.class": "org.apache.flink.metrics.prometheus.PrometheusReporterFactory"
    "metrics.reporter.prom.port": "9249-9250"
    "rest.profiling.enabled": "true"
  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: 3
    upgradeMode: stateless
  cmfRestClassRef:
    name: default
    namespace: operator

Check the status.
```
kubectl get flinkApplication -n <namespace> -oyaml
```
For status details, see Check the Flink application status.

Check the Flink application status

The following are the notable status fields in the CFK-manged FlinkApplication CR:

status:
  cmfSync:                     --- [1]
    errorMessage:              --- [2]
    lastSyncTime:              --- [3]
    status:                    --- [4]
  error:                       --- [5]
  clusterInfo:                 --- [6]
  jobManagerDeploymentStatus:  --- [7]
  jobStatus:
    state:                     --- [8]

[1] The status of the sync between CFK and CMF through the CMFRestClass CR.
[2] Any error message related to the sync between CFK and CMF ([1]), for example, a connection, authentication, or validation error.
[3] The time when the latest sync between CFK and CMF happened.
[4] The sync status. The possible values: CREATED, DELETED, UNKNOWN, FAILED.
[5] Indicates async errors during from the Flink deployment.
This is only set if status.cmfSync.errorMessage ([2]) is empty and status.cmfSync.status: CREATED.

For details about the below status fields, refer to the CMF documentation.

[6] Information about the Flink cluster when deployed. This section is only set if the status.error ([5]) is not set.
[7] Status of the JobManager deployment in Kubernetes.
[8] Status of the Flink job inside the FlinkApplication’s Flink cluster.

It is important to note, that there is a hierarchy of status/error fields in the FlinkApplication.status:

Level 1. The status.cmfSync field needs to be error-free, as this indicates that CFK was able to submit the FlinkApplication to the CMF backend.
Level 2, The CMF backend or the internal Kubernetes Operator might report an error in the status.error field.
Level 3. Once the errors with the above field are resolved, the rest of the status fields, get populated.

The following is an example of error status:

status:
  cfkInternalState: CREATED
  clusterInfo: {}
  cmfSync:
    errorMessage: ""
    lastSyncTime: "2024-11-05T19:15:09Z"
    status: Created
  error: '{"type":"org.apache.flink.kubernetes.operator.exception.ReconciliationException","message":"org.apache.flink.configuration.IllegalConfigurationException:
    JobManager memory configuration failed: Sum of configured JVM Metaspace (256.000mb
    (268435456 bytes)) and JVM Overhead (192.000mb (201326592 bytes)) exceed configured
    Total Process Memory (1 bytes).","additionalMetadata":{},"throwableList":[{"type":"org.apache.flink.configuration.IllegalConfigurationException","message":"JobManager
    memory configuration failed: Sum of configured JVM Metaspace (256.000mb (268435456
    bytes)) and JVM Overhead (192.000mb (201326592 bytes)) exceed configured Total
    Process Memory (1 bytes).","additionalMetadata":{}},{"type":"org.apache.flink.configuration.IllegalConfigurationException","message":"Sum
    of configured JVM Metaspace (256.000mb (268435456 bytes)) and JVM Overhead (192.000mb
    (201326592 bytes)) exceed configured Total Process Memory (1 bytes).","additionalMetadata":{}}]}'
  jobManagerDeploymentStatus: MISSING
  jobStatus:
    checkpointInfo:
      lastPeriodicCheckpointTimestamp: 0
    jobId: a6251e5a0f3f2e00f56874b56bc0780c
    jobName: ""
    savepointInfo:
      lastPeriodicSavepointTimestamp: 0
      savepointHistory: []
    state: ""
  lifecycleState: UPGRADING
  observedGeneration: 5
  reconciliationStatus:
    lastReconciledSpec: '{"spec":{"job":{"jarURI":"local:///opt/flink/examples/streaming/StateMachineExample.jar","parallelism":1,"entryClass":null,"args":[],"state":"suspended","savepointTriggerNonce":null,"initialSavepointPath":null,"checkpointTriggerNonce":null,"upgradeMode":"stateless","allowNonRestoredState":null,"savepointRedeployNonce":null},"restartNonce":null,"flinkConfiguration":{"rest.profiling.enabled":"true","taskmanager.numberOfTaskSlots":"2"},"image":"confluentinc/cp-flink:1.19.1-   cp1","imagePullPolicy":null,"serviceAccount":"flink","flinkVersion":"v1_19","ingress":null,"podTemplate":null,"jobManager":{"resource":{"cpu":1.0,"memory":"1","ephemeralStorage":null},"replicas":1,"podTemplate":{"metadata":{"labels":{"platform.confluent.io/origin":"flink"}}}},"taskManager":{"resource":{"cpu":1.0,"memory":"1","ephemeralStorage":null},"replicas":null,"podTemplate":{"metadata":{"labels":{"platform.confluent.io/   origin":"flink"}}}},"logConfiguration":null,"mode":null},"resource_metadata":{"apiVersion":"flink.apache.org/v1beta1","metadata":{"generation":6},"firstDeployment":true}}'
    reconciliationTimestamp: 1730834099726
    state: UPGRADING
  taskManager:
    labelSelector: ""
    replicas: 0

The following is an example status of a successful FlinkApplication creation:

status:
  cfkInternalState: CREATED
  clusterInfo:
    flink-revision: 89d0b8f @ 2024-06-22T13:19:31+02:00
    flink-version: 1.19.1-cp1
    total-cpu: "2.0"
    total-memory: "2516582400"
  cmfSync:
    errorMessage: ""
    lastSyncTime: "2024-11-05T19:19:10Z"
    status: Created
  jobManagerDeploymentStatus: READY
  jobStatus:
    checkpointInfo:
      lastPeriodicCheckpointTimestamp: 0
    jobId: 522d7ff7f15b4e138ffb9ea4053abbd3
    jobName: State machine job
    savepointInfo:
      lastPeriodicSavepointTimestamp: 0
      savepointHistory: []
    startTime: "1730834237948"
    state: RUNNING
    updateTime: "1730834248753"
  lifecycleState: STABLE
  observedGeneration: 6
  reconciliationStatus:
    lastReconciledSpec: '{"spec":{"job":{"jarURI":"local:///opt/flink/examples/streaming/StateMachineExample.jar","parallelism":1,"entryClass":null,"args":[],"state":"running","savepointTriggerNonce":null,"initialSavepointPath":null,"checkpointTriggerNonce":null,"upgradeMode":"stateless","allowNonRestoredState":null,"savepointRedeployNonce":null},"restartNonce":null,"flinkConfiguration":{"rest.profiling.enabled":"true","taskmanager.numberOfTaskSlots":"2"},"image":"confluentinc/cp-flink:1.19.1-   cp1","imagePullPolicy":null,"serviceAccount":"flink","flinkVersion":"v1_19","ingress":null,"podTemplate":null,"jobManager":{"resource":{"cpu":1.0,"memory":"1200m","ephemeralStorage":null},"replicas":1,"podTemplate":{"metadata":{"labels":{"platform.confluent.io/origin":"flink"}}}},"taskManager":{"resource":{"cpu":1.0,"memory":"1200m","ephemeralStorage":null},"replicas":null,"podTemplate":{"metadata":{"labels":{"platform.confluent.io/   origin":"flink"}}}},"logConfiguration":null,"mode":null},"resource_metadata":{"apiVersion":"flink.apache.org/v1beta1","metadata":{"generation":12},"firstDeployment":true}}'
    lastStableSpec: '{"spec":{"job":{"jarURI":"local:///opt/flink/examples/streaming/StateMachineExample.jar","parallelism":1,"entryClass":null,"args":[],"state":"running","savepointTriggerNonce":null,"initialSavepointPath":null,"checkpointTriggerNonce":null,"upgradeMode":"stateless","allowNonRestoredState":null,"savepointRedeployNonce":null},"restartNonce":null,"flinkConfiguration":{"rest.profiling.enabled":"true","taskmanager.numberOfTaskSlots":"2"},"image":"confluentinc/cp-flink:1.19.1-   cp1","imagePullPolicy":null,"serviceAccount":"flink","flinkVersion":"v1_19","ingress":null,"podTemplate":null,"jobManager":{"resource":{"cpu":1.0,"memory":"1200m","ephemeralStorage":null},"replicas":1,"podTemplate":{"metadata":{"labels":{"platform.confluent.io/origin":"flink"}}}},"taskManager":{"resource":{"cpu":1.0,"memory":"1200m","ephemeralStorage":null},"replicas":null,"podTemplate":{"metadata":{"labels":{"platform.confluent.io/   origin":"flink"}}}},"logConfiguration":null,"mode":null},"resource_metadata":{"apiVersion":"flink.apache.org/v1beta1","metadata":{"generation":12},"firstDeployment":true}}'
    reconciliationTimestamp: 1730834229475
    state: DEPLOYED
  taskManager:
    labelSelector: component=taskmanager,app=app111
    replicas: 1