Cluster & Environment Management¶

1- Provisioning and Scaling Clusters¶

1.0 Sizing Flink Resources¶

Context¶

Sizing a Flink cluster is a complex project task, influenced by many factors, including workload demands, application logic, data characteristics, expected state size, required throughput and latency, concurrency, and hardware characteristics.

Because of those variables, every Flink deployment needs a unique sizing approach. The most effective method is to run a real job, on real hardware and tune Flink to that specific workload.

For architects seeking sizing guidance, it's helpful to consider:

• the workload semantic complexity, with the usage of aggregations, joins, windows, and processing type,
• the input throughput (MB/s or records/second),
• the expected Flink state size (GB): assess number of unique key with average size of data to keep
• the expected latency.

While Kafka sizing estimates are based on throughput and latency, this is a very crude method for Flink, as it overlooks many critical details.

For new Flink deployments, a preliminary estimate can be provided, but it's important to stress its inexact nature.

A simple Flink job can process approximately 10,000 records per second per CPU. However, a more substantial job, based on benchmarks, might process closer to 5,000 records per second per CPU.

Flink Job and Task managers run in JVM, and the Task manager memory allocation looks like in the figure below. Flink memory is used by the Flink code, and is decomposed by 1/ network buffers to exchange partitions among operators, 2/ managed memory for state persistence buffering when using RockDB or ForSt, 3/ Task and Framework Off-heap, 4/ JVM Heap

• Total Process Memory: Memory allocated to the JVM process.

  spec:
    containers:
    - name: taskmanager
      image: flink:latest
      resources:
        requests:
          memory: "4Gi"
          cpu: "2"
        limits:
          memory: "4Gi"
          cpu: "2"
      env:
      - name: FLINK_PROPERTIES
        value: |
          taskmanager.numberOfTaskSlots: 2
          taskmanager.memory.process.size: 4096m
  

• For stateful streaming workloads, managed memory is usually the parameter that matters most. Shrink it and RocksDB gets less off-heap room for its block cache and other structures, so hot state falls back to disk more often and end-to-end latency increases. Grow it without proportionally sizing the JVM heap and the heap side for user functions can be squeezed, with GC events more likely to destabilize or fail the process.

• In context of Confluent Platform for Flink, or Apache Flink, the deployment entity is an Application, which means 1 job manager and 1 to many task managers. So Kubernetes worker nodes host multiple job managers and Task managers. For most of the case the Job Manager use one core and 2GB of RAM. But when the number of keys in state grows it may be needed to go to 2-4 cores and 4-8 GB.

Preconditions / Checklist¶

Assess the following characteristics of each Flink application:

• Desired Throughput and Latency
• Message Size, data skew, number of distinct keys
• Operator types, checkpoint interval, state size, parallelism
• Concurrent Jobs

Assess Hardware constraints:

• Number of worker nodes
• Memory and CPU per worker node
• Network bandwidth

CP Flink Estimation¶

I tentatively built a flink-estimator webapp with backend estimator for Apache Flink or Confluent Platform Cluster sizing.

The tool needs to be simple, it limits the number of parameters, and persists the estimation as json on the local disk to be able to save different estimations or update existing ones.

To access this web app there is a docker image at dockerhub - flink-estimator, or cloning the repository and run with python.

• use docker-compose up -d OR run with: ./run.sh
• Access via web browser http://localhost:8002/

  

1.1 Bring up new cluster/environment.¶

Context¶

Use this recipe when you need a new Flink environment (e.g., dev / stage / prod or a new region) backed by one of:

• Confluent Cloud environment
• A Kubernetes cluster (new or existing).
• CP Flink components installed via Helm (Flink K8s Operator + CMF).
• An existing Kafka cluster (Confluent Platform or Confluent Cloud).

Out of scope: deep infrastructure provisioning automation. The focus on the Flink layer once K8s and Kafka exist.

The Components to install for each deployment approach:

Preconditions / Checklist¶

Procedure¶

We list here the high level steps. For dedicated chapter on Kubernetes deployment

Gotchas¶

1.2 Adjust cluster resources (more TaskManagers, more slots).¶

Context¶

Use this when a Flink job or compute pool is under-provisioned: sustained backpressure, growing Kafka lag, or repeated "NoRecentCheckpoints"/resource exhaustion alerts or when on Confluent Cloud the statement becomes Degraded.

Confluent Platform for Flink or Apache Flink:

