Phase 6: Distributed Execution, High Availability & Resource Management¶
Audience: Principal Cloud Infrastructure Architect & Site Reliability Engineer
Source References:flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/,flink-kubernetes/
1. Cluster Deployment Modes: Session, Per-Job & Application¶
┌────────────────────────────────────────────────────────────────────────┐
│ 1. Session Mode (Multi-Tenant Shared Cluster) │
├────────────────────────────────────────────────────────────────────────┤
│ - Long-running JobManager and TaskManager pool │
│ - Multiple jobs submitted to the same shared TaskManagers │
│ - Pros: Instant job startup (no container provisioning delay) │
│ - Cons: Noisy neighbors! One failing job can crash the entire JVM! │
└────────────────────────────────────────────────────────────────────────┘
┌────────────────────────────────────────────────────────────────────────┐
│ 2. Application Mode (Cloud-Native Gold Standard) │
├────────────────────────────────────────────────────────────────────────┤
│ - One dedicated cluster per job (JobManager + TaskManagers) │
│ - The user program's main() executes DIRECTLY inside the JobManager! │
│ - Pros: Complete resource & security isolation; eliminates client-side │
│ bandwidth bottleneck when submitting multi-gigabyte fat JARs! │
└────────────────────────────────────────────────────────────────────────┘
2. High Availability (HA) & ZooKeeper/Kubernetes Consensus¶
In production, the JobManager cannot be a single point of failure (SPOF).
JobManager HA Leader Election:
[ Standby JobManager 2 ] ──┐
│ Watch Ephemeral Znode
[ Standby JobManager 1 ] ──┐ │
│ ▼
├──► [ ZooKeeper Ensemble / K8s ConfigMap ]
│ ▲
[ Active Leader JobManager ] ─────────────┘ Holds Leader Lock!
│
▼ Persists Checkpoint Metadata & ExecutionGraph state
[ Durable Storage: s3://flink-checkpoints/ha/ ]
2.1 The Failover Handshake¶
- Active JobManager process crashes (e.g., hardware fault).
- Its ephemeral lock in ZooKeeper (
/leader/jobmanager_lock) or Kubernetes lease expires. - Standby JobManager 1 detects the deletion via ZooKeeper Watch and claims leadership.
- Standby reads the latest completed checkpoint metadata pointer from
high-availability.storageDir. - Connects to existing TaskManagers (or spins up new ones), redeploys the ExecutionGraph, and resumes stream processing from the last confirmed checkpoint. Downtime: typically \(< 10\text{ seconds}\).
3. Slot Sharing Groups & Task Colocation¶
Why does Flink not assign 1 operator per slot? If it did, a pipeline with Source(4) -> Window(4) -> Sink(4) would require \(4 + 4 + 4 = 12\) slots. Moreover, the Source would idle while the Window computed, wasting 66% of CPU!
Instead, Flink defaults to Slot Sharing Groups:
Slot Sharing Architecture (Parallelism = 2):
TaskManager Slot 1:
┌────────────────────────────────────────────────────────┐
│ Task Slot 1 (JVM Thread Pool) │
│ [Source Subtask 1] ──► [Window Subtask 1] ──► [Sink 1] │
└────────────────────────────────────────────────────────┘
TaskManager Slot 2:
┌────────────────────────────────────────────────────────┐
│ Task Slot 2 (JVM Thread Pool) │
│ [Source Subtask 2] ──► [Window Subtask 2] ──► [Sink 2] │
└────────────────────────────────────────────────────────┘
3.1 The Golden Sizing Theorem¶
Because different operators in a pipeline share the same slot: $\(\text{Total Required Slots} = \max(\text{Parallelism of any single operator in the job})\)$ - Even a complex pipeline with 50 distinct operators and max parallelism of 16 requires exactly 16 slots! - Balances CPU utilization across TaskManagers automatically.
4. Cloud-Native Reactive Autoscaling¶
In Kubernetes deployments, Flink supports Reactive Mode:
- The Flink job dynamically adjusts its parallelism to match the exact number of TaskManager pods currently alive. - If a Kubernetes Horizontal Pod Autoscaler (HPA) scales TaskManager pods from 4 to 8 based on CPU metrics, Flink automatically triggers a savepoint, rescales operator parallelism to 8, and resumes processing without human intervention!