DEV Community

Cover image for Deploying Apache Flink on a Kubernetes Cluster as an Alternative to GCP Dataflow
Vultr profile image Sanskriti Harmukh
Sanskriti Harmukh for Vultr

Posted on with Aashish Chaurasiya Originally published at docs.vultr.com

Deploying Apache Flink on a Kubernetes Cluster as an Alternative to GCP Dataflow

#flink #kubernetes #dataengineering #gcp

Google Cloud Dataflow is a managed stream and batch processing service built on Apache Beam that offers autoscaling, event-time processing, windowing, stateful computations, and exactly-once guarantees, but it ties deployments to Google Cloud services and ongoing managed service costs. Apache Flink is an open-source distributed processing engine for stateful computations over bounded and unbounded data streams that covers the same ground as a self-hosted deployment without ongoing managed service costs — it supports stateful processing, checkpointing, savepoints, event-time windowing, and Apache Beam pipelines through the Flink Runner, while the Flink Kubernetes Operator deploys and manages Flink clusters using native Kubernetes resources. This guide deploys Apache Flink as an alternative to Google Cloud Dataflow, covering Session and Application Cluster deployments using the Flink Kubernetes Operator, state management, checkpointing, exactly-once processing with Kafka, Apache Beam integration, monitoring, high availability, security, and migration strategies from Google Cloud Dataflow. By the end, you'll have a self-hosted Flink deployment on Kubernetes running Session and Application Clusters with durable state management, exactly-once Kafka processing, Beam pipeline support, Prometheus and Grafana monitoring, high availability, and network/RBAC security controls.

Understanding Flink Architecture

Apache Flink provides many of the stream and batch processing capabilities available in Google Cloud Dataflow through a self-managed Kubernetes deployment. The following table compares Dataflow components and features with their closest Apache Flink equivalents:

Google Cloud Dataflow Apache Flink Equivalent Description
Dataflow Job Flink Job or FlinkDeployment CR Defines and manages stream or batch processing applications
Dataflow Worker TaskManager Executes processing tasks and maintains operator state
Dataflow Control Plane JobManager Coordinates job execution, scheduling, checkpoints, and cluster operations
Dataflow Autoscaling Flink Autoscaler, Reactive Mode, or Horizontal Pod Autoscaler Adjusts processing capacity based on workload and resource demand
Dataflow Windowing Flink Windowing Supports tumbling, sliding, and session windows
Dataflow Exactly-Once Processing Checkpointing and Exactly-Once Sinks Provides consistent state recovery and end-to-end processing guarantees
Dataflow Stateful Processing Keyed State and State Backends Maintains state across events and processing operations
Dataflow Beam Runner Flink Runner Executes Apache Beam pipelines on Flink clusters
Dataflow Snapshots Savepoints Creates application state snapshots for recovery, upgrades, and migration
Cloud Monitoring Prometheus and Grafana Collects and visualizes cluster, job, and resource metrics
Dataflow Templates Application JARs and FlinkSessionJob CRs Packages and deploys reusable processing workloads

Key components of Apache Flink include:

  • JobManager: Coordinates job scheduling, checkpoints, recovery, and TaskManager operations.
  • TaskManager: Executes processing tasks and maintains application state assigned to its task slots.
  • Flink Kubernetes Operator: Automates the deployment, upgrade, scaling, recovery, and lifecycle management of Flink clusters and applications.
  • State Backend: Manages the working state used by stateful Flink operators.
  • Checkpoint Storage: Stores checkpoint and savepoint data in durable storage for recovery and application upgrades.
  • Checkpoints: Automatically created state snapshots used for fault recovery and exactly-once processing.
  • Savepoints: Manually triggered state snapshots used for controlled upgrades, migration, and operational recovery.

Prerequisites

Before you begin, you need to:

  • Have access to a Kubernetes cluster running Kubernetes 1.31 or later with at least 4 CPU cores and 16 GB of RAM per node. A minimum of three worker nodes is recommended.
  • Install kubectl and configure access to the cluster.
  • Install Helm 3 on your management workstation.
  • Install Java 11 or later and Apache Maven.
  • Configure a default Kubernetes StorageClass for persistent Kafka storage. Run kubectl get storageclass to confirm the name of the StorageClass available on your cluster and substitute it wherever this guide references storage.
  • Create a bucket on an S3-compatible object storage service. Record the bucket name, hostname, access key, and secret key.
  • Have a basic understanding of Kubernetes resources and stream-processing concepts.

Install the Flink Kubernetes Operator

The Apache Flink Kubernetes Operator extends Kubernetes with custom resources that simplify the deployment and lifecycle management of Apache Flink clusters and applications. It automates operations such as upgrades, scaling, savepoint management, and recovery while allowing you to manage Flink deployments using native Kubernetes resources.

Install cert-manager

The Flink Kubernetes Operator uses admission webhooks that require TLS certificates. Install cert-manager before deploying the operator.

1. Verify the Kubernetes cluster connection:

$ kubectl cluster-info
Enter fullscreen mode Exit fullscreen mode

2. Verify that all worker nodes are in the Ready state:

$ kubectl get nodes
Enter fullscreen mode Exit fullscreen mode

3. Add the Jetstack Helm repository:

$ helm repo add jetstack https://charts.jetstack.io
Enter fullscreen mode Exit fullscreen mode

4. Update the Helm repository cache:

$ helm repo update
Enter fullscreen mode Exit fullscreen mode

5. Install cert-manager, create the cert-manager namespace, and install the required Custom Resource Definitions:

$ helm install cert-manager jetstack/cert-manager \
    --namespace cert-manager \
    --create-namespace \
    --version v1.21.1 \
    --set crds.enabled=true
Enter fullscreen mode Exit fullscreen mode

6. Verify that all cert-manager pods are running:

$ kubectl get pods -n cert-manager
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                       READY   STATUS    RESTARTS   AGE
cert-manager-559776c68d-7vqzx              1/1     Running   0          9m5s
cert-manager-cainjector-7cdf9b4bd8-th5g2   1/1     Running   0          9m5s
cert-manager-webhook-5f7fd7899-pxt99       1/1     Running   0          9m5s
Enter fullscreen mode Exit fullscreen mode

Install the Apache Flink Kubernetes Operator

1. Add the Apache Flink Helm repository for version 1.15.0. This guide uses the Flink Kubernetes Operator version 1.15.0. Visit the Flink Kubernetes Operator downloads page to confirm the latest release, and replace the version in the repository URL if a newer version is available.

$ helm repo add flink-operator https://downloads.apache.org/flink/flink-kubernetes-operator-1.15.0/
Enter fullscreen mode Exit fullscreen mode

2. Update the Helm repository cache:

$ helm repo update
Enter fullscreen mode Exit fullscreen mode

3. Create a dedicated namespace for the Flink Kubernetes Operator:

$ kubectl create namespace flink-operator
Enter fullscreen mode Exit fullscreen mode

4. Install version 1.15.0 of the Flink Kubernetes Operator in the flink-operator namespace:

$ helm install flink-kubernetes-operator \
    flink-operator/flink-kubernetes-operator \
    --namespace flink-operator \
    --version 1.15.0 \
    --set image.tag=1.15.0
Enter fullscreen mode Exit fullscreen mode

5. Verify that the operator deployment is available:

$ kubectl get deployments -n flink-operator
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                        READY   UP-TO-DATE   AVAILABLE   AGE
flink-kubernetes-operator   1/1     1            1           38s
Enter fullscreen mode Exit fullscreen mode

6. Verify that the operator pod is running:

$ kubectl get pods -n flink-operator
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                        READY   STATUS    RESTARTS   AGE
flink-kubernetes-operator-85cbbc449-4cwjr   2/2     Running   0          59s
Enter fullscreen mode Exit fullscreen mode

Deploy a Flink Session Cluster

A Flink Session Cluster runs a long-lived JobManager and TaskManager deployment that can accept multiple Flink jobs. This mode is useful when you want to submit and manage several jobs without creating a separate cluster for each workload.

1. Create a working directory for the Flink manifests:

$ mkdir -p ~/flink-dataflow
Enter fullscreen mode Exit fullscreen mode

2. Switch to the working directory:

$ cd ~/flink-dataflow
Enter fullscreen mode Exit fullscreen mode

3. Create a namespace for Flink workloads:

$ kubectl create namespace flink
Enter fullscreen mode Exit fullscreen mode

4. Create a service account for Flink:

$ kubectl create serviceaccount flink -n flink
Enter fullscreen mode Exit fullscreen mode

5. Create a role binding that allows the Flink service account to manage resources in the flink namespace:

$ kubectl create rolebinding flink-role-binding \
    --clusterrole=edit \
    --serviceaccount=flink:flink \
    --namespace=flink
Enter fullscreen mode Exit fullscreen mode

6. Create a session-cluster.yaml manifest:

$ nano session-cluster.yaml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration to the file.

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: flink-session-cluster
  namespace: flink
spec:
  image: flink:1.20
  flinkVersion: v1_20
  serviceAccount: flink
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
    state.checkpoints.dir: file:///opt/flink/checkpoints
    state.savepoints.dir: file:///opt/flink/savepoints
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    replicas: 2
    resource:
      memory: "2048m"
      cpu: 1
Enter fullscreen mode Exit fullscreen mode

Note: The local checkpoint and savepoint directories validate the Session Cluster configuration but do not preserve state after pod replacement. The Configure State Management section replaces these paths with durable Object Storage.

The configuration:

  • Deploys a reusable Flink Session Cluster using the Flink Kubernetes Operator.
  • Allocates CPU and memory resources to the JobManager and TaskManagers.
  • Configures two task slots for each TaskManager.
  • Sets local checkpoint and savepoint directories for deployment testing.
  • Uses the flink service account to manage resources in the flink namespace.

Save and close the file.

7. Apply the Flink Session Cluster manifest:

$ kubectl apply -f session-cluster.yaml
Enter fullscreen mode Exit fullscreen mode

8. Verify that the Flink deployment is created:

$ kubectl get flinkdeployment -n flink
Enter fullscreen mode Exit fullscreen mode

Verify that the flink-session-cluster resource displays a STABLE lifecycle state.

9. Verify that the Flink Session Cluster pods are running:

$ kubectl get pods -n flink
Enter fullscreen mode Exit fullscreen mode

10. Verify the Flink services:

$ kubectl get svc -n flink
Enter fullscreen mode Exit fullscreen mode

The output displays the services created for the Flink Session Cluster, including the REST service used to access the Flink Web UI.

Submit a Flink Job

A FlinkSessionJob submits an application to an existing Flink Session Cluster without creating a separate Flink cluster for each workload. The application is packaged as a Java Archive (JAR) file and managed as a Kubernetes custom resource, making deployments and updates consistent with standard Kubernetes workflows.

1. Create a directory for the sample Flink application:

$ mkdir -p ~/flink-dataflow/sample-job/src/main/java/com/example
Enter fullscreen mode Exit fullscreen mode

2. Switch to the project directory:

$ cd ~/flink-dataflow/sample-job
Enter fullscreen mode Exit fullscreen mode

3. Create the Maven project configuration file:

$ nano pom.xml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration to the file.

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>flink-session-job</artifactId>
    <version>1.0.0</version>

    <properties>
        <maven.compiler.release>11</maven.compiler.release>
        <flink.version>1.20.0</flink.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.13.0</version>
            </plugin>
        </plugins>
    </build>
</project>
Enter fullscreen mode Exit fullscreen mode

The configuration:

  • Defines a Maven project for the sample Flink application.
  • Uses Apache Flink 1.20 libraries.
  • Configures Java 11 as the build target.
  • Packages the application into a deployable JAR.

Save and close the file.

4. Create a sample streaming application:

$ nano src/main/java/com/example/DataflowMigrationJob.java
Enter fullscreen mode Exit fullscreen mode

Add the following application.

package com.example;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class DataflowMigrationJob {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(
                "Apache Flink",
                "Kubernetes",
                "Stream Processing")
                .map(String::toUpperCase)
                .print();

        env.execute("Sample Streaming Job");
    }
}
Enter fullscreen mode Exit fullscreen mode

The application:

  • Creates a Flink streaming execution environment.
  • Processes a sample stream of events.
  • Converts each event to uppercase.
  • Prints the processed events to the application logs.

Save and close the file.

5. Build the application:

$ mvn clean package
Enter fullscreen mode Exit fullscreen mode

6. Verify that Maven creates the application JAR:

$ ls target
Enter fullscreen mode Exit fullscreen mode

Verify that the output includes the flink-session-job-1.0.0.jar file.

Note: The sample application demonstrates how to package a Flink job into a deployable JAR. The following steps submit an official Apache Flink example application hosted on Maven Central. Using a publicly accessible JAR provides a reproducible deployment without requiring additional artifact storage or container image customization. Later sections configure durable storage for Flink applications and state.

7. Switch to the Flink manifest directory:

$ cd ~/flink-dataflow
Enter fullscreen mode Exit fullscreen mode

8. Create the session job manifest:

$ nano session-job.yaml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration:

apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
metadata:
  name: sample-streaming-job
  namespace: flink
spec:
  deploymentName: flink-session-cluster
  job:
    jarURI: https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming/1.20.0/flink-examples-streaming-1.20.0-TopSpeedWindowing.jar
    parallelism: 2
    upgradeMode: stateless
    state: running
Enter fullscreen mode Exit fullscreen mode

The configuration:

  • Submits a streaming application to the existing Flink Session Cluster.
  • Downloads the application JAR from Maven Central.
  • Runs the application with a parallelism of 2.
  • Uses the stateless upgrade mode for application updates.

Save and close the file.

9. Apply the session job manifest:

$ kubectl apply -f session-job.yaml
Enter fullscreen mode Exit fullscreen mode

10. Verify that the session job is created:

$ kubectl get flinksessionjobs -n flink
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                   JOB STATUS   LIFECYCLE STATE
sample-streaming-job   RUNNING      STABLE
Enter fullscreen mode Exit fullscreen mode

