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.

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:

Install the Confluent Platform for Apache Flink Kubernetes operator.
Install Confluent Manager for Apache Flink.
Install or upgrade CFK with Flink integration enabled:
```
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.
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:
```
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.
Create a CMF REST class.
Create a Flink environment.
Create a Flink application or run Flink SQL statements.
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.

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.

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.
CFK can authenticate to CMF without authentication or using mTLS.

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.

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 and Provide TLS keys and certificates in Java KeyStore format for the expected keys in the TLS secret.

Create the CMFRestClass CR with the following specification and deploy it with the kubectl apply -f command.

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:

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

Check the status:

kubectl get CMFRestClass default -n <namespace> -oyaml