Kafka Streams 3/4 — Joins & Windowing

1Vue d'ensemble des joins

Kafka Streams supporte plusieurs types de joins selon la nature des sources à combiner.

Le co-partitionnement signifie que les deux topics doivent avoir le même nombre de partitions et que les messages de même clé doivent être dans les mêmes partitions. Si ce n'est pas le cas, Kafka Streams ne peut pas garantir que les deux côtés du join arrivent sur le même thread. (La variante foreign-key du join KTable × KTable, vue en section 5️⃣, lève cette contrainte.)

Co-partitionnement en détail

Pourquoi « mêmes clés dans les mêmes partitions » ? La partition d'un message est décidée à la production par hash(clé) % nombre_de_partitions. La même clé produit le même hash, donc le même numéro de partition — quel que soit le topic. Si orders et customers sont tous deux clés par customerId avec le même nombre de partitions, le client C-42 tombe sur la même partition des deux côtés. La task qui traite cette partition a donc les deux messages en local, sans réseau ni coordination — le join est une simple lecture du state store local.

Co-partitionner, c'est garantir trois conditions ensemble :

1. Même clé de partition des deux côtés — le champ de jointure doit être la clé du message.
2. Même nombre de partitions sur les deux topics.
3. Même partitioner (la fonction de hash) — le partitioner par défaut, sauf si quelqu'un en a posé un custom.

Modèle d'exécution : partition → task → thread. Une task traite une partition de tous les topics co-partitionnés (la task 2 lit la partition 2 de chaque topic). Les tasks sont réparties sur les stream threads et sur les instances, mais une task n'est jamais coupée entre deux threads. Conséquence directe : les deux côtés d'une même clé sont toujours dans la même task, donc sur le même thread — à condition que le co-partitionnement tienne. Un message « en retard » ne change pas de partition (la partition dépend de la clé, pas du moment d'arrivée) ; il reste sur sa task.

2KStream × KTable — enrichir un événement

C'est le join le plus courant. On a un flux d'événements (KStream) et une table qui contient l'état courant d'une entité (KTable). Pour chaque événement, on veut récupérer les informations de l'entité correspondante au moment où l'événement arrive.

Un exemple concret : à chaque commande (KStream), on enrichit avec les informations du client (KTable). Ici le flux orders est partitionné par customerId pour co-partitionner avec customers. Si le client est mis à jour entre deux commandes, la commande suivante verra la nouvelle version du client.

KStream<String, OrderPlaced> orders   = builder.stream("orders");      // source : flux des commandes (clé = customerId)
KTable<String, Customer>     customers = builder.table("customers");    // table d'état : clients (clé = customerId)

KStream<String, OrderEnriched> enriched = orders.join(  // enrichit chaque commande avec son client
    customers,                                          // table de lookup (co-partitionnée par customerId)
    (order, customer) -> OrderEnriched.newBuilder()     // combine commande + client
        .setOrderId(order.getOrderId())
        .setCustomerName(customer.getName())
        .setCustomerTier(customer.getTier())
        .setAmount(order.getAmount())
        .build()
);

Si on veut garder la commande même quand le client n'existe pas dans la table, on utilise leftJoin. Dans ce cas, customer peut être null côté droit.

KStream<String, OrderEnriched> withFallback = orders.leftJoin(  // leftJoin : garde la commande même sans client trouvé
    customers,                                                 // table de lookup
    (order, customer) -> OrderEnriched.newBuilder()            // customer peut être null ici
        .setOrderId(order.getOrderId())
        .setCustomerName(customer != null ? customer.getName() : "UNKNOWN")  // fallback si client absent
        .setAmount(order.getAmount())
        .build()
);

3KStream × GlobalKTable — enrichir sans contrainte

Le join avec une GlobalKTable fonctionne comme le join avec une KTable, mais sans contrainte de co-partitionnement — la GlobalKTable est répliquée en entier sur chaque instance, donc toutes les données sont toujours disponibles localement.

Une autre différence : on peut chercher dans la GlobalKTable avec une clé différente de la clé du message. C'est utile quand la clé de lookup est dans la valeur du message, pas dans sa clé.

GlobalKTable<String, Country> countries = builder.globalTable("countries");  // référentiel répliqué sur chaque instance

KStream<String, OrderEnriched> enriched = orders.join(  // join stream × GlobalKTable
    countries,                                          // table de référence (pas de co-partitionnement requis)
    (customerId, order) -> order.getCountryCode(),      // clé de lookup = un champ de la valeur (pas la clé du stream)
    (order, country) -> OrderEnriched.newBuilder()      // combine commande + pays
        .setOrderId(order.getOrderId())
        .setCountryName(country.getName())
        .build()
);

4KStream × KStream — corréler deux flux dans le temps

Joindre deux KStream est différent des autres joins. Comme les deux sources sont des flux infinis, il faut définir une fenêtre de temps qui délimite la plage dans laquelle deux messages doivent arriver pour être joints.