11. Verify that the Session Cluster runs the submitted job:

$ kubectl get flinkdeployments -n flink
Enter fullscreen mode Exit fullscreen mode

12. Set up port forwarding to access the Flink dashboard:

$ kubectl port-forward svc/flink-session-cluster-rest -n flink 8081:8081
Enter fullscreen mode Exit fullscreen mode

13. Open http://localhost:8081 in your web browser to access the Apache Flink dashboard. The Flink dashboard displays running jobs, TaskManagers, checkpoints, and cluster resources. Verify that the submitted job appears in the Running Jobs section with a RUNNING status. Select the job to view execution details and task metrics.

Deploy a Flink Application Cluster

A Flink Application Cluster creates a dedicated cluster for a single application. It deploys the application with a JobManager and TaskManagers. Each application uses dedicated resources and has an independent lifecycle.

1. Create a Kubernetes Secret using your Object Storage credentials. Replace YOUR_ACCESS_KEY with your Object Storage access key and YOUR_SECRET_KEY with your Object Storage secret key.

$ kubectl create secret generic flink-object-storage \
    --namespace flink \
    --from-literal=AWS_ACCESS_KEY_ID='YOUR_ACCESS_KEY' \
    --from-literal=AWS_SECRET_ACCESS_KEY='YOUR_SECRET_KEY'
Enter fullscreen mode Exit fullscreen mode

2. Verify that the Secret exists:

$ kubectl get secret flink-object-storage -n flink
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                   TYPE     DATA   AGE
flink-object-storage   Opaque   2      31s
Enter fullscreen mode Exit fullscreen mode

3. Create the application cluster manifest:

$ nano flink-application.yaml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration. Replace YOUR_BUCKET_NAME with your Object Storage bucket name and YOUR_OBJECT_STORAGE_HOSTNAME with your Object Storage hostname.

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: flink-application-cluster
  namespace: flink
spec:
  image: flink:1.20
  flinkVersion: v1_20
  mode: native
  serviceAccount: flink

  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"

    state.backend.type: rocksdb
    execution.checkpointing.interval: "60 s"
    execution.checkpointing.mode: EXACTLY_ONCE
    execution.checkpointing.dir: s3://YOUR_BUCKET_NAME/checkpoints
    execution.checkpointing.savepoint-dir: s3://YOUR_BUCKET_NAME/savepoints
    execution.checkpointing.incremental: "true"
    execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
    state.checkpoints.num-retained: "3"

    s3.endpoint: YOUR_OBJECT_STORAGE_HOSTNAME
    s3.path.style.access: "true"

    restart-strategy.type: fixed-delay
    restart-strategy.fixed-delay.attempts: "3"
    restart-strategy.fixed-delay.delay: "10 s"

  podTemplate:
    spec:
      initContainers:
        - name: enable-s3-plugin
          image: flink:1.20
          command:
            - /bin/sh
            - -c
            - |
              mkdir -p /flink-plugins/s3-fs-presto
              cp /opt/flink/opt/flink-s3-fs-presto-*.jar \
                /flink-plugins/s3-fs-presto/
          volumeMounts:
            - name: flink-plugins
              mountPath: /flink-plugins

      containers:
        - name: flink-main-container
          env:
            - name: AWS_ACCESS_KEY_ID
              valueFrom:
                secretKeyRef:
                  name: flink-object-storage
                  key: AWS_ACCESS_KEY_ID
            - name: AWS_SECRET_ACCESS_KEY
              valueFrom:
                secretKeyRef:
                  name: flink-object-storage
                  key: AWS_SECRET_ACCESS_KEY
          volumeMounts:
            - name: flink-plugins
              mountPath: /opt/flink/plugins

      volumes:
        - name: flink-plugins
          emptyDir: {}

  job:
    jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
    parallelism: 2
    upgradeMode: savepoint
    state: running

  jobManager:
    resource:
      cpu: 1
      memory: "2048m"

  taskManager:
    replicas: 2
    resource:
      cpu: 1
      memory: "2048m"
Enter fullscreen mode Exit fullscreen mode

The configuration:

  • Creates a dedicated Flink Application Cluster.
  • Uses RocksDB as the state backend.
  • Creates an incremental checkpoint every 60 seconds.
  • Stores checkpoints and savepoints in Object Storage.
  • Retains completed checkpoints when the application is canceled.
  • Loads Object Storage credentials from a Kubernetes Secret.
  • Enables the Flink S3 filesystem plugin.
  • Uses savepoints to preserve application state during updates.
  • Retries failed jobs three times with a ten-second delay.

Save and close the file.

4. Apply the manifest:

$ kubectl apply -f flink-application.yaml
Enter fullscreen mode Exit fullscreen mode

5. Verify that the FlinkDeployment reaches the STABLE lifecycle state:

$ kubectl get flinkdeployments -n flink
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                        JOB STATUS   LIFECYCLE STATE
flink-application-cluster   RUNNING      STABLE
flink-session-cluster                    STABLE
Enter fullscreen mode Exit fullscreen mode

6. Verify that the JobManager and TaskManager pods are running:

$ kubectl get pods -n flink
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                         READY   STATUS    RESTARTS   AGE
flink-application-cluster-6d9cd654c7-gx7vj   1/1     Running   0          11m
flink-application-cluster-taskmanager-1-1    1/1     Running   0          10m
flink-application-cluster-taskmanager-1-2    1/1     Running   0          10m
flink-session-cluster-7cb847b8cf-wx5px       1/1     Running   0          117m
flink-session-cluster-taskmanager-1-1        1/1     Running   0          22m
Enter fullscreen mode Exit fullscreen mode

7. Verify the services created for the Application Cluster:

$ kubectl get svc -n flink
Enter fullscreen mode Exit fullscreen mode

The flink-application-cluster-rest service exposes the Flink REST API and Web UI.

Configure State Management

Apache Flink uses a state backend to manage application state during processing. The HashMap state backend keeps state in JVM memory and works well for smaller workloads. RocksDB stores working state on TaskManager disks and supports incremental checkpoints for larger workloads. The Application Cluster uses RocksDB with Object Storage to preserve checkpoints and savepoints outside the Kubernetes cluster.

1. Verify the state backend, checkpoint directory, and incremental checkpointing configured for the Application Cluster:

$ kubectl get flinkdeployment flink-application-cluster \
    -n flink \
    -o jsonpath='backend={.spec.flinkConfiguration.state\.backend\.type} dir={.spec.flinkConfiguration.execution\.checkpointing\.dir} incremental={.spec.flinkConfiguration.execution\.checkpointing\.incremental}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

backend=rocksdb dir=s3://YOUR_BUCKET_NAME/checkpoints incremental=true
Enter fullscreen mode Exit fullscreen mode

2. Verify that the application creates checkpoints successfully:

$ kubectl logs deployment/flink-application-cluster \
    -n flink \
    --since=10m | grep "Completed checkpoint"
Enter fullscreen mode Exit fullscreen mode

Output:

INFO  CheckpointCoordinator [] - Completed checkpoint 182 for job 462df3b4543929748a7ede4b313ab797 (784937 bytes, checkpointDuration=14773 ms, finalizationTime=1 ms).
INFO  CheckpointCoordinator [] - Completed checkpoint 183 for job 462df3b4543929748a7ede4b313ab797 (288117 bytes, checkpointDuration=7367 ms, finalizationTime=1 ms).
Enter fullscreen mode Exit fullscreen mode

The output confirms that Flink creates checkpoints successfully. The checkpoint directory verified earlier stores the completed checkpoints in Object Storage.

3. Trigger a savepoint for the running application:

$ kubectl patch flinkdeployment flink-application-cluster \
    -n flink \
    --type merge \
    -p '{"spec":{"job":{"savepointTriggerNonce":1}}}'
Enter fullscreen mode Exit fullscreen mode

4. Monitor the state snapshots:

$ kubectl get flinkstatesnapshots -n flink --watch
Enter fullscreen mode Exit fullscreen mode

Wait until the latest snapshot reaches the COMPLETED state. Press Ctrl+C to stop monitoring.

5. Store the latest state snapshot name in a variable:

$ SNAPSHOT_NAME=$(kubectl get flinkstatesnapshots \
    -n flink \
    --sort-by=.metadata.creationTimestamp \
    -o jsonpath='{.items[-1:].metadata.name}')
Enter fullscreen mode Exit fullscreen mode

6. Verify the snapshot state:

$ kubectl get flinkstatesnapshot "$SNAPSHOT_NAME" \
    -n flink \
    -o jsonpath='{.status.state}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

COMPLETED
Enter fullscreen mode Exit fullscreen mode

7. View the savepoint location:

$ kubectl get flinkstatesnapshot "$SNAPSHOT_NAME" \
    -n flink \
    -o jsonpath='{.status.path}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

s3://YOUR_BUCKET_NAME/savepoints/savepoint-xxxxxxxxxxxxxxxx
Enter fullscreen mode Exit fullscreen mode

Set Up Windowing and Event-Time Processing

Apache Flink uses event timestamps and watermarks to process records that arrive out of order. Windowing groups records by time, while allowed lateness and side outputs handle delayed records.

1. Switch to the sample application directory:

$ cd ~/flink-dataflow/sample-job
Enter fullscreen mode Exit fullscreen mode

2. Create an event-time windowing application:

$ nano src/main/java/com/example/EventTimeWindowingJob.java
Enter fullscreen mode Exit fullscreen mode

Add the following application.

package com.example;

import java.time.Duration;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.util.OutputTag;

public class EventTimeWindowingJob {

    private static final OutputTag<Event> LATE_EVENTS =
            new OutputTag<Event>("late-events") {};

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        long baseTime = 1_700_000_000_000L;

        DataStream<Event> events = env
                .fromElements(
                        new Event("checkout", 10L, baseTime),
                        new Event("checkout", 15L, baseTime + 10_000),
                        new Event("payment", 20L, baseTime + 20_000),
                        new Event("checkout", 12L, baseTime + 30_000),
                        new Event("payment", 25L, baseTime + 40_000),

                        // Advances the watermark beyond the first window.
                        new Event("checkout", 30L, baseTime + 120_000),

                        // Arrives after the watermark and allowed-lateness period.
                        new Event("checkout", 5L, baseTime + 5_000))
                .returns(Types.POJO(Event.class));

        WatermarkStrategy<Event> watermarkStrategy =
                WatermarkStrategy
                        .<Event>forBoundedOutOfOrderness(
                                Duration.ofSeconds(5))
                        .withTimestampAssigner(
                                (event, previousTimestamp) ->
                                        event.timestamp);

        DataStream<Event> timestampedEvents =
                events.assignTimestampsAndWatermarks(
                        watermarkStrategy);

        SingleOutputStreamOperator<Event> tumblingResults =
                timestampedEvents
                        .keyBy(event -> event.type)
                        .window(
                                TumblingEventTimeWindows.of(
                                        Duration.ofMinutes(1)))
                        .allowedLateness(Duration.ofSeconds(30))
                        .sideOutputLateData(LATE_EVENTS)
                        .reduce(new SumEvents())
                        .name("One-minute tumbling window");

        DataStream<Event> slidingResults =
                timestampedEvents
                        .keyBy(event -> event.type)
                        .window(
                                SlidingEventTimeWindows.of(
                                        Duration.ofMinutes(5),
                                        Duration.ofMinutes(1)))
                        .reduce(new SumEvents())
                        .name("Five-minute sliding window");

        DataStream<Event> sessionResults =
                timestampedEvents
                        .keyBy(event -> event.type)
                        .window(
                                EventTimeSessionWindows.withGap(
                                        Duration.ofMinutes(2)))
                        .reduce(new SumEvents())
                        .name("Two-minute session window");

        DataStream<Event> lateEvents =
                tumblingResults.getSideOutput(LATE_EVENTS);

        tumblingResults.print("Tumbling window");
        slidingResults.print("Sliding window");
        sessionResults.print("Session window");
        lateEvents.print("Late event");

        env.execute("Event-Time Windowing Job");
    }

    public static class Event {
        public String type;
        public long value;
        public long timestamp;

        public Event() {
        }

        public Event(String type, long value, long timestamp) {
            this.type = type;
            this.value = value;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "Event{"
                    + "type='" + type + '\''
                    + ", value=" + value
                    + ", timestamp=" + timestamp
                    + '}';
        }
    }

    public static class SumEvents
            implements ReduceFunction<Event> {

        @Override
        public Event reduce(Event first, Event second) {
            return new Event(
                    first.type,
                    first.value + second.value,
                    Math.max(first.timestamp, second.timestamp));
        }
    }
}
Enter fullscreen mode Exit fullscreen mode

The application:

  • Extracts the event time from each record.
  • Allows records to arrive up to five seconds out of order.
  • Groups records into one-minute tumbling windows.
  • Creates five-minute sliding windows every minute.
  • Groups records into sessions using a two-minute inactivity gap.
  • Accepts records until the watermark advances 30 seconds beyond the end of a tumbling window.
  • Sends records that exceed the allowed lateness to a side output.

Save and close the file.

3. Build the application:

$ mvn clean package
Enter fullscreen mode Exit fullscreen mode

4. Verify that Maven creates the application JAR:

$ ls -lh target/flink-session-job-1.0.0.jar
Enter fullscreen mode Exit fullscreen mode

Output:

-rw-r--r-- 1 user user 8.5K Jul 13 00:15 target/flink-session-job-1.0.0.jar
Enter fullscreen mode Exit fullscreen mode

Configure Exactly-Once Processing

Apache Flink combines checkpoints with transactional sinks to provide end-to-end exactly-once processing. A Kafka transactional sink writes records inside a transaction and commits it only after the corresponding checkpoint completes, so a job that fails and restarts from the last checkpoint never emits duplicate output to downstream consumers.

Deploy Apache Kafka

