08 - Kafka + Apache Flink: Pipeline Analytics Real-Time and Iceberg Sink

When Kafka Streams is not enough: reading from Kafka and writing to a data lake

You already have a stream of orders on Kafka and a service that processes it with Kafka Streams. Then requests of a different kind arrive: join the orders stream to the payments stream with a time constraint, compute windows over several topics, and write the result in a format that analysts query with SQL, not into another topic. For this second job Apache Flink is often used: a distributed processing engine, with its own cluster, that reads from Kafka and writes elsewhere. The useful question is not "which is better" but "what changes if I choose it": where it runs, who holds the state, what happens if a partition stops.

This article concentrates on what you can understand without a Flink cluster: event time and watermarks, which decide when a window is closed. A Java 21 model that I compiled and ran demonstrates it. The rest (the SQL syntax for Kafka tables, windows, the Iceberg table) is a template to adapt, not run, because it requires Flink and Kafka: the syntax comes from the documentation of Flink 2.3 ("stable" version) and Apache Iceberg 1.12, consulted on 2026-11-12. I report no benchmarks and no throughput figures.

Event time and watermarks: when is a window closed?

A window "orders of minutes 0-5" can be closed only when you know that no more events from those minutes will arrive. But events arrive out of order. Flink solves the problem with watermarks: a signal that says "no more events with a time earlier than T will arrive". In the SQL DDL you declare it with WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND, which means: consider a disorder of up to 5 seconds acceptable. The Flink documentation adds a decisive rule for Kafka: when an operator has several inputs, its time is the minimum of the inputs' times, and for the Kafka source the output watermark is the minimum among the watermarks of the partitions read. If a partition receives no events, its watermark does not advance and blocks everything; the documented remedy is the table.exec.source.idle-timeout option.

The model applies these rules to an operator with two partitions and 5-minute tumbling windows. The watermark allows one minute of disorder.

import java.util.*;

public class Watermark {
    static final long MIN = 60_000L, RITARDO = 1 * MIN;            // the watermark allows 1 minute of disorder

    // An "operator" with two input partitions: its time is the MINIMUM of the partitions' watermarks.
    static class Operatore {
        final long[] massimo = {Long.MIN_VALUE, Long.MIN_VALUE};   // latest event seen for each partition
        final boolean[] inattiva = {false, false};
        final Map<Long, Integer> conteggi = new TreeMap<>();       // window (start, in minutes) -> count
        final Set<Long> chiuse = new TreeSet<>();
        int scartati = 0;

        long watermark() {
            long w = Long.MAX_VALUE;
            for (int p = 0; p < 2; p++) {
                if (inattiva[p]) continue;                          // a partition declared idle no longer holds things back
                w = Math.min(w, massimo[p] == Long.MIN_VALUE ? Long.MIN_VALUE : massimo[p] - RITARDO);
            }
            return w;
        }

        void ricevi(int partizione, long minuto) {
            long t = minuto * MIN, inizio = (t / (5 * MIN)) * 5;     // 5-minute tumbling windows
            if (chiuse.contains(inizio)) { scartati++; return; }    // the window has already been emitted: late event
            conteggi.merge(inizio, 1, Integer::sum);
            massimo[partizione] = Math.max(massimo[partizione], t);
            for (long i : new ArrayList<>(conteggi.keySet()))       // emits the windows closed by the watermark
                if (watermark() >= (i + 5) * MIN && chiuse.add(i))
                    System.out.println("   emitted window [" + i + "," + (i + 5) + ") = " + conteggi.get(i));
        }
    }

    public static void main(String[] args) {
        System.out.println("== both partitions receive events");
        Operatore a = new Operatore();
        a.ricevi(0, 1); a.ricevi(1, 2); a.ricevi(0, 7); a.ricevi(1, 8); a.ricevi(0, 4);
        System.out.println("   late events dropped: " + a.scartati);

        System.out.println("== partition 1 stops receiving after minute 2");
        Operatore b = new Operatore();
        b.ricevi(1, 2);
        for (long m = 1; m <= 20; m++) b.ricevi(0, m);
        System.out.println("   windows emitted: " + b.chiuse + "  (watermark = " + b.watermark() / MIN + " min)");

        System.out.println("== same situation, partition 1 declared idle");
        b.inattiva[1] = true;
        b.ricevi(0, 21);
        System.out.println("   windows emitted: " + b.chiuse + "  (watermark = " + b.watermark() / MIN + " min)");
    }
}

