Apache Flink Interview Questions — Top 30 with Answers

Fundamentals

Q1: What is Apache Flink and what makes it different from Spark Streaming?

Flink is a distributed stream processing framework with true native streaming — records are processed one at a time as they arrive, not in micro-batches. Key differences from Spark Streaming:

| | Flink | Spark Streaming | |---|---|---| | Processing model | Native streaming (record-at-a-time) | Micro-batch (DStream) or Structured Streaming (micro-batch) | | Latency | Sub-second (milliseconds) | Hundreds of milliseconds minimum | | State management | Rich, natively integrated | External store or RDD state | | Event time | First-class with watermarks | Added later, less ergonomic | | Exactly-once | Yes, with checkpointing | Yes, with idempotent sinks | | Batch support | Yes (DataSet API, unified in Flink 1.12+) | Primary use case |

Flink is preferred when sub-second latency or complex stateful operations (e.g., session windows, pattern detection) are required.

Q2: What is the difference between event time, ingestion time, and processing time?

| Time domain | Definition | When to use | |-------------|------------|-------------| | Event time | When the event actually occurred (embedded in the record payload) | Default for most production pipelines — handles late arrivals correctly | | Ingestion time | When the record arrived at the Flink source | Simpler than event time, no watermarks needed, but late arrivals not handled | | Processing time | Wall clock time when the operator processes the record | Lowest latency; non-deterministic — same job can produce different results on re-run |

Use event time for any pipeline where results must be reproducible and late events are possible (always the case with Kafka).

Q3: What is a watermark in Flink?

A watermark is a special marker injected into the stream that asserts: "I am confident that no event with event time < W will arrive after this point." Watermarks allow Flink to close time windows and emit results without waiting forever.

WatermarkStrategy
  .<MyEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
  .withTimestampAssigner((event, ts) -> event.getTimestamp())

A watermark of W means: events with event_time < W will no longer arrive (with the configured tolerance). When the watermark advances past a window's end time, that window is closed and its aggregate is emitted.

Key gotcha: Watermarks propagate as the minimum across all parallel subtasks. One idle partition with no events blocks watermark progress for all windows — use withIdleness(Duration.ofMinutes(1)) to handle idle sources.

Q4: What happens to late events in Flink?

Events that arrive after their window has already been closed (because the watermark passed the window end) are called late events. Options:

1. Drop silently (default) — late events are discarded
2. Side output — route late events to a separate DataStream for later analysis or reprocessing
3. Allowed lateness — keep the window state alive for an additional duration after the watermark passes; late events trigger window re-computation

stream
  .keyBy(...)
  .window(TumblingEventTimeWindows.of(Time.minutes(1)))
  .allowedLateness(Time.seconds(30))
  .sideOutputLateData(lateOutputTag)
  .aggregate(new MyAggregator())

State Management

Q5: What types of state does Flink support?

| State type | Description | |------------|-------------| | ValueState<T> | Single value per key | | ListState<T> | Ordered list of values per key | | MapState<K, V> | Key-value map per stream key | | ReducingState<T> | Automatically reduced single value | | AggregatingState<IN, OUT> | Aggregation with different input/output types |

All of these are keyed state — scoped to the current key. Operator state (non-keyed, e.g., ListState for Kafka offsets) is scoped to the operator instance.

Q6: What is a state backend and which should you choose?

The state backend determines where state is stored:

| Backend | Storage | When to use | |---------|---------|-------------| | HashMapStateBackend (default) | JVM heap | Small state, fast access, limited by heap | | EmbeddedRocksDBStateBackend | Off-heap RocksDB on local disk | Large state (GBs–TBs), incremental checkpoints, slightly higher latency |

For production with state > a few GB, always use RocksDB. Enable incremental checkpoints to reduce checkpoint write time and storage costs.

env.setStateBackend(new EmbeddedRocksDBStateBackend(true)); // true = incremental

Q7: What is a checkpoint in Flink and how does it work?