End-to-end exactly-once processing requires a message broker with transactional support. Deploy a three-node Apache Kafka cluster with the Strimzi operator to provide the transactional source and sink for the Flink application.

1. Create a namespace for Kafka resources:

$ kubectl create namespace kafka
Enter fullscreen mode Exit fullscreen mode

2. Add the Strimzi Helm repository:

$ helm repo add strimzi https://strimzi.io/charts/
Enter fullscreen mode Exit fullscreen mode

3. Update the Helm repository information:

$ helm repo update
Enter fullscreen mode Exit fullscreen mode

4. Install the Strimzi Kafka Operator:

$ helm install strimzi-kafka-operator \
    strimzi/strimzi-kafka-operator \
    --namespace kafka \
    --version 1.1.0
Enter fullscreen mode Exit fullscreen mode

5. Verify that the Strimzi operator is running:

$ kubectl get pods -n kafka
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                        READY   STATUS    RESTARTS   AGE
strimzi-cluster-operator-687687c99b-l695z   1/1     Running   0          17m
Enter fullscreen mode Exit fullscreen mode

6. Create a Kafka cluster manifest:

$ nano kafka-cluster.yaml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration.

apiVersion: kafka.strimzi.io/v1
kind: KafkaNodePool
metadata:
  name: dual-role
  namespace: kafka
  labels:
    strimzi.io/cluster: flink-kafka
spec:
  replicas: 3
  roles:
    - controller
    - broker
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 10Gi
        deleteClaim: false
  resources:
    requests:
      cpu: 500m
      memory: 1Gi
    limits:
      cpu: "1"
      memory: 2Gi
  template:
    pod:
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
            - weight: 100
              podAffinityTerm:
                topologyKey: kubernetes.io/hostname
                labelSelector:
                  matchLabels:
                    strimzi.io/cluster: flink-kafka
---
apiVersion: kafka.strimzi.io/v1
kind: Kafka
metadata:
  name: flink-kafka
  namespace: kafka
spec:
  kafka:
    version: 4.3.0
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
    config:
      auto.create.topics.enable: false
      default.replication.factor: 3
      min.insync.replicas: 2
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
  entityOperator:
    topicOperator: {}
    userOperator: {}
Enter fullscreen mode Exit fullscreen mode

The node pool runs three Kafka nodes with persistent storage. Each node acts as a broker and KRaft controller. The replication and in-sync replica settings allow the cluster to continue processing transactions when one broker is unavailable. Strimzi requires Kafka and KafkaNodePool resources for KRaft-based clusters.

Save and close the file.

7. Deploy the Kafka cluster:

$ kubectl apply -f kafka-cluster.yaml
Enter fullscreen mode Exit fullscreen mode

8. Wait for the Kafka cluster to become ready:

$ kubectl wait kafka/flink-kafka \
    -n kafka \
    --for=condition=Ready \
    --timeout=600s
Enter fullscreen mode Exit fullscreen mode

9. Verify the Kafka pods:

$ kubectl get pods -n kafka \
    -l strimzi.io/cluster=flink-kafka
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                           READY   STATUS    RESTARTS   AGE
flink-kafka-dual-role-0                        1/1     Running   0          98s
flink-kafka-dual-role-1                        1/1     Running   0          98s
flink-kafka-dual-role-2                        1/1     Running   0          98s
flink-kafka-entity-operator-5d6b58bffc-nq7vz   2/2     Running   0          34s
Enter fullscreen mode Exit fullscreen mode

10. Verify the internal Kafka bootstrap service:

$ kubectl get service flink-kafka-kafka-bootstrap \
    -n kafka
Enter fullscreen mode Exit fullscreen mode

Create the Kafka Topics

Create the input topic for incoming events and the output topic for window results. Each topic uses three partitions and three replicas, and requires at least two in-sync replicas to acknowledge writes, so it tolerates one unavailable broker.

1. Create the Kafka topics manifest:

$ nano flink-topics.yaml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration.

apiVersion: kafka.strimzi.io/v1
kind: KafkaTopic
metadata:
  name: flink-events
  namespace: kafka
  labels:
    strimzi.io/cluster: flink-kafka
spec:
  partitions: 3
  replicas: 3
  config:
    min.insync.replicas: 2
    retention.ms: 604800000
---
apiVersion: kafka.strimzi.io/v1
kind: KafkaTopic
metadata:
  name: flink-window-results
  namespace: kafka
  labels:
    strimzi.io/cluster: flink-kafka
spec:
  partitions: 3
  replicas: 3
  config:
    min.insync.replicas: 2
    retention.ms: 604800000
Enter fullscreen mode Exit fullscreen mode

Save and close the file.

2. Create both topics:

$ kubectl apply -f flink-topics.yaml
Enter fullscreen mode Exit fullscreen mode

3. Wait for both topics to become ready:

$ kubectl wait kafkatopic/flink-events kafkatopic/flink-window-results \
    -n kafka \
    --for=condition=Ready \
    --timeout=180s
Enter fullscreen mode Exit fullscreen mode

4. Verify both topics:

$ kubectl get kafkatopic -n kafka
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                   CLUSTER       PARTITIONS   REPLICATION FACTOR   READY
flink-events           flink-kafka   3            3                    True
flink-window-results   flink-kafka   3            3                    True
Enter fullscreen mode Exit fullscreen mode

Enable Exactly-Once Checkpointing

1. Configure the Flink Session Cluster to create exactly-once checkpoints every 60 seconds:

$ kubectl patch flinkdeployment flink-session-cluster \
    -n flink \
    --type merge \
    -p '{
      "spec": {
        "flinkConfiguration": {
          "execution.checkpointing.interval": "60 s",
          "execution.checkpointing.mode": "EXACTLY_ONCE"
        }
      }
    }'
Enter fullscreen mode Exit fullscreen mode

2. Wait for the Session Cluster to become stable:

$ kubectl wait flinkdeployment/flink-session-cluster \
    -n flink \
    --for=jsonpath='{.status.lifecycleState}'=STABLE \
    --timeout=300s
Enter fullscreen mode Exit fullscreen mode

3. Verify the checkpointing mode and interval:

$ kubectl get flinkdeployment flink-session-cluster \
    -n flink \
    -o jsonpath='mode={.spec.flinkConfiguration.execution\.checkpointing\.mode} interval={.spec.flinkConfiguration.execution\.checkpointing\.interval}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

mode=EXACTLY_ONCE interval=60 s
Enter fullscreen mode Exit fullscreen mode

Add the Kafka Connector

1. Switch to the sample application directory:

$ cd ~/flink-dataflow/sample-job
Enter fullscreen mode Exit fullscreen mode

2. Open the Maven project configuration:

$ nano pom.xml
Enter fullscreen mode Exit fullscreen mode

Append the following dependencies to the existing <dependencies> block.

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-base</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>3.4.0-1.20</version>
</dependency>
Enter fullscreen mode Exit fullscreen mode

The connector base dependency provides the DeliveryGuarantee class. The Kafka connector provides the transactional sink and supports Apache Flink 1.20.

Append the following plugin to the existing <plugins> block.

<plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-shade-plugin</artifactId>
    <version>3.6.0</version>
    <executions>
        <execution>
            <phase>package</phase>
            <goals>
                <goal>shade</goal>
            </goals>
            <configuration>
                <createDependencyReducedPom>false</createDependencyReducedPom>
                <filters>
                    <filter>
                        <artifact>*:*</artifact>
                        <excludes>
                            <exclude>META-INF/*.SF</exclude>
                            <exclude>META-INF/*.DSA</exclude>
                            <exclude>META-INF/*.RSA</exclude>
                        </excludes>
                    </filter>
                </filters>
                <transformers>
                    <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                        <mainClass>com.example.EventTimeWindowingJob</mainClass>
                    </transformer>
                </transformers>
            </configuration>
        </execution>
    </executions>
</plugin>
Enter fullscreen mode Exit fullscreen mode

The Maven Shade plugin packages the Kafka connector and its required libraries in the application JAR. The Flink runtime and connector base dependencies remain provided by the Flink cluster.

Save and close the file.

Configure the Kafka Source and Transactional Sink

1. Update the event-time windowing application to add the Kafka source and transactional sink:

$ nano src/main/java/com/example/EventTimeWindowingJob.java
Enter fullscreen mode Exit fullscreen mode

Append the following imports to the existing import statements.

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
Enter fullscreen mode Exit fullscreen mode

Locate the following class declaration:

public class EventTimeWindowingJob {
Enter fullscreen mode Exit fullscreen mode

Append the following constant immediately after the class declaration:

private static final String KAFKA_BOOTSTRAP_SERVERS =
        "flink-kafka-kafka-bootstrap.kafka.svc.cluster.local:9092";
Enter fullscreen mode Exit fullscreen mode

Locate the hardcoded event source that begins with:

long baseTime = 1_700_000_000_000L;
Enter fullscreen mode Exit fullscreen mode

Remove the baseTime declaration and the complete env.fromElements() block. Replace them with the following Kafka source configuration.

KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
        .setBootstrapServers(KAFKA_BOOTSTRAP_SERVERS)
        .setTopics("flink-events")
        .setGroupId("flink-windowing-job")
        .setStartingOffsets(OffsetsInitializer.earliest())
        .setValueOnlyDeserializer(new SimpleStringSchema())
        .build();

DataStream<Event> events = env
        .fromSource(
                kafkaSource,
                WatermarkStrategy.noWatermarks(),
                "Kafka input source")
        .map(Event::fromCsv)
        .returns(Types.POJO(Event.class));
Enter fullscreen mode Exit fullscreen mode

The Kafka source reads comma-separated event records from the flink-events topic. Flink stores the consumed Kafka offsets in checkpoints so the application can resume consistently after a restart or failure.

Locate the following existing watermark block:

WatermarkStrategy<Event> watermarkStrategy =
        WatermarkStrategy
                .<Event>forBoundedOutOfOrderness(
                        Duration.ofSeconds(5))
                .withTimestampAssigner(
                        (event, previousTimestamp) ->
                                event.timestamp);
Enter fullscreen mode Exit fullscreen mode

Replace it with:

WatermarkStrategy<Event> watermarkStrategy =
        WatermarkStrategy
                .<Event>forBoundedOutOfOrderness(
                        Duration.ofSeconds(5))
                .withTimestampAssigner(
                        (event, previousTimestamp) ->
                                event.timestamp)
                .withIdleness(Duration.ofSeconds(30));
Enter fullscreen mode Exit fullscreen mode

Locate the following constructor inside the Event class:

public Event(String type, long value, long timestamp) {
    this.type = type;
    this.value = value;
    this.timestamp = timestamp;
}
Enter fullscreen mode Exit fullscreen mode

Append the following method after the constructor:

public static Event fromCsv(String record) {
    String[] fields = record.trim().split(",");

    if (fields.length != 3) {
        throw new IllegalArgumentException(
                "Expected record format: type,value,timestamp");
    }

    return new Event(
            fields[0].trim(),
            Long.parseLong(fields[1].trim()),
            Long.parseLong(fields[2].trim()));
}
Enter fullscreen mode Exit fullscreen mode

Locate the following declaration:

DataStream<Event> lateEvents =
        tumblingResults.getSideOutput(LATE_EVENTS);
Enter fullscreen mode Exit fullscreen mode

Append the following Kafka sink configuration after the declaration.

KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
        .setBootstrapServers(KAFKA_BOOTSTRAP_SERVERS)
        .setRecordSerializer(
                KafkaRecordSerializationSchema.builder()
                        .setTopic("flink-window-results")
                        .setValueSerializationSchema(
                                new SimpleStringSchema())
                        .build())
        .setDeliveryGuarantee(
                DeliveryGuarantee.EXACTLY_ONCE)
        .setTransactionalIdPrefix(
                "flink-windowing-")
        .setProperty(
                "transaction.timeout.ms",
                "900000")
        .build();
Enter fullscreen mode Exit fullscreen mode

The bootstrap address connects the Flink application to the internal Kafka service. The transactional ID prefix must be unique for each running Flink application that writes to the Kafka cluster.

The transaction timeout is set to 15 minutes. This gives Flink enough time to recover and complete a checkpoint before Kafka expires an open transaction.

Locate the following statement:

lateEvents.print("Late event");
Enter fullscreen mode Exit fullscreen mode

Append the following sink after the statement and before env.execute().

tumblingResults
        .map(Event::toString)
        .returns(Types.STRING)
        .sinkTo(kafkaSink)
        .name("Exactly-once Kafka sink");
Enter fullscreen mode Exit fullscreen mode

The sink writes tumbling-window results to Kafka transactions. Flink commits each transaction only after the corresponding checkpoint completes. Consumers using the read_committed isolation level do not receive records from aborted transactions.

Save and close the file.

The application:

  • Uses exactly-once checkpoints to maintain consistent state.
  • Writes window results to the flink-window-results topic.
  • Commits Kafka transactions after successful checkpoints.
  • Aborts incomplete transactions during recovery.
  • Allows consumers configured with the read_committed isolation level to ignore uncommitted and aborted records.

2. Build the application:

$ mvn clean package
Enter fullscreen mode Exit fullscreen mode

3. Verify that Maven compiled the application for Java 11:

$ javap -verbose \
    target/classes/com/example/EventTimeWindowingJob.class \
    | grep "major version"
Enter fullscreen mode Exit fullscreen mode

The output shows major version: 55, which corresponds to Java 11.

4. Verify that the JAR bundles the Kafka connector classes:

$ jar tf target/flink-session-job-1.0.0.jar | grep -E 'kafka/(sink/KafkaSink|source/KafkaSource)\.class'
Enter fullscreen mode Exit fullscreen mode

The output lists the KafkaSink and KafkaSource classes, which confirms the connector is shaded into the JAR.

Deploy the Updated Flink Application

The Maven build stores the updated JAR on the local computer. Upload and run the JAR on the Flink Session Cluster.

1. Verify that the Flink REST service exists:

$ kubectl get service flink-session-cluster-rest \
    -n flink
Enter fullscreen mode Exit fullscreen mode

2. Forward local port 8081 to the Flink REST service:

$ kubectl port-forward \
    service/flink-session-cluster-rest \
    -n flink \
    8081:8081 > /tmp/flink-port-forward.log 2>&1 &
Enter fullscreen mode Exit fullscreen mode

3. Store the port-forward process ID:

$ PORT_FORWARD_PID=$!
Enter fullscreen mode Exit fullscreen mode

4. Wait for the Flink REST endpoint:

$ until curl --silent --fail \
    http://localhost:8081/overview > /dev/null; do
    sleep 2
done
Enter fullscreen mode Exit fullscreen mode

5. Upload the application JAR:

$ UPLOAD_RESPONSE=$(curl --silent --show-error --fail \
    --request POST \
    --header "Expect:" \
    --form "jarfile=@target/flink-session-job-1.0.0.jar" \
    http://localhost:8081/jars/upload)
Enter fullscreen mode Exit fullscreen mode

6. Display the upload response:

$ echo "$UPLOAD_RESPONSE"
Enter fullscreen mode Exit fullscreen mode

Output:

{"filename":"/tmp/flink-web-f92ea68e-87f4-453b-ad8f-e97f6b6cb6ff/flink-web-upload/6467357d-4412-400a-8f1a-d4652ce3dddf_flink-session-job-1.0.0.jar","status":"success"}
Enter fullscreen mode Exit fullscreen mode

7. Extract the uploaded JAR identifier:

$ JAR_ID=$(printf '%s' "$UPLOAD_RESPONSE" \
    | sed -E 's/.*"filename":"[^"]*\/([^"]+)".*/\1/')
