Apache Flink Architecture
Technical Analysis

1.1 — Runtime Stack: JobManager, TaskManager & Slot Allocation

The Flink runtime is a two-tier distributed system: a single (HA-enabled) JobManager acts as the cluster master, while one or more TaskManagers execute the actual stream processing workloads. A lightweight Client translates the user program into a serialized JobGraph and submits it — it is not part of the runtime execution path.

1.2 — Dataflow Graph Transformation Pipeline

Flink implements a four-stage compilation from user code to physical execution. Each stage applies progressively more physical reality — from logical semantics to actual parallelised tasks running on TaskManagers.

// Client-side — DataStream API → StreamGraph → JobGraph
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);

DataStream<Order> orders = env
    .addSource(new FlinkKafkaConsumer<>("orders", schema, props))  // StreamNode 1
    .map(Order::enrich)                                         // chained → same JobVertex
    .keyBy(Order::getCustomerId)                               // shuffle boundary → new JobVertex
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .reduce(OrderAgg::merge);                                   // StreamNode 3

// Calling execute() triggers: StreamGraph → JobGraph → submit to JobManager
env.execute("Order Aggregation Pipeline");

1.3 — State Backends: HashMapStateBackend vs. EmbeddedRocksDB

Flink separates two orthogonal concerns: in-flight state storage (the State Backend) and durable checkpoint storage (the Checkpoint Storage). The former determines runtime access latency; the latter determines fault-tolerance durability.

* Flink 1.18+ introduces generalised incremental CP via the Changelog State Backend (DSTL), enabling incremental snapshots for HashMapStateBackend.

1.4 — Chandy-Lamport Variant: Asynchronous Barrier Snapshotting (ABS)

Flink uses a variant of the classic Chandy-Lamport distributed snapshot algorithm, adapted for directed acyclic streaming topologies. The core innovation is the checkpoint barrier — a special control record injected into all source streams simultaneously by the CheckpointCoordinator.

# flink-conf.yaml — production checkpoint config
state.backend: rocksdb
state.backend.incremental: true
state.checkpoints.dir: s3://prod-bucket/flink-checkpoints
execution.checkpointing.interval: 60000          # 60s
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.timeout: 300000           # 5min max
execution.checkpointing.min-pause: 30000
state.backend.rocksdb.memory.managed: true
taskmanager.memory.managed.fraction: 0.4          # 40% for RocksDB

2.1 — Event Time vs. Processing Time

Flink distinguishes three notions of time. Only Event Time provides deterministic, reproducible results regardless of processing delays, network jitter, or replay scenarios.

2.2 — Watermark Mechanics

A watermark W(t) is a monotonically increasing timestamp assertion: "all events with event-time < t have been observed". Watermarks flow as special records through the operator DAG. Every window operator and time-based join uses the watermark to decide when to close a window and emit results.

// BoundedOutOfOrderness Watermark Strategy
WatermarkStrategy<Order> strategy = WatermarkStrategy
    .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(10)) // max expected lateness
    .withTimestampAssigner((order, ts) -> order.getEventTimestamp())
    .withIdleness(Duration.ofSeconds(30));            // unblock idle partitions

DataStream<Order> stream = env
    .fromSource(kafkaSource, strategy, "Kafka Orders")
    .keyBy(Order::getCustomerId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .allowedLateness(Time.minutes(2))       // re-fire window for late arrivals
    .sideOutputLateData(lateTag)              // route very-late to side output
    .reduce(OrderAgg::merge);

2.3 — Exactly-Once Semantics & Two-Phase Commit

Flink's internal checkpoint mechanism guarantees exactly-once state updates within the processing graph. To extend this guarantee to external sinks (Kafka, JDBC, filesystem), Flink uses the Two-Phase Commit (2PC) protocol, tightly coupled to checkpoint lifecyle events.

// End-to-end exactly-once Kafka sink
KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("output-enriched")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build())
    // Enables the 2PC protocol internally
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    // Must be unique per sink to survive restarts; ties tx IDs to checkpoint IDs
    .setTransactionalIdPrefix("fraud-pipeline-sink-v2")
    .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,
        String.valueOf(10 * 60 * 1000)) // < broker transaction.max.timeout.ms
    .build();

