06 - Kafka Streams: Stream Processing Embedded in Java, KTable e Windowing

Contare gli ordini di ogni utente ogni cinque minuti, dentro il tuo servizio

Hai un servizio Java che già legge ordini da un topic. Ti chiedono di mostrare, in tempo reale, quanti ordini ha fatto ogni utente negli ultimi cinque minuti, e di arricchire ogni ordine con il nome del prodotto. Puoi scrivere un ciclo con un consumatore, una mappa in memoria e un database per non perdere i conteggi al riavvio. Oppure usare Kafka Streams: una libreria che si aggiunge alla tua applicazione e fa girare la logica di stream processing dentro i tuoi processi, senza un cluster separato. La documentazione lo dice con una frase netta: Kafka Streams «is not a resource manager, but a library that runs anywhere its stream processing application runs».

La libreria ti dà quattro strumenti. Il flusso (KStream) e la tabella (KTable), due modi di leggere lo stesso topic. Lo stato locale con il suo log di recupero. Le finestre temporali. I join tra flusso e tabella. In questo articolo li vedi eseguiti davvero: ho usato kafka-streams-4.4.0.jar e il TopologyTestDriver di kafka-streams-test-utils (Maven Central, SHA-1 verificati), che fa girare la topologia in memoria, un record alla volta, senza broker. Per questo l'articolo mostra comportamenti veri della libreria, non un modello inventato; non mostra invece comportamenti che richiedono un cluster (rebalance, ripristino dello stato dopo un guasto), che descrivo dalla documentazione 4.4 consultata il 2026-11-12. Gli stati sono in memoria (dsl.store.suppliers.class impostato su quelli in memoria) per non dipendere da RocksDB, il motore su disco predefinito.

Topologia, task e thread: dove sta il parallelismo

Il codice che scrivi descrive una topologia: un grafo di operatori (filtri, trasformazioni, aggregazioni) collegati da flussi di record. La libreria la scompone in task, un numero fisso deciso dalle partizioni dei topic di ingresso: ogni task riceve un insieme di partizioni e quell'assegnazione non cambia mai. Ne deriva la regola che vale per ogni applicazione Streams: il parallelismo massimo è il numero di partizioni di ingresso. Con un topic da cinque partizioni puoi avviare fino a cinque istanze; la sesta resta in attesa e subentra se una cade. È la stessa regola dei gruppi di consumatori dell'articolo sui fondamenti, perché sotto c'è lo stesso meccanismo.

Dentro un'istanza, i task girano su thread (num.stream.threads, default 1) senza stato condiviso. Il listato stampa dalla libreria reale i default che contano per un'applicazione nuova.

import org.apache.kafka.streams.StreamsConfig;
import java.util.*;

public class ConfigStreams {
    public static void main(String[] args) {
        Map<String, Object> p = new HashMap<>();
        p.put(StreamsConfig.APPLICATION_ID_CONFIG, "analisi-ordini");
        p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");   // non viene contattato
        StreamsConfig c = new StreamsConfig(p);
        for (String k : List.of("num.stream.threads", "num.standby.replicas", "commit.interval.ms",
                "statestore.cache.max.bytes", "state.dir", "deserialization.exception.handler",
                "processing.exception.handler", "production.exception.handler", "errors.dead.letter.queue.topic.name"))
            System.out.println(k + " = " + c.values().get(k));
    }
}

// Output:
// num.stream.threads = 1
// num.standby.replicas = 0
// commit.interval.ms = 30000
// statestore.cache.max.bytes = 10485760
// state.dir = /tmp/kafka-streams
// deserialization.exception.handler = class org.apache.kafka.streams.errors.LogAndFailExceptionHandler
// processing.exception.handler = class org.apache.kafka.streams.errors.LogAndFailProcessingExceptionHandler
// production.exception.handler = class org.apache.kafka.streams.errors.DefaultProductionExceptionHandler
// errors.dead.letter.queue.topic.name = null

Quattro righe meritano attenzione. num.standby.replicas = 0 significa che dopo un guasto lo stato di un task va ricostruito rileggendo il suo changelog, e quindi che il recupero richiede tempo proporzionale alla quantità di stato; le repliche di standby lo accorciano a costo di memoria e disco. commit.interval.ms = 30000 vale per la garanzia at-least-once: con exactly_once_v2 scende a 100 (articolo sulle transazioni). state.dir = /tmp/kafka-streams è il percorso predefinito, calcolato dalla directory temporanea del sistema: la documentazione avverte che deve essere unico per ogni istanza che condivide il filesystem, e su molti sistemi /tmp viene svuotato al riavvio, quindi in produzione imposta un percorso persistente. Gli ultimi tre sono i gestori d'errore: due falliscono (LogAndFail...) e uno applica il comportamento predefinito. Torneremo su errors.dead.letter.queue.topic.name, che per default è nullo.