When you need to increase or decrease the cluster’s capacity by adding/removing TaskManagers, or by changing taskmanager.numberOfTaskSlots (slots per TM).

This recipe is primarily cluster/platform facing; job-level parallelism tuning is covered in a separate “Scale a Flink Job” recipe.

Always try to clarify the goal:

• Reduce backpressure / lag.
• Accommodate larger state (checkpoints/savepoints, RocksDB state) or Statement State becoming too big.
• Support more concurrent jobs/statements in a shared pool.

Preconditions / Checklist¶

• Have a good understanding of the performance situation with observability stack like Grafana dasboards and other metrics
• Get a clear assessment of the current number of Task managers and task slots per task manager
• Confirm current job parallelism and utilization
• Revisit current traffic pattern, versus the one used for platform sizing

Inputs / Parameters¶

• Current compute pool limits
• Current number of jobs per compute pool

Procedure¶

1. Diagnose whether you need more TMs, more Slots, or both. Check TM utilization and slot usage:
   • Are TaskManagers at or near 100% busy?
   • Are all slots occupied? Are there idle slots but backpressure at specific operators?
2. Check compute-pool / CFU limits (CP Flink / CCFlink): Ensure your target TM count/size stays within declared the limit; large TMs and more slots affect costs.

3. Adjust TaskManager count for horizontal scaling - (CP and OSS).

   • Adjust max CFU per compute pool, add more compute pool in Confluent Cloud. TM count is controlled indirectly via compute pool CFU and Autopilot/DSA; horizontal scaling is typically automatic.
   • For operator-only K8s Flink, TM count commonly set via deployment/Helm values for the TaskManager deployment (e.g., replicas in K8s or CFK spec).

4. Monitor - Check that:

   • Backpressure decreases.
   • Checkpoint durations stabilize or drop (CP or OSS).
   • TM CPU/memory are in a reasonable range (no massive under-utilization).

5. (Optional) Scale down later (CP or OSS): Once load drops, you can scale TMs down again to save resources. In CCFlink Autopilot will handle this automatically.

6. Adjust Slots Per TaskManager (Vertical Slot Scaling). Change taskmanager.numberOfTaskSlots to alter concurrency per TM. Get at least one core per busy slot for non-trivial jobs.

    "flinkConfiguration": {
    "taskmanager.numberOfTaskSlots": "2"
    }

Rollback¶

If the change makes things worse (more instability, timeouts, cost spikes):

• Revert TM Count / Slots
• If jobs fail or degrade: Temporarily move the most critical jobs to a separate, known-good pool / cluster with conservative settings. Then investigate per-job parallelism / skew, the RocksDB / disk hot spots, the autoscaler behavior.
• Document Effective Settings

2- Upgrading Flink Components¶

This section applies only to self managed Flink deployments. As Confluent Cloud is a managed service, the product version to version update is transparent to the users.

2.1 Upgrade Flink minor/patch versions.¶

Context¶

Preconditions / Checklist¶

Inputs / Parameters¶

Procedure¶

Rollback¶

Gotchas¶

2.2 Rolling upgrades and compatibility considerations.¶

Context¶

Preconditions / Checklist¶

Inputs / Parameters¶

Procedure¶

Rollback¶

Gotchas¶

3 Disaster Recovery & Multi-Region Strategies¶

Disaster recovery is about business continuity. Review the core principles of DR in this note.

DR for Flink depends on the deployment model (Confluent Cloud, Confluent Platform, or open-source) and always requires a DR strategy for Kafka and Schema Registry first. Data and schema replication (exact replication including offsets and schemas) are prerequisites; Flink state recovery builds on that.

• Cloud Providers are offering multiple regions for disaster recovery, and multiple availability zones within a region for high-availability. They do not communicate on how the physical allocation is done between physical data centers. DR may be acceptable to use different building in the same cities, but in general it is a region failover to mitigate earthquakes or cyberattack at the network level of a region.
• If availabilty zones is sufficient to support disaster management, it is the most cost effective solution as a lot of services, like Kafka, support synchronous replications between AZs. Always clarify those topologies.

• Failure to one AZ is easily recoverable.

Assessing resiliency needs¶

• Start by looking at business impact: assess cost of downtime? business metrics used to measure impact? and existing metrics and tools to compute outage impact. 70% of CIO wants to be resilient, 80% have nothing in place, and most has no real metrics.
• Assess what is the current method to detect downtime, and the definition of downtime. Moving to a standby region cost time and effort, getting a clear assessment of what downtime means is very important before triggering the failover. Is it latency? is it missing orders?
• Get an application inventory by category of criticality, including compliance requirements
• What is the process in place to alert human for downtime?
• What are the current observability practice?
• What is the recovery strategy?