// Output:
// == both partitions receive events
//    emitted window [0,5) = 2
//    late events dropped: 1
// == partition 1 stops receiving after minute 2
//    windows emitted: []  (watermark = 1 min)
// == same situation, partition 1 declared idle
//    emitted window [0,5) = 5
//    emitted window [5,10) = 5
//    emitted window [10,15) = 5
//    emitted window [15,20) = 5
//    windows emitted: [0, 5, 10, 15]  (watermark = 20 min)

In the first scene the window [0,5) closes only when minute 8 arrives from partition 1 and takes the watermark to 6 (the minimum of 7-1 and 8-1): it counted 2 events, minutes 1 and 2. The event with minute 4, which arrived later, finds the window already emitted and is dropped (late events dropped: 1). This is the price of every watermark: the tighter it is, the sooner you emit; the wider it is, the fewer late events you lose. In the second scene partition 1 stops after minute 2 while partition 0 continues up to minute 20: no window is emitted, and the watermark stays at 1 minute. In the third you declare partition 1 idle (which is what Flink's idle-timeout does): the watermark jumps to 20 and the four accumulated windows all come out together.

The practical consequence is that a topic with unbalanced or empty partitions is the typical reason why a Flink query "produces nothing". The problem is not your SQL: it is a stalled watermark. Remember it when the results do not show up: look at the watermark first, then at the query. The parallel with Kafka Streams is useful but not identical: there, as seen in the article on Streams, the time that flows is the stream's time (the maximum timestamp seen) and the margin is the grace period.

Kafka tables in Flink SQL, written the way the documentation wants

In Flink SQL a topic is a table. The kafka connector requires at least topic, properties.bootstrap.servers and a format; properties.group.id is the group's identity; scan.startup.mode decides where to start from. To write an updatable result (a row that changes over time) there is the upsert-kafka connector, for which the documentation recommends declaring the primary key in the DDL, and which requires separate formats for key and value (key.format, value.format). The template below puts the two things together: a source table with the watermark, an upsert-type destination table, and the continuous query that links them.

-- Template to adapt (not run: it needs Flink SQL and a Kafka cluster). Syntax from the Flink 2.3 documentation.
CREATE TABLE ordini (
    order_id   STRING,
    user_id    STRING,
    amount     DECIMAL(10, 2),
    created_at BIGINT,                                       -- milliseconds, from the payload
    event_time AS TO_TIMESTAMP_LTZ(created_at, 3),           -- the event time, not the arrival time
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'properties.bootstrap.servers' = 'kafka-1:9092',
    'properties.group.id' = 'flink-analytics',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
);

-- The result goes to an "upsert" topic: it needs the primary key, and one format for the key and one for the value.
CREATE TABLE spesa_per_finestra (
    window_start TIMESTAMP(3),
    user_id      STRING,
    ordini       BIGINT,
    totale       DECIMAL(12, 2),
    PRIMARY KEY (window_start, user_id) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'spesa-per-finestra',
    'properties.bootstrap.servers' = 'kafka-1:9092',
    'key.format' = 'json',
    'value.format' = 'json'
);

INSERT INTO spesa_per_finestra
SELECT window_start, user_id, COUNT(*), SUM(amount)
FROM TABLE(TUMBLE(TABLE ordini, DESCRIPTOR(event_time), INTERVAL '10' MINUTES))
GROUP BY window_start, window_end, user_id;

Three points to note. The TUMBLE function returns the original relation plus the columns window_start, window_end and window_time, and that is why the GROUP BY includes them. The output's primary key, (window_start, user_id), is what allows the upsert to replace the value of a window when late data that corrects it arrives. And NOT ENFORCED is a declaration, not a check: Flink does not verify uniqueness, you are the one who guarantees it. If you need to join the orders stream to the payments stream, the right form is the interval join: it requires at least one equality predicate and a condition that bounds the time on both sides, for example o.order_time BETWEEN s.ship_time - INTERVAL '4' HOUR AND s.ship_time as in the documentation. It works only on append-only tables and with time attributes, and thanks to the time bounds Flink can free the old state.

Guarantees: Flink, checkpoints and Kafka transactions

At intervals (execution.checkpointing.interval, no default value: you have to set it) Flink saves a checkpoint of the state and of the read positions. If a node goes down, the job restarts from the last checkpoint. The checkpoint documentation provides a choice between exactly-once, preferable for most applications, and at-least-once, suited to ultra-low-latency applications. Towards Kafka, exactly-once writing is a sink option: sink.delivery-guarantee defaults to at-least-once and accepts exactly-once and none; with exactly-once you must also set sink.transactional-id-prefix, because Flink uses Kafka transactions. The corollary is the one from the article on transactions: the read_committed consumers downstream will see the data only after the transaction is committed, that is with the latency of one checkpoint.

Iceberg: the table that Flink writes at every checkpoint

Apache Iceberg is a table format for files on shared storage: it adds metadata and snapshots on top of the data files, so that several engines read the same table. The Iceberg documentation says the Flink sink "guarantees exactly-once semantics" and explains the mechanism: the data written stays as temporary files until the checkpoint completes, and only then does Flink commit to Iceberg. Two consequences. First: the data becomes visible to readers at the checkpoint interval, not on every row. Second: the documentation warns that an Iceberg commit can fail while the Flink checkpoint succeeds (or that the completion notification does not arrive), and recommends alerting on the time elapsed since the last successful commit.

-- Template to adapt (not run): Iceberg catalog and table from Flink SQL, syntax from the Iceberg 1.12 documentation.
CREATE CATALOG lago WITH (
    'type' = 'iceberg',
    'catalog-type' = 'hadoop',
    'warehouse' = 'file:///data/warehouse',      -- a path on your shared storage
    'property-version' = '1'
);

CREATE DATABASE IF NOT EXISTS lago.analitica;

CREATE TABLE lago.analitica.spesa_per_finestra (
    window_start TIMESTAMP(3),
    user_id      STRING,
    ordini       BIGINT,
    totale       DECIMAL(12, 2)
) PARTITIONED BY (user_id);

-- A continuous job: the data becomes visible in the table when Flink completes a checkpoint.
INSERT INTO lago.analitica.spesa_per_finestra
SELECT window_start, user_id, COUNT(*), SUM(amount)
FROM TABLE(TUMBLE(TABLE ordini, DESCRIPTOR(event_time), INTERVAL '10' MINUTES))
GROUP BY window_start, window_end, user_id;

The hadoop catalog uses a directory as the metadata registry and is the simplest for trying things out; on shared infrastructure you use the Hive or REST catalog (same CREATE CATALOG syntax with a different catalog-type). A limitation written in the documentation: Iceberg has hidden partitioning, but from Flink DDL you cannot partition by a function of a column, only by columns. Another important note: the documentation warns that expiring the snapshot or deleting orphan files can corrupt the state of the Flink jobs that are writing, because they remember the last committed checkpoint in the snapshot summary: table maintenance must be done with care.

Flink or Kafka Streams? The criterion

Kafka Streams is a library inside your application: no additional cluster, but its world is Kafka (it reads and writes topics, with the guarantees seen). Flink is a system with its own cluster: it costs extra operations, but it reads and writes many systems (Kafka, Iceberg tables, databases), has SQL as an interface and scales independently of the application service. If your problem is enriching events inside a Java microservice, Streams is enough and costs less. If the problem is putting several sources together, producing tables for analysis and giving SQL to those who do not write Java, Flink is the way, with its management costs. In both cases, event time and its lateness margin are the most important choice you will make, because they decide which data you lose and how long you wait.

What you can do now

You can explain why a window closes with a watermark and not with the clock, what it means that the watermark is the minimum across partitions and why an empty partition can block a query. You can write in Flink SQL a Kafka table with a watermark, a tumbling window and an upsert output with a primary key, and an interval join with time constraints. You know that exactly-once towards Kafka goes through transactions and towards Iceberg through the checkpoint, and what latency follows. When you want to bring events from the database to Kafka before analyzing them, see Kafka Connect and Debezium; when you want to check that the pipeline works, the monitoring.