Enter fullscreen mode Exit fullscreen mode

8. Verify the JAR identifier:

$ echo "$JAR_ID"
Enter fullscreen mode Exit fullscreen mode

Output:

6467357d-4412-400a-8f1a-d4652ce3dddf_flink-session-job-1.0.0.jar
Enter fullscreen mode Exit fullscreen mode

9. Submit the event-time application:

$ RUN_RESPONSE=$(curl --silent --show-error --fail \
    --request POST \
    --header "Content-Type: application/json" \
    --data '{
      "entryClass": "com.example.EventTimeWindowingJob",
      "parallelism": 2
    }' \
    "http://localhost:8081/jars/${JAR_ID}/run")
Enter fullscreen mode Exit fullscreen mode

10. Display the submission response:

$ echo "$RUN_RESPONSE"
Enter fullscreen mode Exit fullscreen mode

Output:

{"jobid":"cd7f0424cc4ce10e7e797cd0c55470a7"}
Enter fullscreen mode Exit fullscreen mode

11. Extract the Flink job ID:

$ JOB_ID=$(printf '%s' "$RUN_RESPONSE" \
    | sed -E 's/.*"jobid":"([^"]+)".*/\1/')
Enter fullscreen mode Exit fullscreen mode

12. Verify the job ID:

$ echo "$JOB_ID"
Enter fullscreen mode Exit fullscreen mode

Output:

cd7f0424cc4ce10e7e797cd0c55470a7
Enter fullscreen mode Exit fullscreen mode

13. Check the application state:

$ curl --silent --show-error \
    "http://localhost:8081/jobs/${JOB_ID}" \
    | grep -o '"state":"[^"]*"'
Enter fullscreen mode Exit fullscreen mode

Output:

"state":"RUNNING"
Enter fullscreen mode Exit fullscreen mode

A RUNNING state confirms that Flink submitted the application successfully and started its processing tasks.

14. Stop port forwarding:

$ kill "$PORT_FORWARD_PID"
Enter fullscreen mode Exit fullscreen mode

Stopping the local port-forward does not stop the Flink job. It only closes the local connection to the Flink REST service.

Run Apache Beam Pipelines on Flink

Apache Beam separates pipeline code from the processing engine that runs it. The Flink Runner translates a Beam pipeline into a Flink job and submits it to an existing Flink cluster.

Create the Apache Beam Project

1. Switch to the main project directory:

$ cd ~/flink-dataflow
Enter fullscreen mode Exit fullscreen mode

2. Create a Maven project for the Beam pipeline:

$ mkdir -p beam-job/src/main/java/com/example
Enter fullscreen mode Exit fullscreen mode

3. Switch to the Beam project directory:

$ cd beam-job
Enter fullscreen mode Exit fullscreen mode

4. Create the Maven project configuration:

$ nano pom.xml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration.

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>beam-flink-job</artifactId>
    <version>1.0.0</version>

    <properties>
        <maven.compiler.release>11</maven.compiler.release>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <beam.version>2.75.0</beam.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.beam</groupId>
            <artifactId>beam-sdks-java-core</artifactId>
            <version>${beam.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.beam</groupId>
            <artifactId>beam-runners-flink-1.20</artifactId>
            <version>${beam.version}</version>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>2.0.16</version>
            <scope>runtime</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.13.0</version>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.6.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <createDependencyReducedPom>false</createDependencyReducedPom>

                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>

                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.example.BeamFlinkPipeline</mainClass>
                                </transformer>

                                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>
Enter fullscreen mode Exit fullscreen mode

The configuration:

  • Java 11 keeps the application compatible with the Flink cluster.
  • The Beam SDK provides the pipeline API and transforms.
  • The Flink Runner submits Beam pipelines to Flink 1.20.x.
  • SLF4J provides command-line logging for the Beam client.
  • The Maven Shade plugin creates a runnable JAR with the required dependencies.
  • The transformers set the main class and preserve the service files required by Beam.

Save and close the file.

Create the Beam Pipeline

1. Create the Beam application:

$ nano src/main/java/com/example/BeamFlinkPipeline.java
Enter fullscreen mode Exit fullscreen mode

Add the following application.

package com.example;

import java.util.Arrays;

import org.apache.beam.runners.flink.FlinkPipelineOptions;
import org.apache.beam.runners.flink.FlinkRunner;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.FlatMapElements;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;

public class BeamFlinkPipeline {

    public static void main(String[] args) {
        FlinkPipelineOptions options =
                PipelineOptionsFactory
                        .fromArgs(args)
                        .withValidation()
                        .as(FlinkPipelineOptions.class);

        options.setRunner(FlinkRunner.class);

        Pipeline pipeline = Pipeline.create(options);

        pipeline
                .apply(
                        "Create records",
                        Create.of(
                                "flink processes streaming data",
                                "beam runs on flink",
                                "flink provides fault tolerance"))
                .apply(
                        "Split words",
                        FlatMapElements
                                .into(TypeDescriptors.strings())
                                .via(line -> Arrays.asList(
                                        line.toLowerCase().split("\\s+"))))
                .apply(
                        "Count words",
                        Count.perElement())
                .apply(
                        "Format results",
                        MapElements
                                .into(TypeDescriptors.strings())
                                .via(result ->
                                        result.getKey()
                                                + ": "
                                                + result.getValue()));

        PipelineResult result = pipeline.run();
        result.waitUntilFinish();
    }
}
Enter fullscreen mode Exit fullscreen mode

The pipeline:

  • Creates a bounded collection of text records.
  • Splits each record into words.
  • Counts each word.
  • Formats the word-count results.
  • Waits until the bounded pipeline finishes.

Save and close the file.

Build the Beam Pipeline

1. Build the application JAR:

$ mvn clean package
Enter fullscreen mode Exit fullscreen mode

The Maven Shade plugin may display warnings about overlapping manifests, licenses, classes, and metadata. The build is successful when Maven returns:

BUILD SUCCESS
Enter fullscreen mode Exit fullscreen mode

2. Verify that Maven created the application JAR:

$ ls -lh target/beam-flink-job-1.0.0.jar
Enter fullscreen mode Exit fullscreen mode

Output:

-rw-r--r--  1 darshansiroya  staff   162M 15 Jul 02:09 target/beam-flink-job-1.0.0.jar
Enter fullscreen mode Exit fullscreen mode

3. Verify that Maven compiled the application for Java 11:

$ javap -verbose \
    target/classes/com/example/BeamFlinkPipeline.class \
    | grep "major version"
Enter fullscreen mode Exit fullscreen mode

Output:

major version: 55
Enter fullscreen mode Exit fullscreen mode

4. Verify that the application JAR contains the Flink Runner:

$ jar tf target/beam-flink-job-1.0.0.jar \
    | grep 'FlinkRunner.class' \
    | head -n 1
Enter fullscreen mode Exit fullscreen mode

Output:

org/apache/beam/runners/flink/FlinkRunner.class
Enter fullscreen mode Exit fullscreen mode

Configure the Flink Runner

1. Verify that the Flink Session Cluster REST service exists:

$ kubectl get service flink-session-cluster-rest \
    -n flink
Enter fullscreen mode Exit fullscreen mode

2. Forward local port 8081 to the Flink REST service:

$ kubectl port-forward \
    service/flink-session-cluster-rest \
    -n flink \
    8081:8081 > /tmp/beam-flink-port-forward.log 2>&1 &
Enter fullscreen mode Exit fullscreen mode

3. Store the port-forward process ID:

$ BEAM_PORT_FORWARD_PID=$!
Enter fullscreen mode Exit fullscreen mode

4. Wait for the Flink REST endpoint:

$ until curl --silent --fail \
    http://localhost:8081/overview > /dev/null; do
    sleep 2
done
Enter fullscreen mode Exit fullscreen mode

Submit the Beam Pipeline

1. Create a unique identifier for the submission:

$ RUN_ID=$(date +%Y%m%d%H%M%S)
Enter fullscreen mode Exit fullscreen mode

2. Create a unique Beam job name:

$ BEAM_JOB_NAME="beam-flink-word-count-${RUN_ID}"
Enter fullscreen mode Exit fullscreen mode

3. Verify the generated job name:

$ echo "$BEAM_JOB_NAME"
Enter fullscreen mode Exit fullscreen mode

Output:

beam-flink-word-count-20260715021037
Enter fullscreen mode Exit fullscreen mode

4. Submit the pipeline through the Flink Runner:

$ java -jar target/beam-flink-job-1.0.0.jar \
    --runner=FlinkRunner \
    --flinkMaster=localhost:8081 \
    --parallelism=2 \
    --jobName="${BEAM_JOB_NAME}" \
    2>&1 | tee /tmp/beam-submission.log
Enter fullscreen mode Exit fullscreen mode

Note: Do not stop the command with Ctrl+C. The Beam client translates the pipeline, submits it to Flink, and waits for the bounded job to finish.

5. Extract the Flink job ID from the submission log:

$ BEAM_JOB_ID=$(sed -nE \
    "s/.*Successfully submitted job.*\(([0-9a-f]{32})\).*/\1/p" \
    /tmp/beam-submission.log \
    | tail -n 1)
Enter fullscreen mode Exit fullscreen mode

6. Verify the job ID:

$ test -n "$BEAM_JOB_ID" && \
    echo "$BEAM_JOB_ID"
Enter fullscreen mode Exit fullscreen mode

Output:

a690d68afc390fcebddb6fff968a5304
Enter fullscreen mode Exit fullscreen mode

Verify the Beam Pipeline

1. Verify the name of the submitted job:

$ curl --silent \
    "http://localhost:8081/jobs/${BEAM_JOB_ID}" \
    | grep -o '"name":"[^"]*"'
Enter fullscreen mode Exit fullscreen mode

2. Check the job state:

$ curl --silent \
    "http://localhost:8081/jobs/${BEAM_JOB_ID}" \
    | grep -o '"state":"[^"]*"'
Enter fullscreen mode Exit fullscreen mode

Output:

"state":"FINISHED"
Enter fullscreen mode Exit fullscreen mode

A FINISHED state confirms that the Flink Runner translated and executed the bounded Beam pipeline successfully.

3. Stop port forwarding:

$ kill "$BEAM_PORT_FORWARD_PID"
Enter fullscreen mode Exit fullscreen mode

Set Up Monitoring and Observability

Prometheus collects metrics from Flink components and stores them as time-series data. Grafana uses these metrics to display dashboards, while Alertmanager processes alerts generated by Prometheus.

Install Prometheus and Grafana

1. Create a namespace for the monitoring resources:

$ kubectl create namespace monitoring
Enter fullscreen mode Exit fullscreen mode

2. Add the Prometheus Community Helm repository:

$ helm repo add prometheus-community \
    https://prometheus-community.github.io/helm-charts
Enter fullscreen mode Exit fullscreen mode

3. Update the Helm repository information:

$ helm repo update
Enter fullscreen mode Exit fullscreen mode

4. Install the kube-prometheus-stack Helm chart:

$ helm install monitoring \
    prometheus-community/kube-prometheus-stack \
    --namespace monitoring \
    --version 72.0.0 \
    --wait \
    --timeout 15m
Enter fullscreen mode Exit fullscreen mode

Note: Do not stop the installation with Ctrl+C. The chart installs multiple workloads and Custom Resource Definitions and may take several minutes to complete.

Note: This guide pins version 72.0.0. If you change the version, confirm the Grafana sidecar loads dashboards, because some newer chart releases ship a sidecar image that fails TLS verification against the Kubernetes API server and silently loads no dashboards. Check the chart releases for the latest version.

The kube-prometheus-stack chart installs Prometheus Operator, Prometheus, Alertmanager, Grafana, Kubernetes metrics exporters, dashboards, and alerting rules.

5. Verify the Helm release:

$ helm status monitoring \
    --namespace monitoring
Enter fullscreen mode Exit fullscreen mode

6. Verify the monitoring pods:

$ kubectl get pods \
    --namespace monitoring
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                                     READY   STATUS    RESTARTS   AGE
alertmanager-monitoring-kube-prometheus-alertmanager-0   2/2     Running   0          22h
monitoring-grafana-59768694d4-z9vmm                      3/3     Running   0          22h
monitoring-kube-prometheus-operator-54cddb4bcf-7w5tk     1/1     Running   0          22h
monitoring-kube-state-metrics-797d5dd89-v6mrh            1/1     Running   0          22h
monitoring-prometheus-node-exporter-mv28t                1/1     Running   0          22h
monitoring-prometheus-node-exporter-phdzw                1/1     Running   0          22h
monitoring-prometheus-node-exporter-tbc6k                1/1     Running   0          22h
prometheus-monitoring-kube-prometheus-prometheus-0       2/2     Running   0          22h
Enter fullscreen mode Exit fullscreen mode

7. Verify that the Prometheus Operator Custom Resource Definitions exist:

$ kubectl get crd \
    servicemonitors.monitoring.coreos.com \
    prometheusrules.monitoring.coreos.com
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                    CREATED AT
servicemonitors.monitoring.coreos.com   2026-07-17T22:22:43Z
prometheusrules.monitoring.coreos.com   2026-07-17T22:22:42Z
Enter fullscreen mode Exit fullscreen mode

Enable the Prometheus Reporter in Flink

The Flink Prometheus reporter exposes JobManager and TaskManager metrics in a format that Prometheus can collect. The Flink image used by the Session Cluster already includes the Prometheus metrics plugin.

1. Verify that the Prometheus reporter JAR exists:

$ kubectl exec \
    -n flink \
    deploy/flink-session-cluster -- \
    sh -c \
    'find /opt/flink/plugins -iname "flink-metrics-prometheus-*.jar"'
Enter fullscreen mode Exit fullscreen mode

Output:

/opt/flink/plugins/metrics-prometheus/flink-metrics-prometheus-1.20.5.jar
Enter fullscreen mode Exit fullscreen mode

2. Configure the Session Cluster to expose Prometheus metrics on port 9249:

$ kubectl patch flinkdeployment flink-session-cluster \
    -n flink \
    --type merge \
    -p '{
      "spec": {
        "flinkConfiguration": {
          "metrics.reporter.prom.factory.class": "org.apache.flink.metrics.prometheus.PrometheusReporterFactory",
          "metrics.reporter.prom.port": "9249"
        },
        "jobManager": {
          "podTemplate": {
            "spec": {
              "containers": [
                {
                  "name": "flink-main-container",
                  "ports": [
                    {
                      "name": "metrics",
                      "containerPort": 9249,
                      "protocol": "TCP"
                    }
                  ]
                }
              ]
            }
          }
        },
        "taskManager": {
          "podTemplate": {
            "spec": {
              "containers": [
                {
                  "name": "flink-main-container",
                  "ports": [
                    {
                      "name": "metrics",
                      "containerPort": 9249,
                      "protocol": "TCP"
                    }
                  ]
                }
              ]
            }
          }
        }
      }
    }'