KStream e KTable: due letture dello stesso topic

Un ordine passa da PENDING a CONFIRMED a SHIPPED: tre record con la stessa chiave. Se ti interessa ogni cambiamento, leggi il topic come KStream: ogni record è un evento indipendente. Se ti interessa lo stato attuale, lo leggi come KTable: ogni record con la stessa chiave sostituisce il precedente. La documentazione chiama questo legame «stream-table duality»: un flusso è il registro dei cambiamenti di una tabella, e una tabella è un'istantanea di un flusso. Il topic compattato dell'articolo sui fondamenti sfrutta la stessa idea.

import org.apache.kafka.common.serialization.*;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.BuiltInDslStoreSuppliers;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.common.utils.Bytes;
import java.util.*;

public class StreamVsTable {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();

        // Lo stesso tipo di dato letto in due modi.
        KStream<String, String> eventi = builder.stream("stati-ordine", Consumed.with(Serdes.String(), Serdes.String()));
        eventi.groupByKey().count(Materialized.as("eventi-per-ordine")).toStream()
              .to("conteggi", Produced.with(Serdes.String(), Serdes.Long()));
        builder.table("stati-ordine-tabella", Consumed.with(Serdes.String(), Serdes.String()),
                      Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("stato-attuale"));

        Properties p = new Properties();
        p.put(StreamsConfig.APPLICATION_ID_CONFIG, "demo");
        p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");         // mai contattato
        p.put(StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG, BuiltInDslStoreSuppliers.InMemoryDslStoreSuppliers.class);
        p.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);           // un output per ogni aggiornamento

        try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), p)) {
            TestInputTopic<String, String> in = driver.createInputTopic("stati-ordine",
                new StringSerializer(), new StringSerializer());
            TestInputTopic<String, String> inTab = driver.createInputTopic("stati-ordine-tabella",
                new StringSerializer(), new StringSerializer());
            for (String stato : List.of("PENDING", "CONFIRMED", "SHIPPED")) {
                in.pipeInput("ordine-123", stato);
                inTab.pipeInput("ordine-123", stato);
            }
            TestOutputTopic<String, Long> out = driver.createOutputTopic("conteggi",
                new StringDeserializer(), new LongDeserializer());
            System.out.println("KStream: aggiornamenti del conteggio per ordine-123 -> " + out.readValuesToList());
            KeyValueStore<String, String> stato = driver.getKeyValueStore("stato-attuale");
            System.out.println("KTable: stato attuale di ordine-123 -> " + stato.get("ordine-123"));
        }
    }
}

// Output:
// KStream: aggiornamenti del conteggio per ordine-123 -> [1, 2, 3]
// KTable: stato attuale di ordine-123 -> SHIPPED

Il primo ramo conta gli eventi per chiave: l'uscita è [1, 2, 3], un aggiornamento per ciascun evento. Il secondo ramo è una tabella e conserva soltanto l'ultimo valore: SHIPPED. La riga che disattiva la cache (statestore.cache.max.bytes = 0) serve solo alla dimostrazione: con i 10 MiB predefiniti la libreria fonde più aggiornamenti della stessa chiave prima di emetterli, e vedresti meno righe in uscita. Un'aggregazione (count, aggregate) produce sempre una KTable, anche partendo da un flusso: è ciò che permette alla libreria di correggere il valore se arriva un record fuori ordine.

Lo stato di un'aggregazione vive in uno state store locale. Per non perderlo con la macchina, ogni store ha un changelog: un topic Kafka compattato, partizionato come lo store, in cui la libreria registra ogni aggiornamento. Se il task riparte su un'altra macchina, rilegge il changelog e ricostruisce lo store (documentazione «Architecture»). Non è tutto gratis: ogni aggregazione crea un topic in più, e il tempo di ripartenza dipende dalla sua dimensione.

Finestre: contare per intervalli di tempo

Una finestra raggruppa i record per il timestamp dell'evento, non per il momento in cui arrivano. Le tre forme principali: la tumbling ha finestre fisse e senza sovrapposizione (0-5, 5-10 minuti); la hopping ha finestre di dimensione fissa che avanzano di un passo più piccolo e quindi si sovrappongono; la session raggruppa gli eventi di uno stesso utente separati da meno di un intervallo di inattività. Esiste anche la sliding, che segue ogni evento. Ogni finestra ha un grace period: la documentazione spiega che un record viene scartato se il suo timestamp lo assegna a una finestra ma il tempo corrente dello stream è oltre la fine della finestra più il grace.