Strategies for DR¶

• Backup/restore is for lower priority workload, cost less, and provision cloud resources after the event. RTO/RPO in hours
• Pilot Light: Data is live but services are idle. Provision some resource and scale after event. RPO/RTO on 10s of minutes
• Warm standby: The DR site is always running but smaller scale. Resources will scale after the event. RPO/RTO at the minutes level.
• Multi-site active/active: Zero downtime, near zero data loss. This is really for mission critical services.

DSP Elements to consider¶

When looking at the Flink architecture, it is important to recall that Flink relies on low-latency inter-node communication (TaskManagers exchanging shuffle data via RPCs) and tight state coordination, running one cluster across regions creates severe network bottlenecks and split-brain risks. Most of modern flink deployments are done as FlinkApplication on top of an orchestrator like kubernetes, the worker nodes host those applications, and application is job manager and n task manager. Flink Kubernetes Operator acts as a control plane to manage the application deployment life cycle.

Instead, multi-region resiliency requires deploying independent, fully self-contained Flink clusters in each region, paired with a cross-region streaming data pipeline. Also the source and sink connector selections impact the design of Flink DR solution. Using Kafka or Iceberg connectors have very different impacts.

The following figures illustrate what elements need to be considered for disaster recovery for the different platforms:

• Need to decide which elements can run and being created in the DR site.
• Active means, clients applications write data to the primary cluster. Passive, DR site, captures the replicated data and schemas. Active/active, clients write to either cluster.

• Map applications, flink pipelines with their expected RPO and RTO. When building analytics data products, most likely those metrics can be relaxed.

  Data product	Flink Pipeline DML	Consumers	RTO	RPO
  c360 analytics	dml.c360-fact-cust.sql	BI dashboard	4 hours	20 mn

• How networking hostname and CIDRs are defined in both sites. Here is a simple template:

• Flink Statements are most of the time stateful, and run continuously. Rebuild state cost RTO. A Flink statement stopped and restarted is the same as starting a new job. Assess if sources topics have extactly the same records.
• In case of active/passive, failover decision is triggered by human but automated by scripts and runbooks. Clients must bootstrap to the DR cluster once a failover is triggered.
• Disaster means source site is not reachable.
• Assess if consumers of records from Flink statements are idempotent and accept duplicates (the most recent consumer offsets may not have been committed to the destination cluster when a disaster occurs). Also consumers and producers must be tolerant of an RPO: there may be a small number of messages that have not replicated to the DR cluster at the time of failover.
• Not all kafka topics need to be replicated to DR sites. For Flink pipelines, it may be relevant to replicate only the source topics.
• Cluster Linking is set as bidirectional mode to enable topic data and metadata to sync between two clusters and should be used in all disaster recovery patterns. In the event of a disaster, a bidirectional Cluster Link can reverse the direction of data and metadata to support easy failover and/or fail back.

3.1 Active-passive pattern¶

Context¶

Use this when you need to survive a full region or data-center failure. Flink DR options assume Kafka and Schema Registry DR first: exact replication of data (including consumer offsets) and schemas.

First always assess RTO and RPO, and expected budget.

On Confluent Cloud, Cluster Linking and Schema Linking provide topic and schema replication.

On Confluent Platform or self-managed Flink, you replicate topics (and offsets) and schemas by your own means (e.g. MirrorMaker, cluster links, or shared storage).

Primary cluster means, Kafka cluster getting the write operations. Standby cluster is read-only.

Two main patterns:

• Active/active: Two (or more) regions run identical Flink jobs continuously, both processing the same input (with replication delay in the secondary). Best for low RTO, large state, or when you need exactly-once semantics and both regions to produce the same results.
• Active/passive: Flink runs only in the primary region; in the DR region, Flink jobs are started on failover. Best when state can be recreated quickly (e.g. stateless or small time windows) or when at-least-once/at-most-once is acceptable. It is relevant for use cases that fail forward, and that require failing back to the original region within weeks of the outage.

These patterns are described in Confluent Cloud Flink — Disaster Recovery; a generic DR components view is in figure below:

For Confluent Platform / Kubernetes with shared durable storage and Kafka replication:

Preconditions / Checklist¶