Enter fullscreen mode Exit fullscreen mode

The configuration:

  • Enables the Prometheus reporter in the Flink Session Cluster.
  • Uses port 9249 for the metrics endpoint.
  • Adds a named metrics port to the JobManager and TaskManager containers.
  • Preserves the existing checkpointing and resource settings.

Flink loads metric reporters through the metrics.reporter.<name>.factory.class setting. The Prometheus reporter exposes a pull-based endpoint and uses port 9249 by default.

3. Wait for the FlinkDeployment to become stable:

$ kubectl wait flinkdeployment/flink-session-cluster \
    -n flink \
    --for=jsonpath='{.status.lifecycleState}'=STABLE \
    --timeout=600s
Enter fullscreen mode Exit fullscreen mode

4. Verify that the Session Cluster pods are running:

$ kubectl get pods \
    -n flink \
    -l app=flink-session-cluster
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                     READY   STATUS    RESTARTS   AGE
flink-session-cluster-5754dddff9-h2rvx   1/1     Running   0          50s
flink-session-cluster-taskmanager-1-1    1/1     Running   0          16s
Enter fullscreen mode Exit fullscreen mode

5. Verify the reporter configuration:

$ kubectl get flinkdeployment flink-session-cluster \
    -n flink \
    -o jsonpath='{.spec.flinkConfiguration.metrics\.reporter\.prom\.factory\.class}{"\n"}{.spec.flinkConfiguration.metrics\.reporter\.prom\.port}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

org.apache.flink.metrics.prometheus.PrometheusReporterFactory
9249
Enter fullscreen mode Exit fullscreen mode

6. Verify that the Flink pods expose the metrics port:

$ kubectl get pods \
    -n flink \
    -l app=flink-session-cluster \
    -o jsonpath='{range .items[*]}{.metadata.name}{"\t"}{range .spec.containers[*].ports[*]}{.name}{": "}{.containerPort}{" "}{end}{"\n"}{end}'
Enter fullscreen mode Exit fullscreen mode

Output:

flink-session-cluster-5754dddff9-h2rvx  metrics: 9249 rest: 8081 jobmanager-rpc: 6123 blobserver: 6124
flink-session-cluster-taskmanager-1-1   metrics: 9249 taskmanager-rpc: 6122
Enter fullscreen mode Exit fullscreen mode

7. Test the Prometheus endpoint on each Flink pod:

$ for POD in $(kubectl get pods \
    -n flink \
    -l app=flink-session-cluster \
    -o jsonpath='{.items[*].metadata.name}'); do
    echo "Checking ${POD}"
    kubectl exec -n flink "$POD" -- \
        sh -c 'wget -qO- http://localhost:9249/metrics | head -n 5'
done
Enter fullscreen mode Exit fullscreen mode

Configure Prometheus Scraping

A ServiceMonitor-based configuration requires a Kubernetes Service to discover the Flink metrics endpoints. The ServiceMonitor then instructs the Prometheus Operator to scrape the named metrics port exposed by that Service.

1. Create a metrics Service manifest:

$ nano flink-metrics-service.yaml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration.

apiVersion: v1
kind: Service
metadata:
  name: flink-session-cluster-metrics
  namespace: flink
  labels:
    app: flink-session-cluster
    monitoring: flink
spec:
  clusterIP: None
  selector:
    app: flink-session-cluster
  ports:
    - name: metrics
      port: 9249
      targetPort: metrics
      protocol: TCP
Enter fullscreen mode Exit fullscreen mode

The Service:

  • Selects the Session Cluster JobManager and TaskManager pods.
  • Exposes the Flink Prometheus endpoint on port 9249.
  • Uses the named metrics container port configured earlier.
  • Uses a headless Service so Prometheus can discover each Flink pod as a separate target.

Save and close the file.

2. Create the metrics Service:

$ kubectl apply -f flink-metrics-service.yaml
Enter fullscreen mode Exit fullscreen mode

3. Verify the Service:

$ kubectl get service flink-session-cluster-metrics \
    -n flink
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                            TYPE        CLUSTER-IP   EXTERNAL-IP   PORT(S)    AGE
flink-session-cluster-metrics   ClusterIP   None         <none>        9249/TCP   12s
Enter fullscreen mode Exit fullscreen mode

4. Verify that the Service discovered the Flink pods:

$ kubectl get endpointslices \
    -n flink \
    -l kubernetes.io/service-name=flink-session-cluster-metrics
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                  ADDRESSTYPE   PORTS   ENDPOINTS                 AGE
flink-session-cluster-metrics-9rn5v   IPv4          9249    10.125.0.13,10.125.2.14   23s
Enter fullscreen mode Exit fullscreen mode

5. Create a ServiceMonitor manifest:

$ nano flink-service-monitor.yaml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration.

apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: flink-session-cluster
  namespace: monitoring
  labels:
    release: monitoring
spec:
  namespaceSelector:
    matchNames:
      - flink
  selector:
    matchLabels:
      monitoring: flink
  endpoints:
    - port: metrics
      path: /metrics
      interval: 30s
      scrapeTimeout: 10s
Enter fullscreen mode Exit fullscreen mode

The ServiceMonitor:

  • Discovers the Flink metrics Service in the flink namespace.
  • Selects the Service using the monitoring: flink label.
  • Scrapes the named metrics port every 30 seconds.
  • Uses the release: monitoring label so Prometheus selects the resource.

Save and close the file.

6. Create the ServiceMonitor:

$ kubectl apply -f flink-service-monitor.yaml
Enter fullscreen mode Exit fullscreen mode

7. Verify the ServiceMonitor:

$ kubectl get servicemonitor flink-session-cluster \
    -n monitoring
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                    AGE
flink-session-cluster   10s
Enter fullscreen mode Exit fullscreen mode

8. Verify that its selector matches the metrics Service:

$ kubectl get servicemonitor flink-session-cluster \
    -n monitoring \
    -o jsonpath='{.spec.selector.matchLabels.monitoring}{"\n"}{.spec.endpoints[0].port}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

flink
metrics
Enter fullscreen mode Exit fullscreen mode

9. Wait for Prometheus to discover the new targets:

$ sleep 30
Enter fullscreen mode Exit fullscreen mode

10. Forward local port 9090 to the Prometheus Service:

$ kubectl port-forward \
    service/monitoring-kube-prometheus-prometheus \
    -n monitoring \
    9090:9090 > /tmp/prometheus-port-forward.log 2>&1 &
Enter fullscreen mode Exit fullscreen mode

11. Store the port-forward process ID:

$ PROMETHEUS_PORT_FORWARD_PID=$!
Enter fullscreen mode Exit fullscreen mode

12. Wait for the Prometheus API:

$ until curl --silent --fail \
    http://localhost:9090/-/ready > /dev/null; do
    sleep 2
done
Enter fullscreen mode Exit fullscreen mode

13. Query the Flink scrape targets:

$ curl --silent --get \
    http://localhost:9090/api/v1/query \
    --data-urlencode \
    'query=up{namespace="flink",service="flink-session-cluster-metrics"}'
Enter fullscreen mode Exit fullscreen mode

A value of "1" confirms that the target is reachable and healthy.

14. Verify that Prometheus stores Flink metrics:

$ curl --silent --get \
    http://localhost:9090/api/v1/query \
    --data-urlencode \
    'query=count({__name__=~"flink_.+"})'
Enter fullscreen mode Exit fullscreen mode

A value greater than zero confirms that Prometheus is collecting Flink metrics.

15. Stop port forwarding:

$ kill "$PROMETHEUS_PORT_FORWARD_PID"
Enter fullscreen mode Exit fullscreen mode

Access Grafana and Verify Prometheus

Grafana visualizes the Flink metrics stored in Prometheus. The monitoring stack includes a preconfigured Prometheus data source, so Grafana can query the metrics without requiring an additional connection.

1. Retrieve the Grafana administrator password:

$ GRAFANA_PASSWORD=$(kubectl get secret \
    monitoring-grafana \
    -n monitoring \
    -o jsonpath='{.data.admin-password}' | base64 --decode)
Enter fullscreen mode Exit fullscreen mode

2. Display the password:

$ echo "$GRAFANA_PASSWORD"
Enter fullscreen mode Exit fullscreen mode

Copy the displayed password. Use it to sign in to Grafana.

3. Forward local port 3000 to the Grafana service:

$ kubectl port-forward \
    service/monitoring-grafana \
    -n monitoring \
    3000:80 > /tmp/grafana-port-forward.log 2>&1 &
Enter fullscreen mode Exit fullscreen mode

4. Store the port-forward process ID:

$ GRAFANA_PORT_FORWARD_PID=$!
Enter fullscreen mode Exit fullscreen mode

5. Wait until Grafana becomes available:

$ until curl --silent --fail \
    http://localhost:3000/api/health > /dev/null; do
    sleep 2
done
Enter fullscreen mode Exit fullscreen mode

6. Verify the Grafana health status:

$ curl --silent \
    http://localhost:3000/api/health
Enter fullscreen mode Exit fullscreen mode

Output:

{
  "database": "ok",
  "version": "13.1.0",
  "commit": "b309c9bb3b81a748c3a75289236a27309ed2566a"
}
Enter fullscreen mode Exit fullscreen mode

7. Open http://localhost:3000 in a web browser. The Grafana login screen appears. Sign in using the following credentials:

  • Username: admin
  • Password: The value displayed by the echo "$GRAFANA_PASSWORD" command.

8. Verify that the Prometheus data source is available:

$ curl --silent \
    --user "admin:${GRAFANA_PASSWORD}" \
    http://localhost:3000/api/datasources/name/Prometheus
Enter fullscreen mode Exit fullscreen mode

9. Retrieve the actual Prometheus data source UID:

$ PROMETHEUS_UID=$(curl --silent \
    --user "admin:${GRAFANA_PASSWORD}" \
    http://localhost:3000/api/datasources/name/Prometheus |
    python3 -c 'import sys,json; print(json.load(sys.stdin)["uid"])')
Enter fullscreen mode Exit fullscreen mode

10. Query the Flink targets using the Prometheus data source UID:

$ curl --silent \
    --user "admin:${GRAFANA_PASSWORD}" \
    --get \
    "http://localhost:3000/api/datasources/proxy/uid/${PROMETHEUS_UID}/api/v1/query" \
    --data-urlencode \
    'query=up{namespace="flink",service="flink-session-cluster-metrics"}'
Enter fullscreen mode Exit fullscreen mode

Deploy a Flink Grafana Dashboard

Create a Grafana dashboard for Flink availability, failed checkpoints, job restarts, and backpressure. Store it in a labeled Kubernetes ConfigMap so Grafana loads it automatically.

1. Create the dashboard manifest:

$ nano flink-grafana-dashboard.yaml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration.

apiVersion: v1
kind: ConfigMap
metadata:
  name: flink-grafana-dashboard
  namespace: monitoring
  labels:
    grafana_dashboard: "1"