import org.apache.kafka.common.serialization.*;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.BuiltInDslStoreSuppliers;
import java.time.*;
import java.util.*;

public class Finestre {
    static final long MIN = 60_000L;

    static KeyValue<String, Long> etichetta(Windowed<String> w, Long n) {
        return KeyValue.pair(w.key() + " [" + w.window().start() / MIN + "," + w.window().end() / MIN + ")", n);
    }

    public static void main(String[] args) {
        StreamsBuilder b = new StreamsBuilder();
        KGroupedStream<String, String> perUtente = b.<String, String>stream("ordini").groupByKey();

        perUtente.windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1)))
            .count().toStream().map(Finestre::etichetta).to("tumbling", Produced.with(Serdes.String(), Serdes.Long()));
        perUtente.windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(10), Duration.ofMinutes(1))
                .advanceBy(Duration.ofMinutes(5)))
            .count().toStream().map(Finestre::etichetta).to("hopping", Produced.with(Serdes.String(), Serdes.Long()));
        perUtente.windowedBy(SessionWindows.ofInactivityGapAndGrace(Duration.ofMinutes(30), Duration.ofMinutes(1)))
            .count().toStream().map(Finestre::etichetta).to("sessioni", Produced.with(Serdes.String(), Serdes.Long()));

        Properties p = new Properties();
        p.put(StreamsConfig.APPLICATION_ID_CONFIG, "demo-finestre");
        p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
        p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        p.put(StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG, BuiltInDslStoreSuppliers.InMemoryDslStoreSuppliers.class);
        p.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);

        try (TopologyTestDriver driver = new TopologyTestDriver(b.build(), p)) {
            TestInputTopic<String, String> in = driver.createInputTopic("ordini",
                new StringSerializer(), new StringSerializer());
            // minuti dell'ora dell'evento: 1, 3, 7, 40 e poi un ritardatario con ora 2, arrivato dopo il minuto 40
            for (long minuto : new long[] {1, 3, 7, 40, 2})
                in.pipeInput("u1", "ordine", Instant.ofEpochMilli(minuto * MIN));

            for (String topic : List.of("tumbling", "hopping", "sessioni")) {
                TestOutputTopic<String, Long> out = driver.createOutputTopic(topic,
                    new StringDeserializer(), new LongDeserializer());
                System.out.println(topic + ": " + new TreeMap<>(out.readKeyValuesToMap()));
            }
        }
    }
}

// Output:
// tumbling: {u1 [0,5)=2, u1 [40,45)=1, u1 [5,10)=1}
// hopping: {u1 [0,10)=3, u1 [35,45)=1, u1 [40,50)=1, u1 [5,15)=1}
// sessioni: {u1 [1,1)=null, u1 [1,3)=null, u1 [1,7)=3, u1 [40,40)=1}

Gli eventi dell'utente u1 hanno ora (in minuti) 1, 3, 7 e 40, e per ultimo ne arriva uno con ora 2: il ritardatario. Nella tumbling da 5 minuti la finestra [0,5) conta 2 (i minuti 1 e 3) e non 3, perché il ritardatario è arrivato quando il tempo dello stream era 40, molto oltre la fine della finestra più il grace di un minuto, e viene scartato. Nella hopping da 10 minuti con passo 5 il minuto 7 compare in due finestre, [0,10) (che conta 3) e [5,15); il minuto 40 in altre due. Nelle sessioni, con inattività massima di 30 minuti, i minuti 1, 3 e 7 formano un'unica sessione [1,7) di 3 eventi, e il 40 ne apre un'altra. Le righe con valore null sono le sessioni intermedie ([1,1) e [1,3)) che si sono fuse in quella più grande: la libreria emette un tombstone per ciascuna.

Due avvertenze. Se vuoi un risultato solo quando la finestra si chiude, e non ogni aggiornamento, la documentazione indica suppress(Suppressed.untilWindowCloses(...)). E se imposti un grace troppo piccolo perdi dati in silenzio, troppo grande ritardi l'emissione: la scelta dipende dal ritardo massimo che i tuoi produttori possono accumulare, e va misurata sul tuo flusso.

Arricchire un flusso con una tabella

Il caso più frequente è arricchire gli eventi con dati di riferimento: l'ordine con il nome del prodotto. In Streams è un join tra un KStream (gli ordini) e una KTable (il catalogo), che richiede una cosa che la documentazione chiama co-partizionamento: i due topic devono avere lo stesso numero di partizioni e la stessa strategia di partizionamento per chiave, perché il join avviene su record con la stessa chiave nello stesso task. L'eccezione è il join con una GlobalKTable, che ha una copia completa su ogni istanza e non richiede co-partizionamento: costa però spazio sul disco e traffico di rete, perché ogni istanza legge tutto il topic.