Si un message arrive dans le stream A à 10h00 et qu'on définit une fenêtre de 5 minutes, on le joint avec tous les messages du stream B arrivés entre 9h55 et 10h05. Les messages du stream B en dehors de cette plage ne sont pas joints avec ce message.

KStream<String, OrderPlaced>     orders   = builder.stream("orders");      // 1er flux d'événements (clé = orderId)
KStream<String, PaymentReceived> payments = builder.stream("payments");    // 2ème flux d'événements (clé = orderId)

JoinWindows window = JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5));  // fenêtre ±5 min, sans grace

KStream<String, OrderWithPayment> ordersWithPayment = orders.join(  // corrèle commande et paiement dans la fenêtre
    payments,                                                       // 2ème flux à joindre
    (order, payment) -> OrderWithPayment.newBuilder()               // combine les deux quand ils matchent
        .setOrderId(order.getOrderId())
        .setAmount(order.getAmount())
        .setPaymentMethod(payment.getMethod())
        .build(),
    window                                                          // la fenêtre temporelle du join
);

Comme pour les autres joins, leftJoin garde la commande même sans paiement dans la fenêtre, et outerJoin garde les deux côtés même sans correspondance. À noter : en leftJoin/outerJoin stream-stream, le résultat avec null n'est émis qu'à la fermeture de la fenêtre (sinon on émettrait un null alors que le paiement va peut-être encore arriver) — d'où une latence égale à la durée de fenêtre sur ces résultats partiels.

5KTable × KTable — joindre deux états courants

Le join entre deux KTable combine leurs états courants. À chaque fois que l'une des deux tables est mise à jour pour une clé donnée, le join est recalculé pour cette clé.

KTable<String, Customer> customers = builder.table("customers");          // 1ère table (clé = customerId)
KTable<String, Loyalty>  loyalty   = builder.table("loyalty-accounts");   // 2ème table (clé = customerId)

KTable<String, CustomerView> combined = customers.join(  // join sur la clé primaire (recalculé des deux côtés)
    loyalty,                                             // 2ème table à joindre
    (customer, account) -> CustomerView.newBuilder()     // combine fiche client + compte fidélité
        .setCustomerId(customer.getCustomerId())
        .setName(customer.getName())
        .setPoints(account.getPoints())
        .build()
);

Foreign-key join — joindre sans co-partitionnement

Le join ci-dessus est un join sur la clé primaire : les deux tables doivent être co-partitionnées par la même clé. Quand la table de gauche référence l'autre par un champ (une clé étrangère) plutôt que par sa clé, Kafka Streams propose le foreign-key join (KIP-213) : on fournit un extracteur de clé étrangère, et Streams gère lui-même la repartition interne. Pas de co-partitionnement requis.

KTable<String, Order>    orders    = builder.table("orders");      // table de gauche (clé = orderId)
KTable<String, Customer> customers = builder.table("customers");   // table référencée (clé = customerId)

KTable<String, EnrichedOrder> enriched = orders.join(  // foreign-key join (KIP-213)
    customers,                                         // table cible de la clé étrangère
    order -> order.getCustomerId(),                    // extracteur de FK → Streams repartitionne en interne
    (order, customer) -> EnrichedOrder.newBuilder()    // combine commande + client
        .setOrderId(order.getOrderId())
        .setCustomerName(customer.getName())
        .build()
);

Le coût : des topics internes de souscription et un peu de latence. Mais ça évite de devoir repartitionner manuellement orders par customerId.

6Windowing — définir des tranches de temps

Sans fenêtre, une agrégation s'applique à tous les messages depuis le début du stream. Pour répondre à des questions comme "combien de commandes dans les 5 dernières minutes ?", on utilise des fenêtres.

Kafka Streams propose quatre types de fenêtres.

Tumbling windows — tranches fixes sans chevauchement

Le temps est découpé en intervalles de taille fixe. Chaque message appartient à exactement une fenêtre. La fenêtre la plus simple.

|--- 10h00-10h05 ---|--- 10h05-10h10 ---|--- 10h10-10h15 ---|
     msg1 msg2              msg3                 msg4

Un message à 10h03 est dans la fenêtre 10h00-10h05. Un message à 10h07 est dans 10h05-10h10. Jamais dans la même fenêtre.

KTable<Windowed<String>, Long> ordersPerWindow = stream
    .groupByKey()                                                      // groupe par customerId
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))  // fenêtres fixes de 5 min, sans grace
    .count(Materialized.as("orders-5min-store"));                      // compte les commandes par client et par fenêtre

La clé du résultat est Windowed<String> — elle contient le customerId et la fenêtre de temps. Le client C-1 dans 10h00-10h05 est une entrée distincte de C-1 dans 10h05-10h10.

Hopping windows — fenêtres qui se chevauchent