data:
  flink-monitoring.json: |
    {
      "title": "Apache Flink Monitoring",
      "uid": "flink-monitoring",
      "editable": true,
      "refresh": "30s",
      "schemaVersion": 39,
      "version": 1,
      "timezone": "browser",
      "time": { "from": "now-1h", "to": "now" },
      "panels": [
        {
          "id": 1,
          "title": "JobManager Availability",
          "type": "stat",
          "datasource": { "type": "prometheus", "uid": "prometheus" },
          "gridPos": { "h": 6, "w": 12, "x": 0, "y": 0 },
          "targets": [ { "expr": "flink_jobmanager_numRunningJobs", "legendFormat": "Running jobs" } ]
        },
        {
          "id": 2,
          "title": "Failed Checkpoints",
          "type": "timeseries",
          "datasource": { "type": "prometheus", "uid": "prometheus" },
          "gridPos": { "h": 6, "w": 12, "x": 12, "y": 0 },
          "targets": [ { "expr": "flink_jobmanager_job_numberOfFailedCheckpoints", "legendFormat": "{{job_name}}" } ]
        },
        {
          "id": 3,
          "title": "Job Restarts",
          "type": "timeseries",
          "datasource": { "type": "prometheus", "uid": "prometheus" },
          "gridPos": { "h": 6, "w": 12, "x": 0, "y": 6 },
          "targets": [ { "expr": "flink_jobmanager_job_numRestarts", "legendFormat": "{{job_name}}" } ]
        },
        {
          "id": 4,
          "title": "Max Task Backpressure",
          "type": "timeseries",
          "datasource": { "type": "prometheus", "uid": "prometheus" },
          "gridPos": { "h": 6, "w": 12, "x": 12, "y": 6 },
          "targets": [ { "expr": "max(flink_taskmanager_job_task_backPressuredTimeMsPerSecond)", "legendFormat": "Backpressure ms/s" } ]
        }
      ]
    }
Enter fullscreen mode Exit fullscreen mode

The dashboard:

  • Displays the availability of the Flink JobManager and TaskManager targets.
  • Displays the failed checkpoints and job restarts reported by the running Flink jobs.
  • Displays the highest backpressure value reported by a Flink task.
  • Refreshes the dashboard every 30 seconds.

Save and close the file.

2. Create the dashboard ConfigMap:

$ kubectl apply -f flink-grafana-dashboard.yaml
Enter fullscreen mode Exit fullscreen mode

3. Verify the ConfigMap:

$ kubectl get configmap flink-grafana-dashboard \
    -n monitoring
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                      DATA   AGE
flink-grafana-dashboard   1      8s
Enter fullscreen mode Exit fullscreen mode

4. Verify that the dashboard label exists:

$ kubectl get configmap flink-grafana-dashboard \
    -n monitoring \
    -o jsonpath='{.metadata.labels.grafana_dashboard}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

1
Enter fullscreen mode Exit fullscreen mode

5. Wait for the Grafana dashboard sidecar to load the dashboard:

$ sleep 30
Enter fullscreen mode Exit fullscreen mode

6. Verify that Grafana registered the dashboard:

$ curl --silent \
    --user "admin:${GRAFANA_PASSWORD}" \
    "http://localhost:3000/api/search?query=Apache%20Flink%20Monitoring"
Enter fullscreen mode Exit fullscreen mode

7. Open the following URL in a web browser:

http://localhost:3000/d/flink-monitoring/apache-flink-monitoring
Enter fullscreen mode Exit fullscreen mode

The Apache Flink Monitoring dashboard appears with panels for target availability, failed checkpoints, job restarts, and backpressure.

Configure Flink Alerts

Create Prometheus alerting rules for failed checkpoints, job restarts, and sustained backpressure. Prometheus evaluates these rules and sends active alerts to Alertmanager.

1. Create the alert rules manifest:

$ nano flink-alert-rules.yaml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration.

apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
  name: flink-alert-rules
  namespace: monitoring
  labels:
    release: monitoring
spec:
  groups:
    - name: flink.rules
      rules:
        - alert: FlinkCheckpointFailure
          expr: increase(flink_jobmanager_job_numberOfFailedCheckpoints[10m]) > 0
          for: 1m
          labels:
            severity: warning
          annotations:
            summary: "Flink checkpoint failure detected"
            description: "The Flink job {{ $labels.job_name }} recorded a failed checkpoint during the last 10 minutes."

        - alert: FlinkJobRestart
          expr: increase(flink_jobmanager_job_numRestarts[10m]) > 0
          for: 1m
          labels:
            severity: warning
          annotations:
            summary: "Flink job restart detected"
            description: "The Flink job {{ $labels.job_name }} restarted during the last 10 minutes."

        - alert: FlinkTaskBackpressure
          expr: max by (job_name, task_name) (flink_taskmanager_job_task_backPressuredTimeMsPerSecond) > 500
          for: 5m
          labels:
            severity: warning
          annotations:
            summary: "Flink task backpressure detected"
            description: "The task {{ $labels.task_name }} in job {{ $labels.job_name }} has remained backpressured for more than 500 milliseconds per second for five minutes."
Enter fullscreen mode Exit fullscreen mode

The alert rules:

  • Detect a failed checkpoint recorded during the last 10 minutes.
  • Detect a Flink job restart recorded during the last 10 minutes.
  • Detect task backpressure above 500 milliseconds per second for five minutes.
  • Use the release: monitoring label so Prometheus selects the resource.

The backPressuredTimeMsPerSecond metric reports how many milliseconds during each second a task spent under backpressure. A value above 500 means the task remained backpressured for more than half of the measured period.

Save and close the file.

2. Create the PrometheusRule:

$ kubectl apply -f flink-alert-rules.yaml
Enter fullscreen mode Exit fullscreen mode

3. Verify the alert rules resource:

$ kubectl get prometheusrule flink-alert-rules \
    -n monitoring
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                AGE
flink-alert-rules   4s
Enter fullscreen mode Exit fullscreen mode

4. Verify the rule group and alert names:

$ kubectl get prometheusrule flink-alert-rules \
    -n monitoring \
    -o jsonpath='{.spec.groups[0].name}{"\n"}{range .spec.groups[0].rules[*]}{.alert}{"\n"}{end}'
Enter fullscreen mode Exit fullscreen mode

Output:

flink.rules
FlinkCheckpointFailure
FlinkJobRestart
FlinkTaskBackpressure
Enter fullscreen mode Exit fullscreen mode

5. Forward local port 9090 to Prometheus:

$ kubectl port-forward \
    service/monitoring-kube-prometheus-prometheus \
    -n monitoring \
    9090:9090 > /tmp/prometheus-port-forward.log 2>&1 &
Enter fullscreen mode Exit fullscreen mode

6. Store the port-forward process ID:

$ PROMETHEUS_PORT_FORWARD_PID=$!
Enter fullscreen mode Exit fullscreen mode

7. Wait until Prometheus becomes available:

$ until curl --silent --fail \
    http://localhost:9090/-/ready > /dev/null; do
    sleep 2
done
Enter fullscreen mode Exit fullscreen mode

8. Verify that Prometheus loaded the Flink rules:

$ curl --silent \
    http://localhost:9090/api/v1/rules |
    python3 -c '
import json, sys
data = json.load(sys.stdin)
for group in data["data"]["groups"]:
    for rule in group["rules"]:
        name = rule.get("name", "")
        if name.startswith("Flink"):
            print(name)
'
Enter fullscreen mode Exit fullscreen mode

Output:

FlinkCheckpointFailure
FlinkJobRestart
FlinkTaskBackpressure
Enter fullscreen mode Exit fullscreen mode

9. Stop the Prometheus port-forward process:

$ kill "$PROMETHEUS_PORT_FORWARD_PID"
Enter fullscreen mode Exit fullscreen mode

10. Stop the Grafana port-forward process:

$ kill "$GRAFANA_PORT_FORWARD_PID"
Enter fullscreen mode Exit fullscreen mode

Configure High Availability

Enable the S3 Filesystem Plugin

Flink requires an S3 filesystem plugin to access Object Storage paths. The Flink image includes the S3 plugin in the /opt/flink/opt directory, but Flink only loads plugins from /opt/flink/plugins.

1. Verify that the Object Storage credentials secret exists:

$ kubectl get secret flink-object-storage \
    -n flink
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                   TYPE     DATA   AGE
flink-object-storage   Opaque   2      9d
Enter fullscreen mode Exit fullscreen mode

2. Verify that the Flink image includes the S3 filesystem plugin:

$ kubectl exec \
    -n flink \
    deploy/flink-session-cluster \
    -c flink-main-container -- \
    find /opt/flink/opt \
    -name 'flink-s3-fs-hadoop-*.jar'
Enter fullscreen mode Exit fullscreen mode

Output:

/opt/flink/opt/flink-s3-fs-hadoop-1.20.5.jar
Enter fullscreen mode Exit fullscreen mode

3. Configure an init container that copies the existing Flink plugins and the S3 filesystem plugin into a shared volume:

$ kubectl patch flinkdeployment flink-session-cluster \
    -n flink \
    --type merge \
    -p '{
      "spec": {
        "podTemplate": {
          "spec": {
            "initContainers": [
              {
                "name": "enable-s3-plugin",
                "image": "flink:1.20",
                "command": [
                  "sh",
                  "-c",
                  "cp -R /opt/flink/plugins/. /flink-plugins/ && mkdir -p /flink-plugins/s3-fs-hadoop && cp /opt/flink/opt/flink-s3-fs-hadoop-*.jar /flink-plugins/s3-fs-hadoop/"
                ],
                "volumeMounts": [
                  {
                    "name": "flink-plugins",
                    "mountPath": "/flink-plugins"
                  }
                ]
              }
            ],
            "containers": [
              {
                "name": "flink-main-container",
                "envFrom": [
                  {
                    "secretRef": {
                      "name": "flink-object-storage"
                    }
                  }
                ],
                "volumeMounts": [
                  {
                    "name": "flink-plugins",
                    "mountPath": "/opt/flink/plugins"
                  }
                ]
              }
            ],
            "volumes": [
              {
                "name": "flink-plugins",
                "emptyDir": {}
              }
            ]
          }
        }
      }
    }'
Enter fullscreen mode Exit fullscreen mode

The configuration:

  • Copies the existing Flink plugins into a shared volume.
  • Adds the S3 filesystem plugin to the active plugin directory.
  • Loads the Object Storage credentials from the existing Kubernetes Secret.
  • Applies the plugin configuration to the JobManager and TaskManager pods.

4. Wait for the Session Cluster to become stable:

$ kubectl wait flinkdeployment/flink-session-cluster \
    -n flink \
    --for=jsonpath='{.status.lifecycleState}'=STABLE \
    --timeout=600s
Enter fullscreen mode Exit fullscreen mode

5. Verify that the S3 filesystem plugin is available in the active plugin directory:

$ kubectl exec \
    -n flink \
    deploy/flink-session-cluster \
    -c flink-main-container -- \
    find /opt/flink/plugins/s3-fs-hadoop \
    -name 'flink-s3-fs-hadoop-*.jar'
Enter fullscreen mode Exit fullscreen mode

Output:

/opt/flink/plugins/s3-fs-hadoop/flink-s3-fs-hadoop-1.20.5.jar
Enter fullscreen mode Exit fullscreen mode

6. Verify that the Prometheus plugin remains available:

$ kubectl exec \
    -n flink \
    deploy/flink-session-cluster \
    -c flink-main-container -- \
    find /opt/flink/plugins/metrics-prometheus \
    -name 'flink-metrics-prometheus-*.jar'
Enter fullscreen mode Exit fullscreen mode

Output:

/opt/flink/plugins/metrics-prometheus/flink-metrics-prometheus-1.20.5.jar
Enter fullscreen mode Exit fullscreen mode

7. Verify that the Object Storage credentials are loaded without displaying their values:

$ kubectl exec \
    -n flink \
    deploy/flink-session-cluster \
    -c flink-main-container -- \
    sh -c '
      test -n "$AWS_ACCESS_KEY_ID" &&
      test -n "$AWS_SECRET_ACCESS_KEY" &&
      echo "Object Storage credentials loaded"
    '
Enter fullscreen mode Exit fullscreen mode

Output:

Object Storage credentials loaded
Enter fullscreen mode Exit fullscreen mode

Configure Kubernetes High Availability

Kubernetes high availability uses Kubernetes ConfigMaps for leader election. The configuration runs two JobManager replicas and stores recovery metadata in Object Storage so a standby JobManager can restore the cluster state after the active JobManager fails.

1. Configure high availability on the Session Cluster. Replace YOUR_BUCKET_NAME with your Object Storage bucket name and YOUR_OBJECT_STORAGE_HOSTNAME with your Object Storage hostname without the https:// prefix.

$ kubectl patch flinkdeployment flink-session-cluster \
    -n flink \
    --type merge \
    -p '{
      "spec": {
        "flinkConfiguration": {
          "high-availability.type": "kubernetes",
          "high-availability.storageDir": "s3://YOUR_BUCKET_NAME/ha",
          "s3.endpoint": "YOUR_OBJECT_STORAGE_HOSTNAME",
          "s3.path.style.access": "true"
        },
        "jobManager": {
          "replicas": 2
        }
      }
    }'
Enter fullscreen mode Exit fullscreen mode

The configuration:

  • Enables Kubernetes-based high availability.
  • Stores recovery metadata in the YOUR_BUCKET_NAME Object Storage bucket.
  • Connects Flink to the S3-compatible Object Storage endpoint.
  • Runs two JobManager replicas for leader and standby roles.

2. Wait for the Session Cluster to become stable:

$ kubectl wait flinkdeployment/flink-session-cluster \
    -n flink \
    --for=jsonpath='{.status.lifecycleState}'=STABLE \
    --timeout=600s
Enter fullscreen mode Exit fullscreen mode

3. Verify the high-availability configuration:

$ kubectl get flinkdeployment flink-session-cluster \
    -n flink \
    -o jsonpath='{.spec.flinkConfiguration.high-availability\.type}{"\n"}{.spec.flinkConfiguration.high-availability\.storageDir}{"\n"}{.spec.flinkConfiguration.s3\.endpoint}{"\n"}{.spec.flinkConfiguration.s3\.path\.style\.access}{"\n"}{.spec.jobManager.replicas}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

