Kafka Streams 1/4 — Fondamentaux

flowchart LR
    subgraph "Service Spring Boot"
        KS["Kafka Streams"]
        RS["RocksDB<br/>state store local"]
        KS <--> RS
    end
    K1["Topic source"] -->|consomme| KS
    KS -->|publie| K2["Topic résultat"]
    KS <-->|changelog backup| K3["Topic changelog<br/>_internal"]

flowchart LR
    T1["Topic orders<br/>source"] --> F["Filtre<br/>amount != 0"]
    F --> M["Map<br/>enrichissement"]
    M --> B["Branch"]
    B -->|montant élevé| A["Topic order-alerts<br/>sink"]
    B -->|normal| P["Topic orders-processed<br/>sink"]

@Configuration                                   // classe de configuration Spring
@EnableKafkaStreams                              // active l'auto-configuration Kafka Streams
public class OrderTopology {

    @Bean                                        // expose la topologie comme bean Spring
    public KStream<String, OrderPlaced> orderStream(StreamsBuilder builder) {

        KStream<String, OrderPlaced> stream =
            builder.stream("orders");            // nœud source : lit le topic orders (clé = customerId)

        KStream<String, OrderPlaced> processed = stream
            .filter((customerId, order) -> order.getAmount() != 0)  // ne garde que les commandes à montant non nul
            .mapValues(order -> enrich(order));  // enrichit la valeur sans changer la clé → pas de repartition

        processed.to("orders-processed");        // nœud sink : écrit le résultat vers orders-processed

        return processed;                        // retourne le stream pour un éventuel chaînage
    }
}

KTable<String, Long> customerTotal = builder
    .stream("orders")                                  // source : flux des commandes
    .groupByKey()                                      // groupe par la clé courante (customerId)
    .aggregate(                                        // agrège les montants par client
        () -> 0L,                                      // valeur initiale pour toute nouvelle clé
        (customerId, order, currentTotal) -> currentTotal + order.getAmount(),  // accumulateur : ajoute le montant
        Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>  // matérialise le résultat dans un store
            .as("customer-total-store")                // nom du state store
            .withValueSerde(Serdes.Long())             // serde de la valeur (Long)
    );

flowchart TD
    subgraph "Instance 1 (partitions 0 et 1)"
        KT1["KTable<br/>partitions 0 et 1 seulement"]
        GKT1["GlobalKTable<br/>toutes les partitions"]
    end
    subgraph "Instance 2 (partitions 2 et 3)"
        KT2["KTable<br/>partitions 2 et 3 seulement"]
        GKT2["GlobalKTable<br/>toutes les partitions"]
    end
    T["Topic référentiel<br/>4 partitions"] --> KT1
    T --> KT2
    T --> GKT1
    T --> GKT2

spring:
  kafka:
    bootstrap-servers: localhost:9092            # adresse du cluster Kafka
    streams:
      # Identifiant unique de l'application.
      # Deux instances avec le même application-id forment un cluster Streams —
      # Kafka distribue les partitions entre elles automatiquement.
      application-id: order-processing-service

      # Serde par défaut pour les clés et les valeurs.
      # Peut être surchargé par stream via Consumed.with() ou Produced.with().
      default-key-serde: org.apache.kafka.common.serialization.Serdes$StringSerde         # clé = String
      default-value-serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde       # valeur = Avro

      properties:
        schema.registry.url: http://localhost:8081   # URL du Schema Registry (résolution des schémas Avro)

        # En cas d'erreur de désérialisation d'un message :
        # LogAndContinueExceptionHandler → log et skip le message
        # LogAndFailExceptionHandler → arrêt de l'application (défaut)
        default.deserialization.exception.handler: >
          org.apache.kafka.streams.errors.LogAndContinueExceptionHandler

        # Cache interne pour batcher les writes vers les state stores.
        # 0 = désactive le cache (utile en test pour voir les résultats immédiatement).
        statestore.cache.max.bytes: 10485760

        # Répertoire local où RocksDB stocke les state stores sur disque.
        state.dir: /tmp/kafka-streams

@Configuration                                   // classe de configuration Spring
@EnableKafkaStreams                              // active Kafka Streams
public class KafkaStreamsConfig {

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)  // bean de config attendu par Spring
    public KafkaStreamsConfiguration streamsConfig() {
        Map<String, Object> props = new HashMap<>();                                 // map des propriétés Streams
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-processing-service");  // identité de l'app dans Kafka
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");         // adresse du cluster
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());  // serde clé par défaut
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);    // serde valeur par défaut (Avro)
        props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");  // URL du Schema Registry

        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 3);    // nombre de threads de traitement par instance
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);  // fréquence des commits d'offset (ms)

        return new KafkaStreamsConfiguration(props);              // construit l'objet de config Spring
    }
}