• Kafka (and Schema Registry) DR is in place: topics and consumer offsets replicated to the DR cluster; schemas replicated (e.g. Schema Linking on Confluent Cloud).
• RTO and RPO defined: how long failover can take and how much data loss is acceptable. Cluster Linking is asynchronous—RPO is bounded by mirror lag; RTO depends on client failover and (for active-passive) time to start Flink and restore state.
• For active-active: Flink jobs must be deterministic and tolerate out-of-order input; replicate only input topics and keep config (RBAC, networking) aligned across regions.
• For active-passive: Topic retention in DR must exceed the time needed to rebuild state (e.g. window size or TTL). Use event time (not processing time) for windows and aggregations so replay produces correct results.
• create the appropriate role bindings in the DR environment for applications and users that will failover

Inputs / Parameters¶

Procedure¶

Active/active (Confluent Cloud or multi-region)

1. Replicate only input topics to the DR region (e.g. cluster link from primary to DR). Do not replicate Flink output topics into the other region’s input to avoid feedback loops.
2. Replicate schema from Primary to secondary.
3. Deploy the same Flink statements/jobs in both regions, reading from the local Kafka cluster (primary or DR copy). Ensure queries are deterministic and support out-of-order arrival.
4. Mirror configuration: same service accounts, RBAC, private networking, and table/job definitions in both regions.
5. On regional failure: direct clients (producers/consumers) to the DR cluster (bootstrap and credentials via service discovery/vault); Flink in DR is already running.

Active/passive

1. Set up DR cluster with mirrored topics (and consumer offsets) and schemas. Ensure retention in DR > time to rebuild Flink state (window size or TTL).
2. Keep Flink job definitions and savepoints/checkpoint config in version control or automation. Do not run Flink in DR during normal operation.
3. On failover: switch Kafka consumers to the DR cluster; create/start Flink statements in DR (from savepoint if available, or from earliest offset so state is rebuilt from replicated input). Prefer event-time windows so replay is correct.
4. Confluent Cloud: see Cluster Linking DR and Failover (create DR cluster, cluster link, mirror topics, consumer offset sync, ACL sync). Provision API keys for the DR cluster in advance and store them in a vault for low RTO.

Monitoring

• Monitor mirror lag (Confluent Cloud Metrics API, Console, or REST) to bound RPO. Practice failover to meet RTO.

Rollback¶

• After primary region recovery: decide whether to fail back (switch clients and optionally Flink back to primary) or keep running in DR. If failing back, replicate any new data produced in DR back to primary (e.g. bidirectional cluster link) and then switch clients and Flink back to primary.

Gotchas¶

• All Flink DR depends on Kafka (and SR) DR; without replicated topics and offsets, Flink cannot recover correctly in the DR region.
• Active-passive with stateful jobs: RTO must be greater than the time to restore state from replicated input; otherwise use active-active or accept semantics/state trade-offs.
• Kafka concumers on passive side, must accept some RPO.
• Consumer offset sync (Confluent Cloud) is asynchronous; after failover, consumers may see a small number of duplicates—design for idempotency.
• Confluent Cloud is regional: there is no built-in cross-region state replication; you achieve DR by running Flink in the DR region and feeding it from replicated Kafka/Schema Registry.

Schema Management¶

• When disaster failover happens, Flink can keep reading schema-based data only if schemas have already been replicated to the DR region’s Schema Registry via Schema Linking.
• In the standard DR pattern, primary SR stays READWRITE, secondary SR stays IMPORT, and during failover you must reverse the schema link. See product documentation.
• On failover, ensure Schema Link is caught up, then promote/switch Kafka side.
• If you need Flink to run CREATE TABLE after failover, the DR side must allow schema writes to the active SR and the user/service account must have the required SR/Kafka permissions

3.2 Active-active pattern¶

Active/active really means that a user will loose connection to a site and be routed to another site in milliseconds. Session data as cookies may be recovered.

• Distributed reads is also a requirement for active - active
• The biggest challenge is to avoid data collision in the same data store: entity created in two regions in less than 100ms can collide when replicated.

3.3 Backup/restore of state backend¶

Context¶

Use this when you need to back up Flink job state for recovery, upgrades, or migration, or when designing for regional failure (disaster recovery). Flink state is captured via checkpoints (automatic, lightweight, Flink-managed) and savepoints (manual, portable, user-managed).