import org.apache.kafka.common.serialization.*;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.BuiltInDslStoreSuppliers;
import java.util.*;

public class Arricchimento {
    public static void main(String[] args) {
        StreamsBuilder b = new StreamsBuilder();
        // Ordini: chiave = id prodotto, valore = id ordine. Catalogo: chiave = id prodotto, valore = nome.
        KStream<String, String> ordini = b.stream("ordini", Consumed.with(Serdes.String(), Serdes.String()));
        KTable<String, String> catalogo = b.table("catalogo", Consumed.with(Serdes.String(), Serdes.String()));

        ordini.leftJoin(catalogo, (ordine, prodotto) -> ordine + " -> " + (prodotto == null ? "prodotto sconosciuto" : prodotto))
              .to("arricchiti", Produced.with(Serdes.String(), Serdes.String()));

        Properties p = new Properties();
        p.put(StreamsConfig.APPLICATION_ID_CONFIG, "demo-join");
        p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
        p.put(StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG, BuiltInDslStoreSuppliers.InMemoryDslStoreSuppliers.class);

        try (TopologyTestDriver driver = new TopologyTestDriver(b.build(), p)) {
            TestInputTopic<String, String> inOrdini = driver.createInputTopic("ordini", new StringSerializer(), new StringSerializer());
            TestInputTopic<String, String> inCatalogo = driver.createInputTopic("catalogo", new StringSerializer(), new StringSerializer());

            inOrdini.pipeInput("p-9", "ordine-1");              // il prodotto non è ancora nel catalogo
            inCatalogo.pipeInput("p-9", "Tastiera meccanica");  // ora sì
            inOrdini.pipeInput("p-9", "ordine-2");

            TestOutputTopic<String, String> out = driver.createOutputTopic("arricchiti", new StringDeserializer(), new StringDeserializer());
            out.readValuesToList().forEach(System.out::println);
        }
    }
}

// Output:
// ordine-1 -> prodotto sconosciuto
// ordine-2 -> Tastiera meccanica

Il primo ordine arriva quando il catalogo non ha ancora il prodotto: il leftJoin lo lascia passare e il valore del prodotto è null, quindi esce «prodotto sconosciuto». Dopo l'arrivo del record del catalogo, il secondo ordine viene arricchito correttamente. La lezione è che il join flusso-tabella non è retroattivo: l'ordine 1 non viene riemesso quando la tabella si aggiorna, perché a far scattare il join è solo l'arrivo di un record dal lato flusso. Un inner join avrebbe scartato l'ordine 1; il left join lo conserva con il segnaposto. Scegli in base a cosa è meno grave: perdere un ordine o emetterlo con un dato mancante.

Errori e interrogazioni dello stato

Il gestore d'errore per i record che non si riescono a deserializzare è, per default, LogAndFailExceptionHandler: l'applicazione si ferma. È la scelta prudente, perché un record corrotto scartato in silenzio è un dato perso; LogAndContinueExceptionHandler lo salta e continua. Nelle versioni recenti la configurazione ha una novità: errors.dead.letter.queue.topic.name, che per default è nullo. Se lo imposti, il gestore predefinito costruisce un record per la dead letter queue e lo invia a quel topic. Non vale per gli store globali, e viene ignorato se hai impostato un tuo gestore. Il dettaglio dei pattern di scarto è nell'articolo sulle dead letter queue.

Lo stato locale si può anche interrogare dall'esterno della topologia (le interactive queries, con KafkaStreams.store(...)); non ne mostro un listato perché richiede un'istanza avviata contro un broker. L'idea da portarti dietro è che ogni istanza conosce solo le partizioni che gestisce: per rispondere sull'intero stato serve sapere quale istanza possiede la chiave.

Che cosa sai fare adesso

Sai che Kafka Streams è una libreria e che il suo parallelismo massimo è il numero di partizioni di ingresso. Sai scegliere tra lettura come flusso e come tabella e perché un'aggregazione produce sempre una tabella. Sai prevedere quali eventi entrano in una finestra tumbling, hopping o session e perché un ritardatario può essere scartato. Sai che il join flusso-tabella non è retroattivo e che richiede co-partizionamento. E ricordi di impostare state.dir su un percorso persistente. Quando le pipeline diventano analitiche su più flussi e con scaling indipendente, il confronto è con Apache Flink; per portare i dati dentro e fuori da Kafka senza scrivere codice, con Kafka Connect. Se arrivi da un altro punto della serie, i collegamenti sono: lo Schema Registry.