@Bean                                            // expose le stream comme bean
public KStream<String, OrderPlaced> orderStream(StreamsBuilder builder) {

    SpecificAvroSerde<OrderPlaced> orderSerde = new SpecificAvroSerde<>();  // serde Avro pour OrderPlaced
    orderSerde.configure(                        // configure le serde
        Map.of(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,   // clé : URL du Schema Registry
            "http://localhost:8081"),            // valeur : l'URL
        false  // false = serde de valeur, true = serde de clé
    );

    return builder.stream(                       // crée le nœud source
        "orders",                                // topic lu
        Consumed.with(Serdes.String(), orderSerde)  // serdes explicites clé/valeur pour ce stream
    );
}

flowchart LR
    SM["orders<br/>source"] --> F["Filtre<br/>amount != 0"]
    CR["customer-ref<br/>GlobalKTable"] --> J
    F --> J["Join<br/>enrichissement"]
    J --> B["Branch"]
    B -->|montant élevé| AL["order-alerts<br/>sink"]
    B -->|normal| AG["Agrégation<br/>par pays"]
    AG --> SU["order-summary<br/>sink"]

@Configuration                                   // classe de configuration Spring
@EnableKafkaStreams                              // active Kafka Streams
public class OrderProcessingTopology {

    @Bean                                        // expose la topologie comme bean
    public KStream<String, OrderPlaced> buildTopology(StreamsBuilder builder) {

        GlobalKTable<String, CustomerRef> customerRef = builder.globalTable(  // référentiel client en GlobalKTable
            "customer-ref",                      // topic du référentiel
            Materialized.as("customer-ref-store")  // store local pour le référentiel
        );

        KStream<String, OrderPlaced> orderStream = builder
            .stream("orders")                    // source : flux des commandes
            .filter((customerId, order) -> order.getAmount() != 0);  // élimine les commandes à montant nul

        KStream<String, OrderEnriched> enriched = orderStream.join(  // join stream × GlobalKTable
            customerRef,                         // table de référence à joindre
            (customerId, order) -> customerId,   // clé de lookup dans la GlobalKTable (ici = la clé du stream)
            (order, customer) -> OrderEnriched.newBuilder()  // combine commande + client
                .setOrderId(order.getOrderId())
                .setCustomerName(customer.getName())
                .setTier(customer.getTier())
                .setAmount(order.getAmount())
                .setAlertThreshold(customer.getAlertThreshold())
                .setCountry(order.getCountry())
                .build()
        );

        Map<String, KStream<String, OrderEnriched>> branches = enriched
            .split(Named.as("branch-"))          // découpe le stream en sous-flux
            .branch(                             // 1ère règle (évaluée en premier)
                (customerId, e) -> e.getAmount() > e.getAlertThreshold(),  // montant au-dessus du seuil client
                Branched.as("alert")             // → sous-flux "alert"
            )
            .defaultBranch(Branched.as("normal"));  // tout le reste → sous-flux "normal"

        branches.get("branch-alert")             // récupère le sous-flux des alertes
            .mapValues(e -> OrderAlert.newBuilder()  // transforme en OrderAlert
                .setOrderId(e.getOrderId())
                .setCustomerName(e.getCustomerName())
                .setAmount(e.getAmount())
                .build())
            .to("order-alerts");                 // publie vers le topic order-alerts

        branches.get("branch-normal")            // récupère le sous-flux normal
            .groupBy((customerId, e) -> e.getCountry())  // regroupe par pays (rekey → repartition interne)
            .aggregate(                          // agrège le montant total par pays
                () -> 0L,                        // valeur initiale
                (country, event, total) -> total + event.getAmount(),  // accumulateur : somme des montants
                Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>  // matérialise dans un store
                    .as("country-summary-store")  // nom du store
                    .withValueSerde(Serdes.Long())  // serde de la valeur
            )
            .toStream()                          // KTable → KStream pour publier les mises à jour
            .mapValues(total -> CountrySummary.newBuilder().setTotal(total).build())  // emballe le total dans CountrySummary
            .to("order-summary");                // publie vers order-summary

        return orderStream;                      // retourne le stream principal
    }
}

flowchart LR
    subgraph "Instance 1 (2 threads)"
        TH1["Thread 1<br/>Tâche 0 partition 0<br/>Tâche 1 partition 1"]
        TH2["Thread 2<br/>Tâche 2 partition 2"]
    end
    subgraph "Instance 2 (2 threads)"
        TH3["Thread 3<br/>Tâche 3 partition 3<br/>Tâche 4 partition 4"]
        TH4["Thread 4<br/>Tâche 5 partition 5"]
    end
    K["Topic source<br/>6 partitions"] --> TH1
    K --> TH2
    K --> TH3
    K --> TH4