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

# Flink on Confluent for Kubernetes Overview

Apache Flink® is a stateful, low-latency stream processing framework for running
streaming applications on large volumes of data.

Offered with Confluent Platform, Confluent Manager for Apache Flink® (CMF) is a self-managed service for Flink that
integrates with Apache Kafka®.

To learn more about CMF, see [Overview of Confluent Platform
for Apache Flink](https://docs.confluent.io/platform/current/flink/overview.html).

You can use Confluent for Kubernetes (CFK) to manage the full Flink lifecycle through
Kubernetes custom resources, including CMF, Flink environments, Flink
applications, and Flink SQL statements with their Day-2 resources such as compute
pools, catalogs, databases, and secrets.

The high-level workflow to manage Flink with CFK is:

1. [Install the Confluent Platform for Apache Flink Kubernetes operator](https://docs.confluent.io/platform/current/flink/installation/helm.html#step-2-install-the-af-cp-long-kubernetes-operator).
2. [Install Confluent Manager for Apache Flink](https://docs.confluent.io/platform/current/flink/installation/helm.html#step-3-deploy-cmf-from-the-confluent-helm-repository).
3. Install or upgrade CFK with Flink integration enabled:
   ```bash
   helm upgrade --install confluent-operator \
     confluentinc/confluent-for-kubernetes
   ```

   To configure CFK to listen to multiple namespaces or a single namespace,
   see [Configure CFK to manage Confluent Platform components in different namespaces](co-deploy-cfk.md#co-deployment-different-namespace).

   For example, to configure CFK to manage Flink in two namespaces,
   `confluent` and `default` and only in those namespaces, add the
   `--set namespaceList` and `--set namespaced=true` flags to the
   `helm upgrade` command as shown below:
   ```bash
   helm upgrade --install confluent-operator \
     confluentinc/confluent-for-kubernetes \
     --set namespaceList="{confluent,default}" \
     --set namespaced=true
   ```

   For more information about CFK installation, see [Deploy Confluent for Kubernetes](co-deploy-cfk.md#co-deploy-operator).
4. [Create a CMF REST class](#co-flink-rest-class).
5. [Create a Flink environment](co-manage-flink-environments.md#co-flink-environment).
6. [Create a Flink application](co-manage-flink.md#co-flink-application) or
   [run Flink SQL statements](co-manage-flink-sql.md#co-manage-flink-sql).
7. In the Flink Web UI, verify that the application job you created is running.

An example scenario of using CMF with CFK is available in the [CFK Example
Repository](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/flink/mTLS).

## Requirements and considerations

The following requirements apply to managing Flink applications with CFK. To run
Flink SQL statements, which is a preview feature in CFK 3.3.0, see the
requirements in [Requirements and considerations](co-manage-flink-sql.md#co-flink-sql-requirements).

* To manage Flink applications in CFK, you need the following versions:
  * CMF 2.0 or higher. CMF 1.x is deprecated.
  * CFK 2.10.0 and higher
  * Confluent Platform 8.0 or higher

  For the supported CMF, Confluent Platform, and Flink SQL runtime image combinations, see
  [Versions and Interoperability](https://docs.confluent.io/platform/current/flink/installation/versions-interoperability.html).
* CFK can authenticate to CMF without authentication or using
  mTLS.

<a id="co-flink-rest-class"></a>

## Create a CMF REST Class

When managing CMF in CFK, all Flink custom resources communicate with CMF
through the CMFRestClass custom resource.

You need to first set up a CMF REST Class custom resource (CR).

CMF REST Class is only used by CFK and is not part of CMF.

1. If using mTLS or TLS to connect to the Flink host, create a secret.

   Certificates with appropriate Subject Alternate Names (SANs) are required for
   the mTLS setup.
   * mTLS: You need to create a secret with certs and reference it in the
     CMFRestClass CR in the next step.
   * TLS: The secret is only required if using a self-signed certificate.

   See [Provide TLS keys and certificates in PEM format](co-network-encryption.md#co-certs-pem) and [Provide TLS keys and certificates in Java KeyStore format](co-network-encryption.md#co-certs-jks) for the expected keys in the
   TLS secret.
2. Create the CMFRestClass CR with the following specification and deploy it
   with the `kubectl apply -f` command.
   ```yaml
   apiVersion: platform.confluent.io/v1beta1
   kind: CMFRestClass
   metadata:
     name:                     --- [1]
     namespace:                --- [2]
   spec:
     cmfRest:                  --- [3]
       authentication:
         type:                 --- [4]
       endpoint:               --- [5]
       tls:                    --- [6]
         secretRef:            --- [7]
   ```

   * [1] The name of the REST Class.
   * [2] The namespace of the CMF REST Class.
   * [3] The CMF cluster.
   * [4] To use mTLS authentication, set to `mtls` and specify the certificates
     in [7].
   * [5] The endpoint of the CMF host.
   * [6] Required when the authentication type ([4]) is set to `mtls`.
   * [7] The name of the secret that contains the TLS certificates.

   An example CMFRestClass CR:
   ```yaml
   apiVersion: platform.confluent.io/v1beta1
   kind: CMFRestClass
   metadata:
     name: default
     namespace: operator
   spec:
     cmfRest:
       endpoint: https://cmf-service:80
       authentication:
         type: mtls
         sslClientAuthentication: true
       tls:
         secretRef: cmf-day2-tls
   ```

#### NOTE
By default, CFK allows `CMFRestClass` resources to be referenced from
different namespaces. For example, a FlinkEnvironment or FlinkApplication
can reference a `CMFRestClass` that is defined in another namespace.

If the CFK operator is configured to restrict cross-namespace REST class
usage, `CMFRestClass` references must be in the same namespace as the
referencing FlinkEnvironment or FlinkApplication resource. Cross-namespace
`CMFRestClass` references are then rejected by CFK validation. For more information on how to enable this
restriction on the CFK operator, see [Restrict cross-namespace REST class usage](co-deploy-cfk.md#co-deployment-restrict-cross-namespace).

1. Check the status:
   ```none
   kubectl get CMFRestClass default -n <namespace> -oyaml
   ```