• Checkpoints: Provide automatic recovery from job failures. Flink creates, owns, and may delete them. Stored in state-backend–specific (native) format; optimized for fast restore. In production, store checkpoints in durable, shared storage (e.g. S3, GCS, HDFS, MinIO) so that JobManagers and TaskManagers can recover after process or node failure.
• Savepoints: User-triggered, consistent images of job state for planned operations (version upgrades, graph changes, rescaling). Stored in canonical (backend-independent) or native format; not deleted by Flink. Required when you need to restore a job in another region or after a major change.

By deployment:

• Confluent Cloud: Regional, multi-AZ service. Checkpoints run every minute for in-region fault tolerance. No user-configurable state backend; state is managed by the service. In a region loss, there is no built-in cross-region state restore—you must implement a cross-region DR strategy (see 3.2 below).
• Confluent Platform (CMF) / Kubernetes: Checkpoints and savepoints go to object or distributed storage (e.g. S3, HDFS). CMF supports Savepoint resources (trigger, list, detach, start from savepoint) via REST API. For failover between data centers, durable storage must be accessible from both (or replicated); Kafka topics (and offsets) must be replicated; then Flink applications can be restarted from checkpoints/savepoints in the DR site.
• Open-source Flink (e.g. K8s operator): Same concepts; configure state.checkpoints.dir and optional state.savepoints.dir to durable storage. Use Kubernetes HA with high-availability.storageDir for JobManager metadata.

Preconditions / Checklist¶

• Durable, shared storage for checkpoints (and savepoints) that all Flink processes can access (e.g. S3 bucket, HDFS, GCS, MinIO). For cross-DC DR, storage must be reachable from the DR site or replicated.
• Checkpointing enabled in the job (and optionally configured retention).
• For savepoints: job is running (or you have an existing savepoint path to adopt/import).
• For Confluent Cloud: no direct state-backend configuration; ensure Kafka + Schema Registry DR (e.g. Cluster Linking, Schema Linking) before designing Flink DR.

Inputs / Parameters¶

Procedure¶

Backup (savepoint) — Confluent Platform / CMF

1. Trigger a savepoint via CMF API (applications or statements):
   • POST /cmf/api/v1/environments/{env}/applications/{appName}/savepoints or .../statements/{stmtName}/savepoints
   • Optional body: spec.path, spec.formatType (e.g. CANONICAL), spec.backoffLimit.
2. Poll or watch savepoint status until status.state is COMPLETED; use status.path as the restore path.
3. Optionally detach the savepoint (applications only) to preserve it after deleting the application: POST .../savepoints/{savepointName}/detach. For external/open-source savepoints, register as detached: POST /cmf/api/v1/detached-savepoints with spec.path.

Backup (checkpoint) — self-managed (CP/OSS)

1. Ensure state.checkpoints.dir points to durable storage (e.g. enable checkpointing to S3).
2. Let Flink create checkpoints at the configured interval. Optionally set execution.checkpointing.externalized-checkpoint-retention to retain checkpoints on cancel.
3. For a consistent “backup” suitable for migration or DR, trigger a savepoint (as above) rather than relying only on the last checkpoint.

Restore

1. From savepoint: Start the application with startFromSavepoint set to the savepoint path, CMF savepoint name, or UID. In CMF: create/update application with savepointName/uid/initialSavepointPath, or POST .../applications/{appName}/start?startFromSavepointUid={uid}.
2. From checkpoint (automatic): After a failure, Flink (and K8s HA if configured) restarts the job and restores from the latest successful checkpoint when no savepoint is specified.
3. Confluent Cloud: Failed statements auto-restart using the last checkpoint; there is no user-visible backup/restore of state to another region—use active-active or active-passive (see 3.2).

Rollback¶

• If a restore from savepoint fails or the job misbehaves: stop the job, fix code or state compatibility, then start again from an earlier savepoint or from scratch (no savepoint).
• If you restored with wrong parallelism or backend: take a new savepoint, adjust config, and restore from that savepoint (canonical savepoints support state backend change and rescaling; see Flink checkpoints vs savepoints).

Gotchas¶

• Checkpoints are often incremental and backend-specific; savepoints in canonical format are portable and support state backend change and rescaling. Native savepoints are faster but have limitations (e.g. no state backend change).
• You cannot create a canonical savepoint for a job with an operator using enableAsyncState() (CMF/Flink).
• Deleting an attached savepoint in CMF requires the Flink cluster to be running; use force=true only when necessary (can leave orphaned data in blob storage).
• Confluent Cloud does not expose state backend or checkpoint/savepoint paths; DR is achieved by running Flink in the DR region against replicated Kafka/Schema Registry and accepting state rebuild or active-active/active-passive design.