3.1 — Native Streaming vs. Micro-batching

The fundamental architectural divergence stems from each framework's origin story: Spark was built for batch processing and retrofitted with streaming; Flink was designed as a streaming-first engine from day one.

3.2 — Latency vs. Throughput Trade-offs

4.1 — Where Flink is the Clear Winner

Pattern<Transaction, ?> pattern =
  Pattern.<Transaction>begin("high")
    .where(t -> t.amount > 5000)
  .followedBy("foreign")
    .where(t -> !t.isHomeCountry())
  .within(Time.minutes(10));

4.2 — Kappa Architecture & Table API/SQL

The Kappa Architecture (Jay Kreps, 2014) discards the Lambda architecture's separate batch and speed layers. Instead, a single streaming engine handles both real-time and historical reprocessing — by replaying events from the durable log (Kafka).

Flink's Table API and SQL are the primary enablers of Kappa in production, providing a unified relational abstraction that executes identically on bounded (batch replay) and unbounded (live stream) data. A single SQL query can backfill historical results and continue processing live events without code changes.

-- Flink SQL: unified stream + batch query (Kappa pattern)
-- Works on live Kafka stream AND bounded Iceberg snapshot

CREATE TABLE orders (
  order_id    BIGINT,
  customer_id STRING,
  amount      DECIMAL(10,2),
  event_time  TIMESTAMP(3),
  WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
) WITH ('connector' = 'kafka', 'topic' = 'orders', ...);

SELECT
  TUMBLE_START(event_time, INTERVAL '5' MINUTE) AS window_start,
  customer_id,
  SUM(amount) AS revenue,
  COUNT(*) AS order_count
FROM orders
GROUP BY
  TUMBLE(event_time, INTERVAL '5' MINUTE),
  customer_id;
-- Identical query runs on Kafka (streaming) or Iceberg (batch backfill)

5.1 — Backpressure & Credit-Based Flow Control

Backpressure in Flink is a first-class, desirable signal — it indicates a downstream operator cannot consume data as fast as the upstream produces it. Rather than dropping data (like Storm) or relying on external throttling (like early Spark Streaming), Flink propagates backpressure naturally through its credit-based network layer.

Each network connection between TaskManagers uses Netty for transport. A receiver publishes credit announcements after recycling buffers. The sender only transmits data buffers equal to the announced credits, preventing a fast producer from overwhelming a slow consumer. This decouples Flink's application-level flow control from the underlying TCP window, so one slow task doesn't block unrelated tasks sharing the same TCP connection.

5.2 — Data Skew Strategies

Skew occurs when a keyBy() operation routes disproportionate event volumes to one or a few subtasks. One overloaded subtask becomes the global bottleneck, starving all upstream operators of buffer credits.

// Stage 1: partition with salt
stream.keyBy(e -> e.key + "_"
  + (int)(Math.random() * 32))
  .reduce(localAgg);
// Stage 2: merge partials
result.keyBy(e -> e.originalKey)
  .reduce(globalMerge);

-- Enable mini-batch in Table API
SET 'table.exec.mini-batch.enabled'
  = 'true';
SET 'table.exec.mini-batch.size'
  = '5000';
SET 'table.exec.mini-batch.allow-latency'
  = '200 ms';

// Async I/O for slow external lookups — prevents blocking the stream
DataStream<EnrichedOrder> enriched = AsyncDataStream.unorderedWait(
    orders,
    new AsyncCustomerLookup(redisClient),
    500, TimeUnit.MILLISECONDS,  // timeout per async request
    100                           // max concurrent in-flight requests
);

// RocksDB tuning for write-heavy workloads
EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(true);
backend.setRocksDBOptions(opts -> opts
    .setWriteBufferSize(64 * 1024 * 1024)       // 64 MB memtable
    .setMaxWriteBufferNumber(3)
    .setTargetFileSizeBase(64 * 1024 * 1024));   // SSTable size
env.setStateBackend(backend);

5.3 — Memory Configuration Quick Reference