Kafka Streams 4/4 — State Stores, Interactive Queries & Tests

flowchart LR
    OP["Agrégation / Join"] -->|write| SS["State store<br/>RocksDB local"]
    SS -->|backup async| CL["Topic changelog<br/>compacté"]
    CL -.->|restore au démarrage| SS

.count(Materialized.as("orders-count-store"));   // store persistant par défaut (RocksDB sur disque)

.count(Materialized.as(                          // store in-memory (dans la heap JVM)
    Stores.inMemoryKeyValueStore("orders-count-store")));

props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);  // 1 copie à chaud sur une autre instance (failover rapide)

ReadOnlyKeyValueStore<String, Long> store = streams.store(  // ouvre le state store en lecture seule
    StoreQueryParameters.fromNameAndType(                   // paramètres de la requête
        "customer-summary-store",                           // nom du store à interroger
        QueryableStoreTypes.keyValueStore()                 // type : store clé-valeur
    )
);
Long total = store.get("C-42");   // lecture locale de la clé C-42

KeyQueryMetadata metadata = streams.queryMetadataForKey(  // localise l'instance qui détient la clé
    "customer-summary-store", "C-42", Serdes.String().serializer());  // store, clé, serializer de clé

HostInfo host = metadata.activeHost();   // host de l'instance active pour cette clé
if (host.equals(thisInstance)) {         // la clé est-elle locale à cette instance ?
    return store.get("C-42");            // oui → lecture locale
} else {
    return remoteCall(host, "C-42");     // non → appel RPC vers l'instance distante
}

sequenceDiagram
    participant C as Client REST
    participant A as Instance A
    participant B as Instance B
    C->>A: GET /total/C-42
    A->>A: queryMetadataForKey("customer-summary-store", "C-42")
    Note over A: la clé C-42 vit sur l'instance B
    A->>B: appel RPC /total/C-42
    B->>B: store.get("C-42")
    B-->>A: 1240
    A-->>C: 1240

props.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "instance-a.svc:8080");  // adresse exposée dans les métadonnées (host:port)

props.put(  // configure le handler d'erreur de désérialisation
    StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,  // clé de config
    LogAndContinueExceptionHandler.class   // log + skip le message (vs LogAndFailExceptionHandler = arrêt, le défaut)
);

props.put(  // configure le handler d'erreur de production
    StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,  // clé de config
    MyProductionExceptionHandler.class   // retourne CONTINUE (skip) ou FAIL selon l'exception
);

streams.setUncaughtExceptionHandler(ex -> {   // handler des exceptions non capturées dans un processor
    // REPLACE_THREAD       → remplace le thread, l'instance survit
    // SHUTDOWN_CLIENT      → arrête cette instance (défaut depuis 2.8)
    // SHUTDOWN_APPLICATION  → arrête toutes les instances du même application-id
    return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD;  // ici : survivre à l'erreur
});

Topology topology = buildTopology();   // la topologie à tester
Properties props = new Properties();   // config minimale pour le driver
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test");          // application-id factice
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092"); // aucun broker réel requis
// ⚠️ Désactiver le cache pour voir les résultats immédiatement
props.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);   // cache à 0 → résultats visibles immédiatement

try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {  // driver de test (try-with-resources)

    TestInputTopic<String, OrderPlaced> orders = driver.createInputTopic(  // topic d'entrée simulé
        "orders", new StringSerializer(), orderSerde.serializer());        // nom + serializers clé/valeur

    TestOutputTopic<String, CustomerSummary> out = driver.createOutputTopic(  // topic de sortie simulé
        "customer-summary-output", new StringDeserializer(), summarySerde.deserializer());  // nom + deserializers

    orders.pipeInput("C-A", order(10));   // injecte une commande de 10 pour C-A
    orders.pipeInput("C-A", order(25));   // injecte une commande de 25 pour C-A

    assertThat(out.readKeyValue().value.getTotalAmount()).isEqualTo(35);  // vérifie l'agrégat en sortie (10 + 25)

    KeyValueStore<String, CustomerSummary> store =                        // accès direct au state store
        driver.getKeyValueStore("customer-summary-store");
    assertThat(store.get("C-A").getOrderCount()).isEqualTo(2);            // vérifie le compteur dans le store (2 commandes)
}