- CP Flink SQL
- Setup
- Flink SQL
- More Flink SQL Playing
- Flink Application
- Control Center UI Stops Displaying Issue
- Cleanup
The code and/or instructions here available are NOT intended for production usage. It's only meant to serve as an example or reference and does not replace the need to follow actual and official documentation of referenced products.
kind create cluster --image kindest/node:v1.31.0In order to run the k8s dashboard:
kubectl apply -f https://raw.githubusercontent.com/kubernetes/dashboard/v2.7.0/aio/deploy/recommended.yaml --context kind-kind
kubectl create serviceaccount -n kubernetes-dashboard admin-user
kubectl create clusterrolebinding -n kubernetes-dashboard admin-user --clusterrole cluster-admin --serviceaccount=kubernetes-dashboard:admin-user
token=$(kubectl -n kubernetes-dashboard create token admin-user)
echo $token
kubectl proxyCopy the token displayed on output and use it to login in K8s dashboard at http://localhost:8001/api/v1/namespaces/kubernetes-dashboard/services/https:kubernetes-dashboard:/proxy/#/login
You may need to wait a couple of seconds for dashboard to become available.
Let it run and open another terminal.
In case you are logged out cause of inactivity you may see errors as:
E0225 13:50:19.136484 67149 proxy_server.go:147] Error while proxying request: context canceled
Just stop the proxy before and execute again and login with the new token.
Run (check https://docs.confluent.io/operator/current/co-plan.html#co-long-image-tags - CFK version 3.2.1):
kubectl create namespace confluent
kubectl config set-context --current --namespace=confluent
helm repo add confluentinc https://packages.confluent.io/helm
helm repo update
helm upgrade --install operator confluentinc/confluent-for-kubernetes \
--version "0.1514.19"Check pod is ready:
watch kubectl get podsOnce the operator pod is ready we install our CP nodes:
kubectl apply -f kafka/kafka.yamlAnd wait for all pods (1 kraft, 1 broker, 1 SR, 1 C3) to be ready (it will take some time...):
watch kubectl -n confluent get podsOnce everything is completely ready you should have something like this:
NAME READY STATUS RESTARTS AGE
confluent-operator-66c65956d5-jdw9x 1/1 Running 0 6m36s
controlcenter-ng-0 3/3 Running 0 5m54s
kafka-0 1/1 Running 0 2m38s
kraftcontroller-0 1/1 Running 0 5m54s
schemaregistry-0 1/1 Running 5 (2m35s ago) 5m54s
Now we can forward the port of Control Center Next Generation:
kubectl -n confluent port-forward controlcenter-ng-0 9021:9021 > /dev/null 2>&1 &And then open http://localhost:9021 and check topics myevent and myaggregated have been already created with their corresponding schemas as per kafka/kafka.yaml file.
You should see an error stating The system cannot connect to Confluent Manager for Apache Flink.. It's expected cause we didnt install it yet.
Let's start S3proxy:
kubectl apply -f ./s3proxy/s3proxy-deployment.yaml
kubectl apply -f ./s3proxy/s3proxy-init-job.yamlAnd wait for pods to be ready:
watch kubectl -n confluent get podsYou should see something as:
NAME READY STATUS RESTARTS AGE
confluent-operator-5dcb4c6b99-5dv57 1/1 Running 0 6m56s
controlcenter-ng-0 3/3 Running 1 (3m2s ago) 6m35s
kafka-0 1/1 Running 1 (4m40s ago) 4m50s
kraftcontroller-0 1/1 Running 0 6m35s
s3proxy-695cc84bf-zgp77 1/1 Running 0 56s
s3proxy-init-pq9zm 0/1 Completed 0 56s
schemaregistry-0 1/1 Running 4 (5m5s ago) 6m35s
And then port-forward to access locally:
kubectl -n confluent port-forward service/s3proxy 8000:8000 > /dev/null 2>&1 &You can use a client as Cyberduck to connect to. Open the profile file ./cyberduck/S3_http.cyberduckprodfile. Configure:
Protocol S3Proxy (http, path-style)
Server localhost
Port 8000
Access Key ID admin
Secret Access Key password
It should list just a single folder named warehouse.
Install certificate manager:
kubectl create -f https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yamlWait until an endpoint IP is assigned when executing the following:
watch kubectl get endpoints -n cert-manager cert-manager-webhookInstall Flink Kubernetes Operator:
kubectl config set-context --current --namespace=confluent
helm repo add confluentinc https://packages.confluent.io/helm
helm repo update
helm upgrade --install cp-flink-kubernetes-operator confluentinc/flink-kubernetes-operator --version "1.130.3" --set watchNamespaces="{confluent}"Install Confluent Manager for Apache Flink:
helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink \
--version "~2.3.0" --set cmf.sql.production=false \
--namespace confluentCheck pods are ready (CFO and CMF):
watch kubectl -n confluent get podsOpen port forwarding for CMF:
kubectl port-forward service/cmf-service 8080:80 -n confluent > /dev/null 2>&1 &At this point no error The system cannot connect to Confluent Manager for Apache Flink. should show up in Control Center home page.
Let's first create our environment:
confluent flink environment create env1 --url http://localhost:8080 --kubernetes-namespace confluentIn case you get an error like this:
Error: you must log out of Confluent Cloud to use this command
Suggestions:
Log out with `confluent logout`.
Make sure to login into confluent cloud confluent login and logout after confluent logout (in case doing the logout only gives you an error), and execute again the creation of environment.
We can list it:
confluent flink environment list --url http://localhost:8080It should also be listed from the Control Center UI.
Now we can create our catalog (basically allowing Flink to automatically recognize our kafka cluster, topics and schema registry - no need to create tables and specify Flink connectors as usual with Apache Flink):
confluent flink catalog create flink/catalog.json --url http://localhost:8080You should get something like:
+---------------+--------------------------+
| Creation Time | 2025-11-29T18:59:34.197Z |
| Name | kafka-cat |
| Databases | |
+---------------+--------------------------+
You should also be able to see the catalog created in Control Center.
With CMF 2.1+ catalogs no longer embed Kafka clusters; you must create a separate KafkaDatabase under the catalog. The CLI does not yet support the creation of databases inside the catalog so we will use the CMF REST API:
curl -H "Content-Type: application/json" -X POST http://localhost:8080/cmf/api/v1/catalogs/kafka/kafka-cat/databases -d @flink/database.jsonYou can list the databases with:
curl http://localhost:8080/cmf/api/v1/catalogs/kafka/kafka-cat/databasesThe kafka cluster/database should also be displayed under the catalog in the ControlCenter.
Now we can create our compute pool (There is no cp-flink-sql 2.x image in CP 8.2, so SQL compute pools must stay on 1.19):
confluent flink compute-pool create flink/compute-pool.json --environment env1 --url http://localhost:8080You cal list:
confluent flink compute-pool list --environment env1 --url http://localhost:8080And also should be able to see on ControlCenter.
Now first lets excute our script that will populate the input topic myevent for that we will forward first the ports of the broker and schema registry:
kubectl port-forward -n confluent pod/kafka-0 9094:9094 > /dev/null 2>&1 &
kubectl port-forward svc/schemaregistry 8081:8081 -n confluent > /dev/null 2>&1 &And then run the script:
./kafka/create_events.shYou can see on Control Center the topic myeventstart to get populated.
Now in another shell we can open a flink shell using our compute pool and query our topic (this should lead to the instantiation of the corresponding shared Flink job manager):
confluent --url http://localhost:8080 --environment env1 --compute-pool pool flink shellWe can run the following query over our topic:
SELECT
window_start,
category,
SUM(`value`) AS total_value,
COUNT(`id`) AS event_count
FROM
TABLE(
TUMBLE(
TABLE `myevent` ,
DESCRIPTOR($rowtime),
INTERVAL '10' SECOND
)
)
GROUP BY
window_start,
window_end,
category;Meanwhile in another shell you could run:
watch kubectl -n confluent get podsAnd see the Flink cluster task manager being instantiated (as per our compute pool template definition) to execute our query. Once running the Flink sql shell should start receiving the results from our query. For now we are not sinking those results to anywhere so just receiving it on our shell.
You should also see your statement being executed in Control Center.
If you navigate to your warehouse bucket in S3proxy you should see something like:
Let's review the meaning of each folder/file:
checkpoints/This is the checkpoint storage directory. A Flink checkpoint is a consistent snapshot of a running job’s state that Flink periodically saves so it can resume processing exactly from where it left off after a failure.88c..That is the Flink JobID (one folder per running job). You will probably just have one for the Flink SQL you submitted on the shell.chk-15,chk-14, ... Each chk-N is one completed checkpoint (N is the checkpoint ID counter). We have configuredstate.checkpoints.num-retainedequal to 10. So only the last 10 checkpoints at each moment will be kept._metadataThis is the checkpoint “manifest” file. It contains the serialized metadata that tells Flink: which operators/tasks were snapshotted, what state handles exist, where the state files are and how to restore the job from that checkpoint. Since our job’s state is small, Flink will store it in a very compact way in _metadata.
cli-20260119-...It’s not the checkpoint state itself — it’s the stuff Flink needs so the cluster/job can recover after a JobManager failure. What's configured byhigh-availability.storageDirset tos3://warehouse/.blob\This is Flink’s BLOB store (binary large objects). It holds artifacts needed to run/recover the job, such as: uploaded JARs (if any), serialized execution plan pieces, sometimes SQL client artifacts, etc.job_88c…/Per-job subdirectory in the HA/BLOB storage (again that long hex is the JobID).blob_p-…The permanent blob entry.
completedCheckpoint…This is NOT the checkpoint data. It’s HA metadata that represents the CompletedCheckpointStore pointer/registry (basically “what checkpoints are considered completed and usable for recovery” and references to them).submittedJobGraph…This is the submitted JobGraph (the job definition that the dispatcher/jobmanager uses to recover the job after failover). If the JM dies and comes back, this is one of the things it consults to resubmit/recreate execution.
Let's stop our process by quitting the query in the shell and exiting. Once you hit Q quiting the execution of the query the cluster created should NOT be terminated because we are running a SHARED compute pool. Now you can quit the sql shell quit;.
You should be able to see the Flink pods getting terminated and removed with:
watch kubectl -n confluent get podsIf we check on Cyberduck again:
job-result-store/It's used by Flink’s Dispatcher / HA layer to persist the final outcome of a job so that: the cluster knows whether a job FINISHED, FAILED, or was CANCELED; the result survives JobManager restarts; clients (SQL Client, REST API) can still query job status after termination. This is not checkpoint state and not HA blobs — it’s purely job lifecycle metadata.cli-20260119-…This is the Dispatcher / client session ID (same prefix you saw elsewhere). Think: “this SQL Client / session submitted jobs under this namespace”7fb…This is the JobID of the job completed, failed or canceled.88c….jsonUsually contains: jobId, application / job name, final status = CANCELED/FINISHED/FAILED, timestamps, possibly partial execution metadata. It’s intentionally small. A job result marked asDIRTY.jsonmeans: “This job did not terminate in a globally clean, finalized way.” Typical causes: Job was CANCELLED, JobManager crashed during termination, HA recovery was interrupted.
If we download the result json file it should look something similar to:
{"result":{"id":"88c0f8d3dcc45a31a6878f2d4d2012f6","application-status":"CANCELED","accumulator-results":{"cp-foreground-sink":"rO0ABXNyACVvcmcuYXBhY2hlLmZsaW5rLnV0aWwuT3B0aW9uYWxGYWlsdXJlAAAAAAAAAAEDAAFMAAxmYWlsdXJlQ2F1c2V0ABVMamF2YS9sYW5nL1Rocm93YWJsZTt4cHBzcgATamF2YS51dGlsLkFycmF5TGlzdHiB0h2Zx2GdAwABSQAEc2l6ZXhwAAAAAXcEAAAAAXVyAAJbQqzzF/gGCFTgAgAAeHAAAAA9AAAAOQAAAAAAAALoJTBiZDM2Y2RiLWIwNTItNDNkZi1iMjgyLWE4Mzg1MWU0NzdjYgAAAAAAAALoAAAAAHh4"},"net-runtime":7192080},"version":1}We could now direct from Control Center execute our statement (selecting the compute pool, catalog and database):
SELECT
window_start,
category,
SUM(`value`) AS total_value,
COUNT(`id`) AS event_count
FROM
TABLE(
TUMBLE(
TABLE `myevent` ,
DESCRIPTOR($rowtime),
INTERVAL '10' SECOND
)
)
GROUP BY
window_start,
window_end,
category;It will reuse the same shared compute pool.
You can now view the results from inside Control Center.
Stop the statement from Control Center.
We can deploy our full statement that will populate the other topic myaggreagted with the query results:
confluent --environment env1 flink statement create flink-statement \
--catalog kafka-cat \
--database main-kafka-cluster \
--compute-pool pool \
--parallelism 1 \
--sql $'INSERT INTO `myaggregated`
/*+ OPTIONS(\'properties.transaction.timeout.ms\'=\'300000\') */
SELECT
CAST(null as bytes) ,
window_start,
category,
SUM(`value`) AS total_value,
COUNT(`id`) AS event_count
FROM
TABLE(
TUMBLE(
TABLE \`myevent\` ,
DESCRIPTOR($rowtime),
INTERVAL \'10\' SECOND
)
)
GROUP BY
window_start,
window_end,
category;' \
--url http://localhost:8080You should get as response something like this:
+---------------+-------------------------------------------------------+
| Creation Date | 2025-07-28T23:48:14.523Z |
| Name | flink-statement |
| Statement | INSERT INTO `myaggregated` /*+ |
| | OPTIONS('properties.transaction.timeout.ms'='300000') |
| | */ SELECT CAST(null as bytes) , window_start, |
| | category, SUM(`value`) AS total_value, |
| | COUNT(`id`) AS event_count FROM TABLE( TUMBLE( |
| | TABLE `myevent` , DESCRIPTOR($rowtime), |
| | INTERVAL '10' SECOND ) ) GROUP BY |
| | window_start, window_end, category; |
| Compute Pool | pool |
| Status | PENDING |
| Status Detail | Statement execution pending. |
| Parallelism | 1 |
| Stopped | false |
| SQL Kind | INSERT_INTO |
| Append Only | true |
+---------------+-------------------------------------------------------+
Before the cluster starts running the statement will show as Pending in Control Center.
Once running if you go in Control Center to the message viewer of the topic myaggregated you should be able to see it getting populated.
You can check the status of the statement execution by running:
confluent flink statement list --environment env1 --url http://localhost:8080You can also open the Flink UI by executing:
kubectl port-forward service/flink-statement-rest 8082:8081 -n confluent > /dev/null 2>&1 &And then access http://localhost:8082
From Control Center from the statement view you can also open the Apache Flink Dashboard (this is embedding on Control Center and no port forward is needed for this).
We can call a manual backup (savepoint) of our statement by executing (replace the statement name by whatever applicable in your case):
confluent flink savepoint create save-explicit-1 \
--statement flink-statement \
--environment env1 \
--url http://localhost:8080 \
--path s3://warehouse/savepoints/manual/ \
--backoff-limit 0 \
-o yamlImportant: You need to pass the path s3://warehouse/savepoints/manual/ because even if it's defined in the compute pool configuration it's not being passed to the Flink Job Manager so without that it will use an empty string and result in an error.
After check its status:
confluent flink savepoint describe save-explicit-1 \
--statement flink-statement \
--environment env1 \
--url http://localhost:8080 \
-o yamlYou should get something like:
apiVersion: cmf.confluent.io/v1
kind: Savepoint
metadata:
name: save-explicit-1
creationTimestamp: "2026-01-19T14:30:34.873Z"
uid: 96f59db9-d96d-4503-94b2-4b1114c6d1c1
labels: {}
annotations: {}
spec:
path: s3://warehouse/savepoints/manual/
backoffLimit: 0
formatType: CANONICAL
status:
state: COMPLETED
path: s3://warehouse/savepoints/manual/savepoint-c17acc-21c567bbc4c0
triggerTimestamp: "2026-01-19T14:30:35.168600917Z"
resultTimestamp: "2026-01-19T14:30:50.450157383Z"
failures: 0
And you should see in Cyberduck something like:
Let's stop first our process by executing:
confluent flink statement stop \
flink-statement \
--environment env1 \
--url http://localhost:8080If you execute:
confluent flink statement list --environment env1 --url http://localhost:8080You should see the reference to your stopped statement.
Also you should be able to see that no incoming messages are coming to the topic myaggregated.
We could resume it by executing:
confluent flink statement resume flink-statement --environment env1 --url http://localhost:8080You should see in Control Center the same statement now is Pending.
And we should be able to see the flink pods being recreated.
Checking its status again:
confluent flink statement list --environment env1 --url http://localhost:8080And new messages arriving in the topic myaggregated.
It was capable to resume using the last checkpoint.
First lets stop again our statement:
confluent flink statement stop \
flink-statement \
--environment env1 \
--url http://localhost:8080Now if we list our savepoints:
confluent flink savepoint list \
--statement flink-statement \
--environment env1 \
--url http://localhost:8080 \
-o yamlNow we can leverage the CMF REST API (since this is not yet available on Confluent CLI):
SQL="$(curl -sS \
"http://localhost:8080/cmf/api/v1/environments/env1/statements/flink-statement" \
| jq -r '.spec.statement')"
PROPS="$(curl -sS \
"http://localhost:8080/cmf/api/v1/environments/env1/statements/flink-statement" \
| jq -c '.spec.properties')"
curl -sS -X PUT \
-H "Content-Type: application/json" \
"http://localhost:8080/cmf/api/v1/environments/env1/statements/flink-statement" \
-d @- <<EOF
{
"apiVersion": "cmf.confluent.io/v1",
"kind": "Statement",
"metadata": {
"name": "flink-statement"
},
"spec": {
"statement": $(jq -Rn --arg s "$SQL" '$s'),
"properties": $PROPS,
"flinkConfiguration": {},
"computePoolName": "pool",
"parallelism": 1,
"stopped": false,
"startFromSavepoint": {
"savepointName": "save-explicit-1",
"allowNonRestoredState": false,
"savepointRedeployNonce": 1
}
}
}
EOFIf we check our pods we should see the restart happening. And we can check the status of the statement:
confluent flink statement list --environment env1 --url http://localhost:8080Stop all statements. And now only from Control Center let's execute:
CREATE TABLE myevent2 (
`id` STRING,
`value` INT NOT NULL,
`event_time` TIMESTAMP(3),
`category` STRING,
WATERMARK FOR `event_time` AS `event_time`
);For using event_time as our watermark.
We can then populate it:
INSERT INTO myevent2
/*+ OPTIONS('properties.transaction.timeout.ms'='300000') */
SELECT `id`,
`value`,
`event_time`,
`category`
FROM myevent;Leave it running we should see the new created topic myevent2 getting populated.
Now we create also a new myaggregated2 table/topic:
CREATE TABLE myaggregated2 (
window_start TIMESTAMP(3) NOT NULL,
category STRING,
total_value INT NOT NULL,
event_count BIGINT NOT NULL
);And populate it as before but using event_time and not $rowtime:
INSERT INTO `myaggregated2`
/*+ OPTIONS('properties.transaction.timeout.ms'='300000') */
SELECT
window_start,
category,
CAST(SUM(`value`) AS INT) AS total_value,
COUNT(`id`) AS event_count
FROM
TABLE(
TUMBLE(
TABLE `myevent2`,
DESCRIPTOR(`event_time`),
INTERVAL '10' SECOND
)
)
GROUP BY
window_start,
window_end,
category;Make sure no statements are running. If you want to minimize resource usage stop the shared compute pool also.
We will be leveraging the standard flink-sql-runner-example (https://github.com/apache/flink-kubernetes-operator/tree/main/examples/flink-sql-runner-example).
Compile:
cd flink/flink-sql-runner-example
mvn clean verifyBuild the docker image and load in kind (it may take a bit to load cause the flink image is not so small):
DOCKER_BUILDKIT=1 docker build . -t flink-sql-runner-example:latest
kind load docker-image flink-sql-runner-example:latestAnd now create our application:
cd ..
confluent flink application create application-sql-minimal.json --environment env1 --url http://localhost:8080Check pods are ready (1 job manager and 1 task manager):
watch kubectl get podsYou may loose access to Control Center UI. In such cases you will need to stop the process ocuppying the port 9021 (the forward of the C3 pod) and restart the forwarding. If you are on a mac do as follows:
sudo lsof -i :9021This will give you the PID of the process that you need to kill.
Then reexecute:
kubectl -n confluent port-forward controlcenter-ng-0 9021:9021 > /dev/null 2>&1 &Check myaggregatedis being populated.
rm -fr flink/flink-sql-runner-example/target
kind delete cluster