A checkpoint is a consistent snapshot of all operator state at a point in time. Flink uses the Chandy-Lamport algorithm: the JobManager injects checkpoint barriers into all source partitions. When a barrier flows through an operator, that operator snapshots its state and forwards the barrier downstream. When all sinks acknowledge receipt of all barriers, the checkpoint is complete.

Key configs:

env.enableCheckpointing(60_000); // checkpoint every 60s
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000);
env.getCheckpointConfig().setCheckpointTimeout(120_000);

On failure, Flink restores all operator state from the latest successful checkpoint and re-reads source data from the checkpointed offset.

Q8: What is a savepoint and how is it different from a checkpoint?

| | Checkpoint | Savepoint | |---|---|---| | Purpose | Automatic failure recovery | Manual upgrade, migration, A/B test | | Triggered by | Flink automatically (periodic) | User explicitly (flink savepoint <jobId>) | | Format | Backend-specific, optimized | Canonical, portable format | | Expiry | Discarded on job restart by default | Persists until explicitly deleted | | Incremental | Yes (RocksDB) | No (always full) |

Use savepoints when deploying a new job version — stop the old job with a savepoint, start the new version from that savepoint.

Windows

Q9: What window types does Flink support?

| Type | Definition | Example use case | |------|------------|-----------------| | Tumbling window | Fixed-size, non-overlapping | 1-minute aggregations | | Sliding window | Fixed-size, overlapping by stride | 5-min window every 1 min (rolling average) | | Session window | Gaps-based; window closes after inactivity | User session grouping (close after 30s idle) | | Global window | Single window for all elements; needs custom trigger | Custom eviction logic |

Q10: What is a trigger in Flink windows?

A trigger defines when a window should fire (produce output). Default triggers fire when the watermark passes the window end. Custom triggers enable:

• Count trigger: Fire every N elements
• Event-time trigger: Fire at a specific timestamp
• Processing-time trigger: Fire on wall clock schedule
• Composite triggers: Fire on multiple conditions (e.g., count OR time, whichever comes first)

.trigger(CountTrigger.of(100))  // fire every 100 elements

Exactly-Once and Transactions

Q11: How does Flink achieve exactly-once with Kafka end-to-end?

Flink uses the two-phase commit protocol with the Kafka transactional producer:

1. Pre-commit: At each checkpoint, the Kafka sink flushes buffered records to Kafka but does not commit them (opens a Kafka transaction)
2. Commit: Once all operators acknowledge the checkpoint, the JobManager notifies sinks to commit their Kafka transactions
3. Recovery: If the job fails before commit, the open Kafka transaction is aborted on restart

Consumer downstream must use isolation.level=read_committed to only see committed records.

Interview answer: "Flink's Kafka sink uses Kafka transactions aligned to checkpoint boundaries. Records are pre-written but not visible until the checkpoint succeeds. If the job crashes, the transaction is aborted and the source re-reads from the last checkpoint offset."

Q12: What is the difference between at-least-once and exactly-once checkpointing modes?

With EXACTLY_ONCE checkpointing: Flink uses checkpoint barriers alignment — an operator buffers records from channels that have already sent the barrier, waiting until all channels deliver the barrier before snapshotting. This guarantees no record is double-counted.

With AT_LEAST_ONCE: Flink doesn't align barriers — operators snapshot immediately when the first barrier arrives and continue processing. Faster, but records between the first and last barrier may be processed twice after recovery.

Use EXACTLY_ONCE for financial or deduplication-sensitive pipelines. Use AT_LEAST_ONCE when the downstream sink is idempotent and throughput is more important.

Kafka Integration

Q13: How do you consume from Kafka in Flink 1.17+?

Use the KafkaSource connector (DataStream API):

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("broker:9092")
    .setTopics("orders")
    .setGroupId("flink-consumer-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

DataStream<String> stream = env.fromSource(
    source,
    WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5)),
    "Kafka Source"
);

The new KafkaSource supports both batch and streaming modes and integrates with the unified Source API (FLIP-27).

Q14: How do you write to Kafka from Flink with exactly-once?

KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("broker:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("output-topic")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build())
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .setTransactionalIdPrefix("flink-tx-")
    .build();

stream.sinkTo(sink);

Requires Kafka transaction.timeout.ms to be longer than execution.checkpointing.interval × 2. Set transaction.timeout.ms=900000 (15 minutes) to match a 5-minute checkpoint interval with headroom.

Sessionization

Q15: How do you build a sessionization pipeline in Flink?

Raw events → keyBy(userId) → session window (30s gap) → session aggregator → output

stream
  .keyBy(event -> event.getUserId())
  .window(EventTimeSessionWindows.withGap(Time.seconds(30)))
  .aggregate(new SessionAggregator(), new SessionWindowFunction())
  .addSink(sink);

The session window automatically merges windows when new events arrive within the gap. Each user's session is independent.

Interview considerations:

• Late events can reopen a closed session — use allowedLateness() to handle this
• Very long sessions with many events can cause large state — consider a maximum session size limit
• Sessionization state size grows with unique keys × average session length

Backpressure and Performance

Q16: What causes backpressure in Flink and how do you diagnose it?

Backpressure occurs when a downstream operator can't consume records as fast as upstream operators produce them. Flink uses a credit-based flow control mechanism — if a downstream buffer is full, upstream stops sending.

Diagnosis in Flink UI:

• Red/orange flame icon on a task indicates backpressure
• Navigate to the job graph and check backPressured metric
• outPoolUsage near 100% on a task means its output buffer is full

Common causes:

• Slow external I/O in a sink (e.g., slow Elasticsearch writes)
• CPU-intensive UDF with no async I/O
• GC pressure from large heap state
• Skewed key distribution causing one subtask to do more work

Fixes: Scale out the slow operator independently (via setParallelism()), use async I/O for external calls, switch to RocksDB to reduce GC pressure.

Q17: What is AsyncIO in Flink and when should you use it?

AsyncIO lets you make non-blocking external calls (e.g., database lookups, REST APIs) from a Flink operator without blocking the processing thread.

AsyncDataStream.unorderedWait(
    stream,
    new AsyncEnrichmentFunction(),  // implements AsyncFunction
    5, TimeUnit.SECONDS,            // timeout
    100                              // concurrent requests in-flight
);

Use unorderedWait when output order doesn't need to match input order (higher throughput). Use orderedWait when order must be preserved (slightly lower throughput due to buffering).

Without AsyncIO: One blocking call per record serializes throughput. With AsyncIO and 100 concurrent requests, you get 100x throughput improvement for I/O-bound enrichment.

Q18: How do you handle data skew in Flink?

Skew occurs when one key generates far more records than others (e.g., a hot product_id), causing one subtask to lag while others are idle.

Approaches:

1. Pre-aggregate with salting: Add a random prefix to the key, aggregate in two stages — first by key + salt, then by key.
2. Increase parallelism for the skewed operator: Use setParallelism() on the specific operator rather than globally.
3. Custom partitioner: Spread a known hot key across multiple subtasks using a range or modulo scheme.
4. KeyGroup rebalancing: For skewed session windows, use a sliding pre-aggregation to flatten the load.

SQL and Table API

Q19: What is the Flink Table API and how does it relate to the DataStream API?

The Table API and Flink SQL provide a relational abstraction over streams and batches. They compile down to DataStream programs at runtime.

TableEnvironment tEnv = StreamTableEnvironment.create(env);
tEnv.executeSql("""
    CREATE TABLE orders (
        order_id STRING,
        amount DOUBLE,
        order_time TIMESTAMP(3),
        WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'orders',
        'format' = 'json'
    )
""");

Table result = tEnv.sqlQuery("""
    SELECT TUMBLE_START(order_time, INTERVAL '1' MINUTE), SUM(amount)
    FROM orders
    GROUP BY TUMBLE(order_time, INTERVAL '1' MINUTE)
""");

You can convert between Table and DataStream with tEnv.toDataStream(table) and tEnv.fromDataStream(stream).

Q20: What is a dynamic table in Flink SQL?

A dynamic table is Flink's abstraction of a continuously changing table backed by a stream. As new events arrive, the table is updated. Two query modes:

• Append-only: New rows are only inserted (no updates or deletes) — maps to an unbounded INSERT stream
• Changelog stream: Rows can be inserted, updated, or deleted — maps to a +I / -U / +U / -D changelog stream (needed for aggregations with retractions)

Dynamic tables allow standard SQL syntax (SELECT, GROUP BY, JOIN) to work on unbounded streams.

Production Operations

Q21: How do you upgrade a Flink job without losing state?

1. Take a savepoint: flink savepoint <jobId> hdfs://savepoints/
2. Stop the running job: flink cancel <jobId> (or stop with savepoint: flink stop <jobId>)
3. Update the job JAR or topology
4. Restart from savepoint: flink run -s hdfs://savepoints/<path> -c com.example.MyJob my-job.jar

Constraints:

• Operator UIDs must be stable across versions — always set uid("unique-name") on stateful operators
• State schema changes require custom StateDescriptor migration or a manual state migration step
• Adding/removing operators may require state compatibility planning

Q22: What is the difference between stop and cancel in Flink?

| Command | Behavior | |---------|----------| | flink cancel <jobId> | Immediately terminates the job — no savepoint, state is lost (but last checkpoint survives) | | flink stop <jobId> | Injects a MAX_WATERMARK, waits for all windows to close, takes a savepoint, then stops gracefully |

Always use flink stop for planned maintenance or upgrades. Use flink cancel only for jobs in an unrecoverable bad state.

Q23: How do you tune checkpoint performance in a high-throughput Flink job?

Key levers:

1. Use incremental checkpoints (RocksDB): Only changed SST files are written per checkpoint vs. full state.
2. Set minPauseBetweenCheckpoints: Prevents checkpoints from back-to-back — gives the pipeline breathing room.
3. Use aligned vs. unaligned checkpoints: Unaligned checkpoints (enableUnalignedCheckpoints(true)) reduce checkpoint time under backpressure by snapshotting in-flight records, at the cost of checkpoint size.
4. Tune RocksDB: Set state.backend.rocksdb.predefined-options=SPINNING_DISK_OPTIMIZED or FLASH_SSD_OPTIMIZED based on storage.
5. Checkpoint storage: Use S3 with fs.s3a.fast.upload=true to pipeline uploads.

Q24: What metrics do you monitor in a production Flink cluster?

| Metric | Threshold | Meaning | |--------|-----------|---------| | numRecordsOutPerSecond | Baseline | Throughput sanity check | | currentInputWatermark | < now - tolerance | Watermark lag — pipeline is processing old events | | numberOfFailedCheckpoints | > 0 | Checkpoint instability — investigate | | lastCheckpointDuration | > checkpointInterval×0.8 | Checkpoints taking too long | | buffers.inPoolUsage | > 80% sustained | Backpressure building upstream | | fullRestarts | > 0 | Job restarted from checkpoint | | JVM GC pause | > 100ms | Heap pressure — consider RocksDB |

Advanced Patterns

Q25: What is the CEP (Complex Event Processing) library in Flink?

Flink CEP allows you to detect patterns in event streams. Example: detect fraud by finding 3 transactions > $100 from different countries within 10 minutes for the same user.

Pattern<Transaction, ?> fraudPattern = Pattern.<Transaction>begin("first")
    .where(SimpleCondition.of(tx -> tx.getAmount() > 100))
    .next("second")
    .where(SimpleCondition.of(tx -> tx.getAmount() > 100))
    .within(Time.minutes(10));

PatternStream<Transaction> patternStream = CEP.pattern(
    stream.keyBy(Transaction::getUserId),
    fraudPattern
);

patternStream.process(new PatternProcessFunction<>() {
    @Override
    public void processMatch(Map<String, List<Transaction>> match, ...) {
        // emit fraud alert
    }
});

Q26: What is broadcast state and when do you use it?

Broadcast state allows one stream (usually a small control/configuration stream) to be broadcast to all parallel instances of an operator, where it can be joined with records from the main stream.

Use case: Apply a set of dynamically-updated rules (e.g., fraud detection rules, ML model features) to every transaction event. The rules stream is broadcast; the transactions stream is partitioned by key.

BroadcastStream<Rule> broadcastRules = rulesStream.broadcast(rulesDescriptor);
stream.connect(broadcastRules)
    .process(new BroadcastProcessFunction<>() {
        @Override
        public void processElement(Transaction tx, ReadOnlyContext ctx, Collector<Alert> out) {
            // look up rule from broadcast state
        }
        @Override
        public void processBroadcastElement(Rule rule, Context ctx, Collector<Alert> out) {
            // update broadcast state with new rule
        }
    });

Q27: What is the difference between process time and event time joins in Flink?

Interval join (event time): Join events from two streams within a time window relative to the event timestamps. Maintains state for both streams until the watermark advances past the join window.

ordersStream
    .keyBy(Order::getOrderId)
    .intervalJoin(shipmentsStream.keyBy(Shipment::getOrderId))
    .between(Time.hours(-1), Time.hours(1))
    .process(new ProcessJoinFunction<>() {...});

Window join (event time or processing time): Both streams are windowed; joined if they fall in the same window.

Key difference: Interval join is more flexible (asymmetric bounds), keeps less state for one-sided delays, and handles late events more gracefully. Window join requires both events to land in the same window.

Scenario Questions

Q28: Design a real-time fraud detection pipeline using Flink.

Kafka (transactions) → Flink Source → keyBy(userId)
  → CEP pattern (3 high-value txns in 10min from different countries)
  → Enrich with user profile (AsyncIO → Redis)
  → Score with ML model (stateless UDF)
  → Filter (score > threshold)
  → Kafka (fraud-alerts) → Alert Service
  → Elasticsearch (fraud-events) → Dashboard

Key design decisions:

• Use event time with 5s watermark — transaction timestamps are in payload
• CEP pattern with within(Time.minutes(10)) on per-user keyed stream
• RocksDB backend for user-level state (high cardinality)
• Savepoints before deploying new fraud rules; rule updates via broadcast state
• Checkpoint every 30s with incremental RocksDB — transaction volume is high

Q29: How would you handle a Flink job that keeps failing with OutOfMemoryError?

Diagnosis path:

1. Check whether it's heap OOM or off-heap (native memory). RocksDB uses off-heap; the JVM uses heap.
2. If heap OOM: reduce taskmanager.memory.jvm-overhead.fraction, switch to RocksDB backend to move state off-heap.
3. If RocksDB OOM: tune state.backend.rocksdb.memory.managed=true (managed memory pool) to prevent unbounded growth.
4. Check for state accumulation: any MapState or ListState that never expires is a memory leak. Add state TTL.
5. Reduce parallelism if each task is holding too much in-flight data.

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(7))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .build();
valueStateDescriptor.enableTimeToLive(ttlConfig);

Q30: Flink vs Spark Structured Streaming — when would you choose each?

Choose Flink when:

• Sub-second latency is required (Flink achieves ~10–100ms; Spark minimum ~100ms)
• Complex stateful operations: session windows, CEP, iterative algorithms
• Long-running stateful pipelines with large state (terabytes in RocksDB)
• You need native exactly-once with Kafka sources and sinks
• Your team works in Java/Scala and values fine-grained operator control

Choose Spark Structured Streaming when:

• You already have a large Spark ecosystem (Databricks, Delta Lake, MLlib)
• Batch and streaming jobs share the same codebase
• The pipeline is primarily stateless transformations and simple aggregations
• Python is the primary language (PySpark has better Structured Streaming support than PyFlink)
• Latency > 1 second is acceptable

Interview answer: "I use Flink for latency-sensitive pipelines (sub-second alerting, session analytics) and complex stateful logic. I use Spark Structured Streaming when the team is Spark-native, the pipeline is mostly batch-with-streaming-extension, or Delta Lake integration is central to the architecture."