Comme les tumbling windows, mais les fenêtres avancent par pas plus petits que leur taille. Résultat : les fenêtres se chevauchent et un message peut appartenir à plusieurs fenêtres en même temps.

|--- 10h00-10h10 ---|
      |--- 10h05-10h15 ---|
            |--- 10h10-10h20 ---|

Un message à 10h07 est dans les fenêtres 10h00-10h10 et 10h05-10h15.

KTable<Windowed<String>, Long> ordersHopping = stream
    .groupByKey()                                                     // groupe par customerId
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(10), Duration.ofMinutes(1))  // fenêtre de 10 min (+1 min grace)
        .advanceBy(Duration.ofMinutes(5)))                            // avance de 5 min → fenêtres qui se chevauchent
    .count(Materialized.as("orders-hopping-store"));                  // compte par client et par fenêtre

Sliding windows — fenêtres autour des événements

Une sliding window est centrée sur chaque paire de messages. Deux messages se retrouvent dans la même fenêtre s'ils sont arrivés dans l'intervalle défini l'un de l'autre. La fenêtre n'est pas à des intervalles fixes — elle se crée autour des événements.

KTable<Windowed<String>, Long> ordersSliding = stream
    .groupByKey()                                                              // groupe par customerId
    .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)))  // même fenêtre si <5 min d'écart
    .count(Materialized.as("orders-sliding-store"));                           // compte par client et par fenêtre

Session windows — fenêtres basées sur l'activité

Une session window n'a pas de taille fixe. Elle regroupe les messages d'une même clé qui arrivent proches dans le temps, et se ferme automatiquement quand il n'y a plus d'activité pendant un certain délai appelé le gap.

Client C-1 :
10h00 msg1  10h02 msg2  10h04 msg3      silence 15 min      10h25 msg4  10h27 msg5
|---------- session 1 (6 min) ----------|                   |-- session 2 (4 min) --|

KTable<Windowed<String>, Long> sessions = stream
    .groupByKey()                                                                  // groupe par customerId
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(10)))  // nouvelle session après 10 min d'inactivité
    .count(Materialized.as("sessions-store"));                                     // compte les commandes par session

Comparaison des quatre types

7Grace period — accepter les messages tardifs

Dans un système distribué, les messages n'arrivent pas toujours dans l'ordre chronologique. Un message produit à 10h03 peut arriver dans Kafka après un message produit à 10h07 à cause d'un retard réseau.

Par défaut, quand une fenêtre se ferme, tous les messages qui arrivent après sont ignorés. Le grace period est un délai supplémentaire pendant lequel la fenêtre continue d'accepter des messages tardifs.

TimeWindows windowWithGrace = TimeWindows
    .ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1));  // fenêtre de 5 min + 1 min de tolérance aux retards

KTable<Windowed<String>, Long> orders = stream
    .groupByKey()                                       // groupe par customerId
    .windowedBy(windowWithGrace)                        // applique la fenêtre avec son grace period
    .count(Materialized.as("orders-with-grace-store")); // compte par client et par fenêtre

8Événement en retard : avec quelle version de la table joint-on ?

Le join KStream × KTable est un lookup : au moment où Kafka traite l'événement, il lit la valeur courante de l'entité dans la table. Quand un événement est traité en retard par rapport au moment où il s'est produit, ça soulève une question.

Exemple : une commande est passée à 10h00 (client domicilié à Paris). À 12h00, le client déménage (→ Lyon). À 14h00 seulement, la commande de 10h00 est traitée (retard réseau, rejeu). Une KTable normale la joint avec l'adresse Lyon — la valeur du moment du traitement, pas celle du moment de la commande.

Le retard ne casse rien côté partitions (la commande reste sur sa partition) ; il change juste avec quelle version on joint. Par défaut : la version courante.

Si on a besoin de la valeur au timestamp de l'événement (lookup as-of), il faut une versioned KTable (KIP-889, Kafka 3.5+) : un store qui conserve un historique par clé.

KTable<String, Customer> customers = builder.table(  // KTable versionnée (permet le lookup as-of)
    "customers",                                     // topic source
    Materialized.as(Stores.persistentVersionedKeyValueStore(  // store versionné
        "customers-store", Duration.ofDays(7)))      // 7 jours d'historique conservé par clé
);

Le join interroge alors la version temporellement correcte — utile dès qu'on doit reconstituer l'état d'une entité tel qu'il était à un instant passé.

Sources

• Kafka Streams — Joining
• Kafka Streams — Windowing
• Confluent — Kafka Streams Windowing
• KIP-213 — Foreign-key join
• KIP-889 — Versioned state stores
• kafka-streams-1-fondamentaux — KTable, GlobalKTable, co-partitionnement
• kafka-streams-2-operations-transformations — groupement, agrégations
• kafka-streams-4-state-stores-iq-tests — state stores, Interactive Queries, tests