kubernetes
s3://YOUR_BUCKET_NAME/ha
YOUR_OBJECT_STORAGE_HOSTNAME
true
2
Enter fullscreen mode Exit fullscreen mode

4. Verify the JobManager deployment status:

$ kubectl get flinkdeployment flink-session-cluster \
    -n flink \
    -o jsonpath='{.status.jobManagerDeploymentStatus}{"\n"}{.status.error}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

READY
Enter fullscreen mode Exit fullscreen mode

5. Verify the JobManager pods:

$ kubectl get pods \
    -n flink \
    -l app=flink-session-cluster,component=jobmanager
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                     READY   STATUS    RESTARTS   AGE
flink-session-cluster-7b76c4b74d-9fk26   1/1     Running   0          45h
flink-session-cluster-7b76c4b74d-kb8jg   1/1     Running   0          45h
Enter fullscreen mode Exit fullscreen mode

6. Verify the ConfigMaps used for high-availability coordination:

$ kubectl get configmaps \
    -n flink \
    -l app=flink-session-cluster
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                                                DATA   AGE
flink-config-flink-session-cluster                                  2      45h
flink-session-cluster-6edae41fdf49cfa041659b86ee1ce10a-config-map   2      45h
flink-session-cluster-cluster-config-map                            5      45h
pod-template-flink-session-cluster                                  1      45h
Enter fullscreen mode Exit fullscreen mode

Verify JobManager Pod Recovery

Delete one JobManager pod to verify that Kubernetes restores the replica count and that the Flink REST service becomes available after the pod is replaced.

1. Store the name of one JobManager pod:

$ JOBMANAGER_POD=$(kubectl get pods \
    -n flink \
    -l app=flink-session-cluster,component=jobmanager \
    -o jsonpath='{.items[0].metadata.name}')
Enter fullscreen mode Exit fullscreen mode

2. Display the stored pod name:

$ echo "$JOBMANAGER_POD"
Enter fullscreen mode Exit fullscreen mode

Output:

flink-session-cluster-7b76c4b74d-5p5c8
Enter fullscreen mode Exit fullscreen mode

3. Delete the selected JobManager pod:

$ kubectl delete pod \
    -n flink \
    "$JOBMANAGER_POD"
Enter fullscreen mode Exit fullscreen mode

4. Wait for the JobManager Deployment to restore both replicas:

$ kubectl rollout status \
    deployment/flink-session-cluster \
    -n flink \
    --timeout=600s
Enter fullscreen mode Exit fullscreen mode

5. Verify the JobManager pods:

$ kubectl get pods \
    -n flink \
    -l app=flink-session-cluster,component=jobmanager
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                     READY   STATUS    RESTARTS   AGE
flink-session-cluster-7b76c4b74d-29v5s   1/1     Running   0          87s
flink-session-cluster-7b76c4b74d-kb8jg   1/1     Running   0          46h
Enter fullscreen mode Exit fullscreen mode

The newer pod replaces the deleted JobManager replica.

6. Verify that the FlinkDeployment remains stable:

$ kubectl get flinkdeployment flink-session-cluster \
    -n flink \
    -o jsonpath='{.status.lifecycleState}{"\n"}{.status.jobManagerDeploymentStatus}{"\n"}{.status.error}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

STABLE
READY
Enter fullscreen mode Exit fullscreen mode

7. Forward local port 8081 to the Flink REST service:

$ kubectl port-forward \
    service/flink-session-cluster-rest \
    -n flink \
    8081:8081
Enter fullscreen mode Exit fullscreen mode

Keep this terminal open. Open a second terminal and query the cluster overview.

$ curl --silent --fail \
    http://localhost:8081/overview
Enter fullscreen mode Exit fullscreen mode

A successful response confirms that the Session Cluster REST endpoint is available after Kubernetes replaces the deleted JobManager pod.

Configure Checkpoint Recovery

Store checkpoints and savepoints in Object Storage so Flink can recover job state after a pod failure.

1. Configure Object Storage as the checkpoint and savepoint location. Replace YOUR_BUCKET_NAME with your Object Storage bucket name.

$ kubectl patch flinkdeployment flink-session-cluster \
    -n flink \
    --type merge \
    -p '{
      "spec": {
        "flinkConfiguration": {
          "state.checkpoints.dir": null,
          "state.savepoints.dir": null,
          "execution.checkpointing.storage": "filesystem",
          "execution.checkpointing.dir": "s3://YOUR_BUCKET_NAME/checkpoints",
          "execution.checkpointing.savepoint-dir": "s3://YOUR_BUCKET_NAME/savepoints",
          "execution.checkpointing.externalized-checkpoint-retention": "RETAIN_ON_CANCELLATION",
          "execution.checkpointing.num-retained": "3"
        }
      }
    }'
Enter fullscreen mode Exit fullscreen mode

The configuration:

  • Removes the local checkpoint and savepoint paths.
  • Stores checkpoint data in Object Storage.
  • Stores savepoints in a separate Object Storage path.
  • Retains externalized checkpoints when a job is canceled.
  • Keeps the three most recent completed checkpoints.

2. Wait for the Session Cluster to become stable:

$ kubectl wait flinkdeployment/flink-session-cluster \
    -n flink \
    --for=jsonpath='{.status.lifecycleState}'=STABLE \
    --timeout=600s
Enter fullscreen mode Exit fullscreen mode

3. Verify the checkpoint recovery configuration:

$ kubectl get flinkdeployment flink-session-cluster \
    -n flink \
    -o jsonpath='{.spec.flinkConfiguration.execution\.checkpointing\.storage}{"\n"}{.spec.flinkConfiguration.execution\.checkpointing\.dir}{"\n"}{.spec.flinkConfiguration.execution\.checkpointing\.savepoint-dir}{"\n"}{.spec.flinkConfiguration.execution\.checkpointing\.externalized-checkpoint-retention}{"\n"}{.spec.flinkConfiguration.execution\.checkpointing\.num-retained}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

filesystem
s3://YOUR_BUCKET_NAME/checkpoints
s3://YOUR_BUCKET_NAME/savepoints
RETAIN_ON_CANCELLATION
3
Enter fullscreen mode Exit fullscreen mode

4. Verify that the local checkpoint and savepoint settings were removed:

$ kubectl get flinkdeployment flink-session-cluster \
    -n flink \
    -o jsonpath='{.spec.flinkConfiguration.state\.checkpoints\.dir}{"\n"}{.spec.flinkConfiguration.state\.savepoints\.dir}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

The command returns two empty lines.

5. Forward local port 8081 to the Flink REST service:

$ kubectl port-forward \
    service/flink-session-cluster-rest \
    -n flink \
    8081:8081
Enter fullscreen mode Exit fullscreen mode

Keep this terminal open. Open a second terminal and submit a long-running example job.

$ JOB_SUBMISSION=$(kubectl exec \
    -n flink \
    deploy/flink-session-cluster \
    -c flink-main-container -- \
    /opt/flink/bin/flink run \
    -d \
    /opt/flink/examples/streaming/StateMachineExample.jar)
Enter fullscreen mode Exit fullscreen mode

6. Display the submission result:

$ echo "$JOB_SUBMISSION"
Enter fullscreen mode Exit fullscreen mode

The output includes the submitted job ID.

7. Store the job ID:

$ JOB_ID=$(printf '%s\n' "$JOB_SUBMISSION" |
    grep -Eo '[0-9a-f]{32}' |
    tail -n 1)
Enter fullscreen mode Exit fullscreen mode

8. Verify the stored job ID:

$ echo "$JOB_ID"
Enter fullscreen mode Exit fullscreen mode

9. Wait for Flink to complete at least one checkpoint:

$ sleep 70
Enter fullscreen mode Exit fullscreen mode

10. Query the latest completed checkpoint:

$ curl --silent --fail \
    "http://localhost:8081/jobs/${JOB_ID}/checkpoints" |
    python3 -c '
import json, sys

data = json.load(sys.stdin)
latest = data.get("latest", {}).get("completed")

if not latest:
    raise SystemExit("No completed checkpoint found")

print("Checkpoint ID:", latest["id"])
print("Status:", latest["status"])
print("External path:", latest["external_path"])
'
Enter fullscreen mode Exit fullscreen mode

Output:

Checkpoint ID: 10
Status: COMPLETED
External path: s3://YOUR_BUCKET_NAME/checkpoints/d06ad801e7ff022a1811e2ab16533cdf/chk-10
Enter fullscreen mode Exit fullscreen mode

11. Cancel the example job after verification:

$ kubectl exec \
    -n flink \
    deploy/flink-session-cluster \
    -c flink-main-container -- \
    /opt/flink/bin/flink cancel "$JOB_ID"
Enter fullscreen mode Exit fullscreen mode

Return to the first terminal and press Ctrl+C to stop port forwarding.

Configure Security

Configure Namespace-Scoped RBAC

The Flink JobManager uses the flink service account to manage TaskManager pods and high-availability ConfigMaps. Create namespace-scoped permissions so the service account can manage these resources only in the flink namespace.

1. Verify that the Flink service account exists:

$ kubectl get serviceaccount flink \
    -n flink
Enter fullscreen mode Exit fullscreen mode

Output:

NAME    AGE
flink   13d
Enter fullscreen mode Exit fullscreen mode

2. Create an RBAC manifest:

$ nano flink-runtime-rbac.yaml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration.

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: flink-runtime
  namespace: flink
rules:
  - apiGroups:
      - ""
    resources:
      - pods
      - configmaps
    verbs:
      - get
      - list
      - watch
      - create
      - update
      - patch
      - delete
  - apiGroups:
      - apps
    resources:
      - deployments
      - deployments/finalizers
    verbs:
      - get
      - list
      - watch
      - create
      - update
      - patch
      - delete
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: flink-runtime
  namespace: flink
subjects:
  - kind: ServiceAccount
    name: flink
    namespace: flink
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: Role
  name: flink-runtime
Enter fullscreen mode Exit fullscreen mode

The configuration:

  • Grants access only in the flink namespace.
  • Allows the JobManager to manage TaskManager pods and high-availability ConfigMaps.
  • Allows Flink to manage the Kubernetes Deployment used by the Session Cluster.
  • Binds the permissions to the existing flink service account.

Save and close the file.

3. Create the Role and RoleBinding:

$ kubectl apply -f flink-runtime-rbac.yaml
Enter fullscreen mode Exit fullscreen mode

4. Verify that the RoleBinding references the flink-runtime Role:

$ kubectl get rolebinding flink-runtime \
    -n flink
Enter fullscreen mode Exit fullscreen mode

Output:

NAME            ROLE                 AGE
flink-runtime   Role/flink-runtime   19s
Enter fullscreen mode Exit fullscreen mode

5. Verify that the service account can create TaskManager pods in the flink namespace:

$ kubectl auth can-i create pods \
    -n flink \
    --as=system:serviceaccount:flink:flink
Enter fullscreen mode Exit fullscreen mode

Output:

yes
Enter fullscreen mode Exit fullscreen mode

6. Verify that the same permission does not apply in the default namespace:

$ kubectl auth can-i create pods \
    -n default \
    --as=system:serviceaccount:flink:flink
Enter fullscreen mode Exit fullscreen mode

Output:

no
Enter fullscreen mode Exit fullscreen mode

Protect Object Storage Credentials

Store Object Storage credentials in a Kubernetes Secret instead of adding them directly to the FlinkDeployment. The Session Cluster loads the Secret values as environment variables without displaying the credential values in the deployment configuration.

1. Verify that the Secret contains the required keys without displaying their values:

$ kubectl get secret flink-object-storage \
    -n flink \
    -o go-template='{{range $key, $value := .data}}{{printf "%s\n" $key}}{{end}}'
Enter fullscreen mode Exit fullscreen mode

Output:

AWS_ACCESS_KEY_ID
AWS_SECRET_ACCESS_KEY
Enter fullscreen mode Exit fullscreen mode

2. Verify that the Session Cluster references the Object Storage Secret:

$ kubectl get flinkdeployment flink-session-cluster \
    -n flink \
    -o jsonpath='{.spec.podTemplate.spec.containers[?(@.name=="flink-main-container")].envFrom[0].secretRef.name}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

flink-object-storage
Enter fullscreen mode Exit fullscreen mode

Restrict Flink Network Access

Apply a Kubernetes NetworkPolicy to control inbound and outbound traffic for the Session Cluster. The policy allows Flink component communication, Prometheus metrics collection, Kafka access, DNS queries, and outbound HTTPS connections.

1. Verify that the Kubernetes NetworkPolicy API is available:

$ kubectl api-resources \
    --api-group=networking.k8s.io |
    grep NetworkPolicy
Enter fullscreen mode Exit fullscreen mode

Output:

networkpolicies   netpol       networking.k8s.io/v1   true         NetworkPolicy
Enter fullscreen mode Exit fullscreen mode

Note: The NetworkPolicy API may be available even when the cluster network plugin does not enforce NetworkPolicy resources. Verify that your cluster uses a compatible network plugin before relying on the policy for traffic isolation.

2. Create a NetworkPolicy manifest:

$ nano flink-network-policy.yaml
Enter fullscreen mode Exit fullscreen mode

Add the following configuration.

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: flink-session-cluster
  namespace: flink
spec:
  podSelector:
    matchLabels:
      app: flink-session-cluster
  policyTypes:
    - Ingress
    - Egress
  ingress:
    - from:
        - podSelector: {}
    - from:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: monitoring
      ports:
        - protocol: TCP
          port: 9249
  egress:
    - to:
        - podSelector: {}
    - to:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: kafka
      ports:
        - protocol: TCP
          port: 9092
    - to:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: kube-system
          podSelector:
            matchLabels:
              k8s-app: kube-dns
      ports:
        - protocol: UDP
          port: 53
        - protocol: TCP
          port: 53
    - to:
        - ipBlock:
            cidr: 0.0.0.0/0
      ports:
        - protocol: TCP
          port: 443
