<a id="co-manage-flink"></a>

# 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:

1. [Install Flink and Confluent for Kubernetes](co-flink-overview.md#co-flink-overview), and
   [create a CMF REST class](co-flink-overview.md#co-flink-rest-class).
2. [Create a Flink environment](co-manage-flink-environments.md#co-flink-environment).

To run Flink SQL statements instead of a packaged program, see
[Manage Flink SQL Statements Using Confluent for Kubernetes](co-manage-flink-sql.md#co-manage-flink-sql).

<a id="co-flink-application"></a>

## Create a Flink application

To create a Flink application resource in CFK:

1. Create a FlinkApplication CR using the following spec and deploy the resource
   using the `kubectl apply -f` command.
   ```yaml
   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](co-flink-overview.md#co-flink-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](co-manage-flink-environments.md#co-flink-environment).
   * [5] Flink configurations.
   * [6] The service account that runs Flink.
   * [7] FlinkJobManager
   * [8] FlinkTaskManager
   * [9] FlinkJob

   An example FlinkApplication CR:
   ```yaml
   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
   ```
2. Check the status.
   ```bash
   kubectl get flinkApplication -n <namespace> -oyaml
   ```

   For status details, see [Check the Flink application status](#co-flink-application-status).

<a id="co-flink-application-status"></a>

## Check the Flink application status

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

```yaml
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`:

1. 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.
2. Level 2, The CMF backend or the internal Kubernetes Operator might report an
   error in the `status.error` field.
3. 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:

```yaml
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:

```yaml
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
```