Enter fullscreen mode Exit fullscreen mode

The policy:

  • Allows communication between Flink pods and other pods in the flink namespace.
  • Allows pods in the monitoring namespace to collect Prometheus metrics on port 9249.
  • Allows Flink jobs to connect to services in the kafka namespace on port 9092.
  • Allows DNS queries to the cluster DNS pods.
  • Allows outbound HTTPS connections on port 443, including connections to Object Storage.

Save and close the file.

3. Create the NetworkPolicy:

$ kubectl apply -f flink-network-policy.yaml
Enter fullscreen mode Exit fullscreen mode

4. Verify the NetworkPolicy:

$ kubectl get networkpolicy flink-session-cluster \
    -n flink
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                    POD-SELECTOR                AGE
flink-session-cluster   app=flink-session-cluster   9s
Enter fullscreen mode Exit fullscreen mode

5. Wait for the Session Cluster to remain stable:

$ kubectl wait flinkdeployment/flink-session-cluster \
    -n flink \
    --for=jsonpath='{.status.lifecycleState}'=STABLE \
    --timeout=600s
Enter fullscreen mode Exit fullscreen mode

6. Verify that the JobManager and TaskManager pods remain ready:

$ kubectl get pods \
    -n flink \
    -l app=flink-session-cluster
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                     READY   STATUS    RESTARTS   AGE
flink-session-cluster-64d45f7d57-sjkjf   1/1     Running   0          3d
flink-session-cluster-64d45f7d57-vx2s7   1/1     Running   0          3d
flink-session-cluster-taskmanager-1-1    1/1     Running   0          3d
flink-session-cluster-taskmanager-1-2    1/1     Running   0          3d
Enter fullscreen mode Exit fullscreen mode

Verify the Deployment

Verify the Flink Cluster Status

Verify the Session Cluster resources and confirm that the configured processing capacity is available.

1. Verify that the Session Cluster has reached the stable state:

$ kubectl get flinkdeployment flink-session-cluster \
    -n flink \
    -o jsonpath='{.status.lifecycleState}{"\n"}{.status.jobManagerDeploymentStatus}{"\n"}{.status.error}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

STABLE
READY
Enter fullscreen mode Exit fullscreen mode

The empty third line confirms that the FlinkDeployment does not report an error.

2. Verify the JobManager and TaskManager pods:

$ kubectl get pods \
    -n flink \
    -l app=flink-session-cluster
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                                     READY   STATUS    RESTARTS   AGE
flink-session-cluster-64d45f7d57-sjkjf   1/1     Running   0          3d
flink-session-cluster-64d45f7d57-vx2s7   1/1     Running   0          3d
flink-session-cluster-taskmanager-1-1    1/1     Running   0          3d
flink-session-cluster-taskmanager-1-2    1/1     Running   0          3d
Enter fullscreen mode Exit fullscreen mode

3. Verify the Flink services:

$ kubectl get services \
    -n flink \
    -l app=flink-session-cluster
Enter fullscreen mode Exit fullscreen mode

Output:

NAME                            TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)    AGE
flink-session-cluster-metrics   ClusterIP   None            <none>        9249/TCP   7d23h
flink-session-cluster-rest      ClusterIP   34.118.235.78   <none>        8081/TCP   3d
Enter fullscreen mode Exit fullscreen mode

4. Verify the configured TaskManager replicas and task slots:

$ kubectl get flinkdeployment flink-session-cluster \
    -n flink \
    -o jsonpath='{.spec.taskManager.replicas}{"\n"}{.spec.flinkConfiguration.taskmanager\.numberOfTaskSlots}{"\n"}'
Enter fullscreen mode Exit fullscreen mode

Output:

2
2
Enter fullscreen mode Exit fullscreen mode

The Session Cluster runs two TaskManager replicas with two task slots each, providing four task slots in total.

Produce Test Events

Create a temporary Kafka client pod and produce timestamped events to the flink-events topic.

1. Create a temporary Kafka client pod:

$ kubectl run kafka-client \
    -n kafka \
    --image=quay.io/strimzi/kafka:1.1.0-kafka-4.3.0 \
    --restart=Never \
    --command -- sleep infinity
Enter fullscreen mode Exit fullscreen mode

2. Wait for the pod to become ready:

$ kubectl wait pod/kafka-client \
    -n kafka \
    --for=condition=Ready \
    --timeout=120s
Enter fullscreen mode Exit fullscreen mode

3. Create a timestamp aligned with the current minute:

$ BASE_TIME=$(( $(date +%s) / 60 * 60 * 1000 ))
Enter fullscreen mode Exit fullscreen mode

4. Produce test events to the flink-events topic:

$ printf "test:checkout,10,$((BASE_TIME + 5000))\n\
test:checkout,15,$((BASE_TIME + 15000))\n\
test:payment,20,$((BASE_TIME + 20000))\n\
test:checkout,12,$((BASE_TIME + 30000))\n\
test:payment,25,$((BASE_TIME + 40000))\n\
test:checkout,30,$((BASE_TIME + 120000))\n" |
kubectl exec -i kafka-client \
    -n kafka -- \
    bin/kafka-console-producer.sh \
    --bootstrap-server flink-kafka-kafka-bootstrap:9092 \
    --topic flink-events \
    --property parse.key=true \
    --property key.separator=:
Enter fullscreen mode Exit fullscreen mode

The producer uses test as the Kafka record key. Each record value contains the event type, numeric value, and event-time timestamp. The final record advances event time so Flink can complete the earlier window.

Verify Window Results

1. Wait for the one-minute window and the next checkpoint to complete:

$ sleep 90
Enter fullscreen mode Exit fullscreen mode

2. Read committed results from the output topic:

$ kubectl exec -i kafka-client \
    -n kafka -- \
    bin/kafka-console-consumer.sh \
    --bootstrap-server flink-kafka-kafka-bootstrap:9092 \
    --topic flink-window-results \
    --from-beginning \
    --consumer-property isolation.level=read_committed \
    --max-messages 2 \
    --timeout-ms 30000
Enter fullscreen mode Exit fullscreen mode

Confirm that the consumer displays the aggregated checkout and payment events. The read_committed isolation level hides incomplete and aborted Kafka transactions.

Test Failure Recovery

Delete a TaskManager pod to simulate a failure, and verify that Kubernetes replaces the pod and Flink returns the streaming job to the RUNNING state.

1. Store the name of a running TaskManager pod:

$ TASKMANAGER_POD=$(kubectl get pods \
    -n flink \
    -l app=flink-session-cluster \
    -o name | grep taskmanager | head -n 1)
Enter fullscreen mode Exit fullscreen mode

2. Delete the TaskManager pod to simulate a failure:

$ kubectl delete "$TASKMANAGER_POD" -n flink
Enter fullscreen mode Exit fullscreen mode

3. Wait for two TaskManager pods to reach the Running state:

$ until [ "$(kubectl get pods \
    -n flink \
    -l app=flink-session-cluster \
    --field-selector=status.phase=Running \
    -o name | grep -c taskmanager)" -eq 2 ]; do
    sleep 5
done
Enter fullscreen mode Exit fullscreen mode

4. Verify that the replacement TaskManager pod is running:

$ kubectl get pods \
    -n flink \
    -l app=flink-session-cluster
Enter fullscreen mode Exit fullscreen mode

5. Forward local port 8081 to the Flink REST service:

$ kubectl port-forward \
    service/flink-session-cluster-rest \
    -n flink \
    8081:8081 > /tmp/flink-recovery-port-forward.log 2>&1 &
Enter fullscreen mode Exit fullscreen mode

6. Store the port-forward process ID:

$ RECOVERY_PORT_FORWARD_PID=$!
Enter fullscreen mode Exit fullscreen mode

7. Wait for the REST endpoint to become available:

$ until curl --silent --fail \
    http://localhost:8081/overview > /dev/null; do
    sleep 2
done
Enter fullscreen mode Exit fullscreen mode

8. Retrieve the running Flink job ID:

$ JOB_ID=$(curl --silent \
    http://localhost:8081/jobs/overview \
    | python3 -c 'import sys,json; jobs=json.load(sys.stdin)["jobs"]; print(next(job["jid"] for job in jobs if job["name"] == "Event-Time Windowing Job" and job["state"] == "RUNNING"))')
Enter fullscreen mode Exit fullscreen mode

9. Verify the job ID:

$ echo "$JOB_ID"
Enter fullscreen mode Exit fullscreen mode

10. Verify that the Flink job returned to the RUNNING state:

$ curl --silent --show-error \
    "http://localhost:8081/jobs/${JOB_ID}" \
    | grep -o '"state":"[^"]*"'
Enter fullscreen mode Exit fullscreen mode

11. Stop port forwarding:

$ kill "$RECOVERY_PORT_FORWARD_PID"
Enter fullscreen mode Exit fullscreen mode

12. Delete the temporary Kafka client pod:

$ kubectl delete pod kafka-client -n kafka
Enter fullscreen mode Exit fullscreen mode

Migrate from GCP Dataflow to Apache Flink

Migrating from GCP Dataflow to Apache Flink involves changing the pipeline runner, replacing Google Cloud-specific services, and adapting deployment, state, scaling, and monitoring workflows to Kubernetes.

Beam Pipeline Migration

Apache Beam pipelines can run on Flink when they use Beam-portable transforms and connectors.

  • Migration: Replace DataflowRunner with FlinkRunner in the pipeline options.
  • Connector changes: Replace Google Cloud-specific sources and sinks, such as Pub/Sub, BigQuery, and GCS connectors, with portable alternatives such as Kafka, JDBC-compatible databases, and Object Storage.
  • Validation: Run the pipeline locally or on a test Flink Session Cluster before migrating production workloads. Verify serialization, dependency packaging, parallelism, and event-time behavior.

Stateful Pipeline Migration

Dataflow state cells and timers must be reviewed before moving stateful processing to Flink.

  • Migration: Map per-key Dataflow state to Flink keyed state, such as ValueState, ListState, MapState, or ReducingState.
  • State access: Apply keyBy() before accessing keyed state and confirm that the selected key produces the same state partitioning used by the Dataflow pipeline.
  • Recovery: Configure durable checkpoints and savepoints before migrating stateful production jobs. Test application restarts and upgrades using representative state volumes.

Windowing and Trigger Migration

Apache Beam window definitions are portable, but runner-specific timing behavior may differ.

  • Migration: Retain compatible tumbling, sliding, and session window assignments when moving Beam pipelines to the Flink Runner.
  • Late data: Verify watermarks, allowed lateness, accumulation modes, and side outputs using delayed and out-of-order events.
  • Triggers: Test processing-time, event-time, and composite triggers because firing frequency and pane timing can differ between runners.

Template and Scheduling Migration

Dataflow Templates package reusable pipelines for repeatable execution.

  • Migration: Replace Dataflow Templates with versioned Flink application JARs or container images.
  • Deployment: Submit reusable jobs through FlinkSessionJob resources, the Flink REST API, or dedicated FlinkDeployment application clusters.
  • Scheduling: Replace Cloud Scheduler-based submissions with Kubernetes CronJobs or an external workflow orchestrator such as Apache Airflow.

Data Storage Considerations

Dataflow workloads commonly depend on Google Cloud storage and messaging services.

  • Google Cloud Storage: Migrate pipeline files, checkpoints, and intermediate data to S3-compatible Object Storage. Update storage paths, endpoints, and credentials in the Flink configuration.
  • Pub/Sub: Replace Pub/Sub sources and sinks with Kafka topics when the workload requires portable event streaming.
  • BigQuery: Replace BigQuery sinks with a compatible database, data warehouse, JDBC destination, or Object Storage format based on the workload requirements.
  • Credentials: Replace Google service-account credentials with Kubernetes Secrets, workload-specific service accounts, and storage access keys.

Things to Take Care During Migration

  • Google Cloud connectors: Beam connectors designed specifically for GCP may not work with the Flink Runner without replacement or additional configuration.
  • Autoscaling: Dataflow manages worker scaling automatically. Flink requires Reactive Mode, Kubernetes Horizontal Pod Autoscaling, or operator-managed scaling policies.
  • Streaming Engine: Dataflow Streaming Engine optimizations do not have a direct Flink equivalent. Re-evaluate memory, network, state backend, and checkpoint settings.
  • Access control: Replace Google Cloud IAM roles with Kubernetes RBAC, namespace permissions, service accounts, and external storage policies.
  • Monitoring: Replace Cloud Monitoring and Dataflow metrics with Prometheus, Grafana, Flink metrics, and Kubernetes alerts.
  • Large state: Tune checkpoint intervals, timeouts, incremental checkpoints, state backend storage, and restart strategies before migrating workloads with large state.
  • Parallelism: Dataflow worker counts do not map directly to Flink parallelism. Benchmark source partitions, task slots, operator parallelism, and backpressure before production migration.
  • Delivery guarantees: Verify that every source and sink supports the required delivery guarantee. End-to-end exactly-once processing requires checkpoint-compatible sources and transactional or idempotent sinks.

Next Steps

Apache Flink is now running on Kubernetes as a self-hosted alternative to GCP Dataflow, with Session and Application Clusters, exactly-once Kafka processing, Beam pipeline support, monitoring, high availability, and security controls in place. From here, you can:

  • Configure the Flink Autoscaler or a Kubernetes Horizontal Pod Autoscaler to adjust TaskManager capacity automatically based on workload.
  • Add TLS to the Kafka listeners and the Flink REST endpoint before exposing them outside the cluster.
  • Wire Alertmanager to a notification channel, such as Slack, PagerDuty, or email, so the FlinkCheckpointFailure and FlinkTaskBackpressure alerts reach your team.
  • Benchmark parallelism, task slots, and checkpoint intervals against your production workload before migrating additional GCP Dataflow jobs.

For the full guide with additional tips, visit the original article on Vultr Docs.

Top comments (0)